diff --git a/cli/package-lock.json b/cli/package-lock.json
index 6a122921..8b469f6a 100644
--- a/cli/package-lock.json
+++ b/cli/package-lock.json
@@ -1,12 +1,12 @@
 {
   "name": "@yc-software/qm",
-  "version": "0.1.5",
+  "version": "0.1.6",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@yc-software/qm",
-      "version": "0.1.5",
+      "version": "0.1.6",
       "license": "MIT",
       "bin": {
         "qm": "dist/bin/qm.js"
diff --git a/cli/package.json b/cli/package.json
index d7349677..4fbc357d 100644
--- a/cli/package.json
+++ b/cli/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@yc-software/qm",
-  "version": "0.1.5",
+  "version": "0.1.6",
   "license": "MIT",
   "description": "Control-plane CLI for portable QM deployments on Docker, Fly, and AWS.",
   "type": "module",
diff --git a/cli/src/backends/aws.ts b/cli/src/backends/aws.ts
index 9c5db74b..13ede174 100644
--- a/cli/src/backends/aws.ts
+++ b/cli/src/backends/aws.ts
@@ -49,7 +49,12 @@ import {
   streamLabeled,
 } from "../util.ts";
 import { doctorCommon } from "./doctor.ts";
-import { awsObjectStoreBucket, declaredVariables, terraformVarsDrift } from "../terraform.ts";
+import {
+  assertTerraformScaffoldSupportsConfig,
+  awsObjectStoreBucket,
+  declaredVariables,
+  terraformVarsDrift,
+} from "../terraform.ts";
 import {
   currentDeploymentLayerState,
   deploymentLayerBody,
@@ -94,6 +99,7 @@ function awsTopology(
   configDir: string,
 ): { aws: AwsConfig; workloads: string[]; plugins: ResolvedPlugin[] } {
   const aws = requireAws(config);
+  assertTerraformScaffoldSupportsConfig(config, configDir);
   const discovered = discoverPlugins(configDir, config);
   if (discovered.errors.length) throw new CliError(discovered.errors.join("\n"));
   const workloads = [...runnableServices(config.services), ...discovered.plugins.map((plugin) => plugin.name)];
@@ -286,7 +292,9 @@ function workloadEnvironment(config: QmConfig, workload: string): Record<string,
   const plugin = config.plugins.find((entry) => entry.name === workload);
   return Object.fromEntries(
     Object.entries({
-      CORE_API_URL: `http://core.${requireAws(config).networking.cloudMapNamespace}:8080`,
+      ...(plugin?.coreAccess === false
+        ? {}
+        : { CORE_API_URL: `http://core.${requireAws(config).networking.cloudMapNamespace}:8080` }),
       ...orgEnv(workload, config.orgId, config.publicUrl, config.services.includes("portal")),
       ...plugin?.env,
       PORT: "8080",
@@ -298,7 +306,12 @@ function workloadSecrets(config: QmConfig, workload: string, available?: Record<
   const secrets = secretsForService(config, workload).filter(
     (secret) => secret.required || Boolean(available?.[secret.name]),
   );
-  if (!isServiceName(workload) && !secrets.some((secret) => secret.name === "CORE_SIGNING_SECRET")) {
+  const plugin = config.plugins.find((entry) => entry.name === workload);
+  if (
+    !isServiceName(workload) &&
+    plugin?.coreAccess !== false &&
+    !secrets.some((secret) => secret.name === "CORE_SIGNING_SECRET")
+  ) {
     const signing = computedSecrets(config).find((secret) => secret.name === "CORE_SIGNING_SECRET");
     if (signing) return [...secrets, signing];
   }
@@ -323,9 +336,10 @@ export function renderTaskDefinition(
   if (service === "core" && usesFlySandboxes(config)) resolveAwsSandboxPin(config, () => undefined);
   const internalPort = isServiceName(service) ? serviceDef(service).docker.internalPort : 8080;
   const executionRoleArn = spec.executionRoleArn ?? `arn:aws:iam::${aws.accountId}:role/${aws.cluster}-task-execution`;
-  const taskRoleArn =
-    spec.taskRoleArn ??
-    `arn:aws:iam::${aws.accountId}:role/${aws.cluster}-${service === "core" ? "core-task" : "task"}`;
+  const managedTaskRole = spec.assumeRoleArns?.length
+    ? `${aws.cluster}-${service}-task`
+    : `${aws.cluster}-${service === "core" ? "core-task" : "task"}`;
+  const taskRoleArn = spec.taskRoleArn ?? `arn:aws:iam::${aws.accountId}:role/${managedTaskRole}`;
   const secrets = workloadSecrets(config, service, secretArns)
     .flatMap((secret) =>
       containerSecretNames(service, secret).map((name) => ({
@@ -764,6 +778,7 @@ interface DeploymentManifest {
   sandboxImage?: string;
   dbSnapshot?: string;
   tasks: Record<string, string>;
+  counts?: Record<string, number>;
   imageProvenance?: Record<string, DeploymentImageProvenance>;
   layer?: { key: string; sha256: string };
 }
@@ -1098,9 +1113,11 @@ function recordDeploymentManifest(
     dbSnapshot?: string;
     layer?: DeploymentManifest["layer"];
     imageProvenance?: DeploymentManifest["imageProvenance"];
+    counts?: DeploymentManifest["counts"];
   },
 ): DeploymentManifest {
   const current = currentDeploymentManifest(aws);
+  const counts = release.counts ?? current?.counts;
   const manifest: DeploymentManifest = {
     id: release.id ?? randomUUID(),
     ...(current ? { previous: current.id } : {}),
@@ -1109,6 +1126,7 @@ function recordDeploymentManifest(
     ...(release.sandboxImage ? { sandboxImage: release.sandboxImage } : {}),
     ...(release.dbSnapshot ? { dbSnapshot: release.dbSnapshot } : {}),
     tasks,
+    ...(counts ? { counts } : {}),
     ...(release.imageProvenance ? { imageProvenance: release.imageProvenance } : {}),
     ...(release.layer ? { layer: release.layer } : {}),
   };
@@ -1142,6 +1160,7 @@ function recordCarriedSandboxPin(aws: AwsConfig, image: string): DeploymentManif
   if (current.sandboxImage === image) return current;
   return recordDeploymentManifest(aws, current.tasks, {
     sandboxImage: image,
+    ...(current.counts ? { counts: current.counts } : {}),
     ...(current.imageLabel ? { imageLabel: current.imageLabel } : {}),
     ...(current.layer ? { layer: current.layer } : {}),
     ...(current.imageProvenance ? { imageProvenance: current.imageProvenance } : {}),
@@ -1256,9 +1275,8 @@ function trustedDeploymentBaseline(
   action = "deploy --only",
 ): DeploymentManifest {
   const aws = requireAws(config);
-  const allWorkloads = Object.keys(aws.services);
   const current = currentDeploymentManifest(aws);
-  if (!current || allWorkloads.some((workload) => !current.tasks[workload])) {
+  if (!current || workloads.some((workload) => !current.tasks[workload])) {
     throw new CliError(
       "the first AWS deployment must include every workload; omit --only until a complete deployment manifest exists",
     );
@@ -1292,6 +1310,27 @@ function trustedDeploymentBaseline(
   return current;
 }
 
+function adoptMissingWorkloads(
+  manifest: DeploymentManifest,
+  snapshot: ReturnType<typeof serviceSnapshot>,
+  workloads: string[],
+): boolean {
+  const missing = workloads.filter((workload) => !manifest.tasks[workload]);
+  const active = missing.filter((workload) => snapshot.counts[workload] !== 0);
+  if (active.length) {
+    throw new CliError(
+      `selected workload ${active.join(", ")} is missing from the current deployment manifest and must be scaled to zero before adoption`,
+    );
+  }
+  if (!missing.length) return false;
+  manifest.counts ??= {};
+  for (const workload of missing) {
+    manifest.tasks[workload] = snapshot.tasks[workload]!;
+    manifest.counts[workload] = snapshot.counts[workload]!;
+  }
+  return true;
+}
+
 async function applyServiceTargets(
   config: QmConfig,
   targets: Record<string, string>,
@@ -1587,11 +1626,12 @@ export async function awsUp(config: QmConfig, _configDir: string, opts: AwsUpOpt
     const before = serviceSnapshot(config, allServices);
     const selected = new Set(services);
     if (allServices.some((service) => !selected.has(service))) {
-      trustedDeploymentBaseline(
+      const baseline = trustedDeploymentBaseline(
         config,
         before,
         allServices.filter((name) => !selected.has(name)),
       );
+      adoptMissingWorkloads(baseline, before, services);
     }
     const images: Record<string, string> = {};
     for (const service of services) {
@@ -1644,6 +1684,15 @@ export async function awsUp(config: QmConfig, _configDir: string, opts: AwsUpOpt
     assertAwsDeployImage(config);
     before = serviceSnapshot(config, allServices);
     current = currentDeploymentManifest(aws);
+    const selected = new Set(services);
+    if (allServices.some((service) => !selected.has(service))) {
+      current = trustedDeploymentBaseline(
+        config,
+        before,
+        allServices.filter((name) => !selected.has(name)),
+      );
+      if (adoptMissingWorkloads(current, before, services)) manifestTransaction(aws, current, current.id);
+    }
     if (usesFlySandboxes(config)) {
       if (services.includes("core")) {
         const pin = resolveAwsSandboxPin(config, () => current);
@@ -1701,14 +1750,6 @@ export async function awsUp(config: QmConfig, _configDir: string, opts: AwsUpOpt
         }
       }
     }
-    const selected = new Set(services);
-    if (allServices.some((service) => !selected.has(service))) {
-      trustedDeploymentBaseline(
-        config,
-        before,
-        allServices.filter((name) => !selected.has(name)),
-      );
-    }
     dockerLogin(aws);
     const images: Record<string, string> = {};
     const selectedImageProvenance: Record<string, DeploymentImageProvenance> = {};
@@ -1766,14 +1807,19 @@ export async function awsUp(config: QmConfig, _configDir: string, opts: AwsUpOpt
       );
     }
     const releaseTasks = { ...before.tasks, ...targets };
+    const releaseCounts = Object.fromEntries(
+      allServices.map((service) => [service, workloadDesiredCount(config, service)]),
+    );
     const releaseImageProvenance = { ...current?.imageProvenance, ...selectedImageProvenance };
     const sameTasks = current && allServices.every((service) => current!.tasks[service] === releaseTasks[service]);
+    const sameCounts = current && canonicalJson(current.counts ?? {}) === canonicalJson(releaseCounts);
     const sameImageProvenance =
       current && canonicalJson(current.imageProvenance ?? {}) === canonicalJson(releaseImageProvenance);
     releaseSucceeded = true;
     if (
       !current ||
       !sameTasks ||
+      !sameCounts ||
       !sameImageProvenance ||
       current.imageLabel !== label ||
       layerChanged ||
@@ -1781,6 +1827,7 @@ export async function awsUp(config: QmConfig, _configDir: string, opts: AwsUpOpt
     ) {
       recorded = recordDeploymentManifest(aws, releaseTasks, {
         id: releaseId,
+        counts: releaseCounts,
         ...(sandboxPinImage ? { sandboxImage: sandboxPinImage } : {}),
         imageLabel: label,
         ...(dbSnapshot ? { dbSnapshot } : {}),
@@ -1972,16 +2019,21 @@ export async function awsRollback(
       layerNeedsSync = currentManifest?.layer?.sha256 !== targetManifest.layer!.sha256;
     }
     const targets = Object.fromEntries(services.map((service) => [service, targetManifest!.tasks[service]!]));
+    const targetCounts = Object.fromEntries(
+      services.map((service) => [service, targetManifest.counts?.[service] ?? before!.counts[service]!]),
+    );
     const changedTargets = Object.fromEntries(
       services
-        .filter((service) => before!.tasks[service] !== targets[service])
+        .filter(
+          (service) => before!.tasks[service] !== targets[service] || before!.counts[service] !== targetCounts[service],
+        )
         .map((service) => [service, targets[service]!]),
     );
     if (Object.keys(changedTargets).length) {
       await applyServiceTargets(
         config,
         changedTargets,
-        Object.fromEntries(Object.keys(changedTargets).map((service) => [service, before!.counts[service]!])),
+        Object.fromEntries(Object.keys(changedTargets).map((service) => [service, targetCounts[service]!])),
       );
       applied = true;
     }
@@ -1990,12 +2042,12 @@ export async function awsRollback(
       Object.fromEntries(
         services.map((service) => [
           service,
-          { taskDefinition: targets[service]!, desiredCount: before!.counts[service]! },
+          { taskDefinition: targets[service]!, desiredCount: targetCounts[service]! },
         ]),
       ),
     );
     if (targetLayerBody && layerOpts && layerNeedsSync) {
-      if (before.counts.core === 0) {
+      if (targetCounts.core === 0) {
         note(
           "deployment layer sync deferred: core is scaled to zero, so `check --live` will flag the layer until the next `qm up` applies it",
         );
@@ -2222,6 +2274,7 @@ export async function awsSecretsPush(config: QmConfig, configDir: string, envFil
         rotated = true;
         if (Object.keys(changed).length) {
           recordDeploymentManifest(aws, targets, {
+            ...(baseline.counts ? { counts: baseline.counts } : {}),
             ...(baseline.sandboxImage ? { sandboxImage: baseline.sandboxImage } : {}),
             imageLabel: baseline.imageLabel ?? aws.imageLabel,
             ...(baseline.layer ? { layer: baseline.layer } : {}),
@@ -3316,6 +3369,7 @@ export async function awsPinSandbox(
         { ...baseline.tasks, core: taskDefinition },
         {
           sandboxImage: image,
+          ...(baseline.counts ? { counts: baseline.counts } : {}),
           imageLabel: baseline.imageLabel ?? aws.imageLabel,
           ...(layer ? { layer } : {}),
           ...(baseline.imageProvenance ? { imageProvenance: baseline.imageProvenance } : {}),
diff --git a/cli/src/backends/docker.ts b/cli/src/backends/docker.ts
index 232f078a..542329c2 100644
--- a/cli/src/backends/docker.ts
+++ b/cli/src/backends/docker.ts
@@ -342,7 +342,8 @@ function serviceEnv(ctx: DockerCtx, service: ServiceName): Record<string, string
 
 function secretEnvKeys(ctx: DockerCtx, service: string): Set<string> {
   const keys = new Set(Object.keys(secretValues(ctx, service)));
-  if (ctx.signingSecret) keys.add("CORE_SIGNING_SECRET");
+  const plugin = ctx.config.plugins.find((entry) => entry.name === service);
+  if (ctx.signingSecret && plugin?.coreAccess !== false) keys.add("CORE_SIGNING_SECRET");
   if (service === "core") {
     keys.add("DATABASE_URL");
     for (const key of ctx.sandboxSecretKeys) keys.add(key);
@@ -607,14 +608,14 @@ export async function dockerUp(
       "no",
     ];
     const wiring = {
-      CORE_API_URL: "http://core:8080",
+      ...(p.coreAccess === false ? {} : { CORE_API_URL: "http://core:8080" }),
       ...orgEnv(p.name, config.orgId, config.publicUrl, config.services.includes("portal")),
       PORT: "8080",
     };
     const env = {
       ...wiring,
       ...p.env,
-      ...(ctx.signingSecret ? { CORE_SIGNING_SECRET: ctx.signingSecret } : {}),
+      ...(ctx.signingSecret && p.coreAccess !== false ? { CORE_SIGNING_SECRET: ctx.signingSecret } : {}),
       ...secretValues(ctx, p.name),
     };
     const cleanup = pushEnvArgs(args, env, secretEnvKeys(ctx, p.name));
diff --git a/cli/src/backends/fly.ts b/cli/src/backends/fly.ts
index 0940eb03..390faeed 100644
--- a/cli/src/backends/fly.ts
+++ b/cli/src/backends/fly.ts
@@ -814,7 +814,7 @@ function pluginTomlContent(
   plugin: ResolvedPlugin,
 ): string {
   const env: Record<string, string> = {
-    CORE_API_URL: `http://${appPrefix}-core.internal:8080`,
+    ...(plugin.coreAccess === false ? {} : { CORE_API_URL: `http://${appPrefix}-core.internal:8080` }),
     ...orgEnv(plugin.name, orgId, publicUrl, hasPortal),
     PORT: "8080",
     ...plugin.env,
@@ -961,6 +961,21 @@ function unsetDisabledFlyPublisherToken(config: QmConfig, appPrefix: string): vo
   note(`removed the disabled Fly app publisher token from ${app}`);
 }
 
+function unsetCorelessPluginCoreSecrets(plugins: ResolvedPlugin[], appPrefix: string): string[] {
+  const changed: string[] = [];
+  for (const plugin of plugins) {
+    if (plugin.coreAccess !== false) continue;
+    const app = `${appPrefix}-${plugin.name}`;
+    const existing = secretNames(app);
+    const names = ["CORE_API_URL", "CORE_SIGNING_SECRET"].filter((name) => existing?.has(name));
+    if (!names.length) continue;
+    fly(["secrets", "unset", "--stage", "-a", app, ...names]);
+    changed.push(app);
+    note(`removed core access from ${app}`);
+  }
+  return changed;
+}
+
 export async function flyUp(config: QmConfig, configDir: string, opts: FlyUpOpts = {}): Promise<void> {
   if (opts.imageLabel && opts.imageFrom) {
     throw new CliError("--image-label and --image-from select different image sources and cannot be combined");
@@ -1080,11 +1095,12 @@ export async function flyUp(config: QmConfig, configDir: string, opts: FlyUpOpts
       ];
       if (gateSecrets(app, app, writeDerived(ctx, s), required, s)) missingAny = true;
     }
+    const corePluginNames = plugins.filter((plugin) => plugin.coreAccess !== false).map((plugin) => plugin.name);
     for (const p of plugins) {
       const app = pluginApp(ctx, p.name);
-      const required = secretsForService(config, p.name, [p.name])
+      const required = secretsForService(config, p.name, corePluginNames)
         .filter((secret) => secret.required)
-        .flatMap((secret) => runtimeSecretNames(p.name, secret, [p.name]));
+        .flatMap((secret) => runtimeSecretNames(p.name, secret, corePluginNames));
       if (gateSecrets(app, `${app} (plugin: ${p.kind})`, writePluginDerived(ctx, p), required, p.name)) {
         missingAny = true;
       }
@@ -1101,6 +1117,7 @@ export async function flyUp(config: QmConfig, configDir: string, opts: FlyUpOpts
       unsetDisabledSecurityScreenToken(config, ctx.appPrefix);
       unsetDisabledFlyPublisherToken(config, ctx.appPrefix);
     }
+    unsetCorelessPluginCoreSecrets(plugins, ctx.appPrefix);
 
     for (const phase of flyDeployPhases(services)) await deployPhase(ctx, phase, imageSource, timing);
     await deployPlugins(ctx, plugins, imageSource, timing);
@@ -1379,6 +1396,9 @@ export async function flyDoctor(config: QmConfig, configDir: string, envFile?: s
   requireFlyAuth();
   const prefix = appPrefixOf(config);
   const pluginNames = discovered.plugins.map((plugin) => plugin.name);
+  const corePluginNames = discovered.plugins
+    .filter((plugin) => plugin.coreAccess !== false)
+    .map((plugin) => plugin.name);
   const failures: string[] = [];
   for (const workload of [...runnableServices(config.services), ...pluginNames]) {
     const app = `${prefix}-${workload}`;
@@ -1387,18 +1407,23 @@ export async function flyDoctor(config: QmConfig, configDir: string, envFile?: s
       step(`${app}: not created yet — secret checks run after the first \`qm up\``);
       continue;
     }
-    const declared = secretsForService(config, workload, pluginNames);
+    if (discovered.plugins.some((plugin) => plugin.name === workload && plugin.coreAccess === false)) {
+      for (const name of ["CORE_API_URL", "CORE_SIGNING_SECRET"]) {
+        if (existing.has(name)) failures.push(`${app}: unexpected ${name} on a coreless plugin`);
+      }
+    }
+    const declared = secretsForService(config, workload, corePluginNames);
     const required = new Set([
       ...declared
         .filter((secret) => secret.required)
-        .flatMap((secret) => runtimeSecretNames(workload, secret, pluginNames)),
+        .flatMap((secret) => runtimeSecretNames(workload, secret, corePluginNames)),
       ...flyProviderSecrets(config, workload),
     ]);
     const missing = [...required].filter((name) => !existing.has(name));
     if (missing.length) failures.push(`${app}: missing ${missing.join(", ")}`);
     else step(`${app} required secrets: ok`);
     for (const secret of declared.filter((item) => !item.required)) {
-      for (const name of runtimeSecretNames(workload, secret, pluginNames)) {
+      for (const name of runtimeSecretNames(workload, secret, corePluginNames)) {
         if (existing.has(name)) step(`${app} optional secret ${name}: configured`);
         else warn(`${app} optional secret ${name}: not configured`);
       }
@@ -1627,7 +1652,9 @@ export async function flySecretsPush(config: QmConfig, configDir: string, envFil
   const path = resolve(envFile ?? join(configDir, ".env"));
   const values = existsSync(path) ? readEnvFile(path) : new Map<string, string>();
   const prefix = appPrefixOf(config);
-  const pluginNames = discoverPlugins(configDir, config).plugins.map((plugin) => plugin.name);
+  const plugins = discoverPlugins(configDir, config).plugins;
+  const pluginNames = plugins.map((plugin) => plugin.name);
+  const corePluginNames = plugins.filter((plugin) => plugin.coreAccess !== false).map((plugin) => plugin.name);
   const operatorSecrets = computedSecrets(config).filter((item) => item.managedBy === "operator");
   for (const secret of operatorSecrets) {
     const supplied = deploymentSecretValue(secret.name, values.get(secret.name));
@@ -1642,7 +1669,7 @@ export async function flySecretsPush(config: QmConfig, configDir: string, envFil
   for (const plugin of pluginNames) ensureApp(`${prefix}-${plugin}`, ctx.flyOrg, ctx.orgId, ctx.appPrefix);
   unsetDisabledSecurityScreenToken(config, prefix);
   unsetDisabledFlyPublisherToken(config, prefix);
-  const stagedApps = new Set<string>();
+  const stagedApps = new Set(unsetCorelessPluginCoreSecrets(plugins, prefix));
   for (const secret of operatorSecrets) {
     const supplied = deploymentSecretValue(secret.name, values.get(secret.name));
     if (!secret.required && !supplied) {
@@ -1654,7 +1681,7 @@ export async function flySecretsPush(config: QmConfig, configDir: string, envFil
       throw new CliError(`required secret ${secret.name} is missing, a placeholder, or too short`);
     }
     const destinations = new Map<string, Set<string>>();
-    for (const [workload, names] of secretDestinations(secret, pluginNames)) {
+    for (const [workload, names] of secretDestinations(secret, corePluginNames)) {
       destinations.set(`${prefix}-${workload}`, names);
     }
     for (const [app, names] of destinations) {
diff --git a/cli/src/config.ts b/cli/src/config.ts
index 2277e65c..31b0cf5f 100644
--- a/cli/src/config.ts
+++ b/cli/src/config.ts
@@ -36,6 +36,7 @@ export interface PluginEntry {
   image?: string;
   env?: Record<string, string>;
   secrets?: PluginSecret[];
+  coreAccess?: boolean;
 }
 
 export interface SandboxConfig {
@@ -63,6 +64,7 @@ export interface AwsServiceConfig {
   architecture?: "arm64" | "amd64";
   taskRoleArn?: string;
   executionRoleArn?: string;
+  assumeRoleArns?: string[];
   buildArgs?: Record<string, string>;
   dockerfile?: string;
   targetGroup?: string;
@@ -917,6 +919,20 @@ function validatePlugins(raw: unknown, path: string): PluginEntry[] {
     }
     if (e["env"] !== undefined) entry.env = validateStringMap(e["env"], path, `plugins[${i}].env`);
     if (e["secrets"] !== undefined) entry.secrets = validatePluginSecrets(e["secrets"], path, i);
+    if (e["coreAccess"] !== undefined) {
+      if (typeof e["coreAccess"] !== "boolean") {
+        throw new CliError(`${path}: plugins[${i}].coreAccess must be a boolean`);
+      }
+      entry.coreAccess = e["coreAccess"];
+    }
+    if (entry.coreAccess === false) {
+      const forbidden = ["CORE_API_URL", "CORE_SIGNING_SECRET"].filter(
+        (name) => entry.env?.[name] !== undefined || entry.secrets?.some((secret) => secret.name === name),
+      );
+      if (forbidden.length) {
+        throw new CliError(`${path}: plugins[${i}] cannot declare ${forbidden.join(", ")} when coreAccess is false`);
+      }
+    }
     return entry;
   });
 }
@@ -1146,6 +1162,25 @@ function validateAws(
     for (const role of ["taskRoleArn", "executionRoleArn"] as const) {
       if (value[role] !== undefined) service[role] = roleArn(value[role], `services.${name}.${role}`);
     }
+    if (value["assumeRoleArns"] !== undefined) {
+      const arns = validateStringArray(value["assumeRoleArns"], path, `aws.services.${name}.assumeRoleArns`);
+      if (arns.length === 0) {
+        throw new CliError(`${path}: "aws.services.${name}.assumeRoleArns" must contain at least one IAM role ARN`);
+      }
+      for (const arn of arns) {
+        if (!/^arn:aws:iam::[0-9]{12}:role\/[A-Za-z0-9_+=,.@/-]{1,512}$/.test(arn)) {
+          throw new CliError(
+            `${path}: "aws.services.${name}.assumeRoleArns" must contain commercial AWS IAM role ARNs`,
+          );
+        }
+      }
+      service.assumeRoleArns = [...new Set(arns)];
+      if (!service.taskRoleArn && `${cluster}-${name}-task`.length > 64) {
+        throw new CliError(
+          `${path}: "aws.services.${name}.assumeRoleArns" requires an explicit taskRoleArn because the derived IAM role name exceeds 64 characters`,
+        );
+      }
+    }
     if (value["architecture"] !== undefined) {
       if (value["architecture"] !== "arm64" && value["architecture"] !== "amd64") {
         throw new CliError(`${path}: "aws.services.${name}.architecture" must be "arm64" or "amd64"`);
@@ -1206,6 +1241,27 @@ function validateAws(
     }
     services[name] = service;
   }
+  const effectiveTaskRoleArn = (name: string, service: AwsServiceConfig): string => {
+    if (service.taskRoleArn) return service.taskRoleArn;
+    if (name === "core") return `arn:aws:iam::${accountId}:role/${cluster}-core-task`;
+    const roleName = service.assumeRoleArns ? `${cluster}-${name}-task` : `${cluster}-task`;
+    return `arn:aws:iam::${accountId}:role/${roleName}`;
+  };
+  const taskRoleOwners = new Map<string, string[]>();
+  for (const [name, service] of Object.entries(services)) {
+    const role = effectiveTaskRoleArn(name, service);
+    taskRoleOwners.set(role, [...(taskRoleOwners.get(role) ?? []), name]);
+  }
+  for (const [name, service] of Object.entries(services)) {
+    if (!service.assumeRoleArns) continue;
+    const role = effectiveTaskRoleArn(name, service);
+    const owners = taskRoleOwners.get(role)!;
+    if (owners.length > 1) {
+      throw new CliError(
+        `${path}: "aws.services.${name}.taskRoleArn" must be unique because assumeRoleArns grants workload-scoped permissions; ${role} is also used by ${owners.filter((owner) => owner !== name).join(", ")}`,
+      );
+    }
+  }
   for (const name of enabledServices) {
     if (!services[name]) throw new CliError(`${path}: "aws.services.${name}" is required because ${name} is enabled`);
   }
diff --git a/cli/src/plugins.ts b/cli/src/plugins.ts
index 4346f758..fc987fbd 100644
--- a/cli/src/plugins.ts
+++ b/cli/src/plugins.ts
@@ -11,6 +11,7 @@ export interface ResolvedPlugin {
   dockerfile?: string;
   env: Record<string, string>;
   secrets?: PluginSecret[];
+  coreAccess?: boolean;
 }
 
 const subdirs = (dir: string): string[] => {
@@ -34,6 +35,7 @@ export function discoverPlugins(configDir: string, config: QmConfig): { plugins:
     const image = entry?.image;
     const env = entry?.env ?? {};
     const secrets = entry?.secrets ?? [];
+    const coreAccess = entry?.coreAccess;
     const hasDockerfile = sourceDirs.has(name);
     const dir = join(pluginsRoot, name);
 
@@ -47,7 +49,14 @@ export function discoverPlugins(configDir: string, config: QmConfig): { plugins:
       continue;
     }
     if (image) {
-      plugins.push({ name, kind: "image", image, env, secrets });
+      plugins.push({
+        name,
+        kind: "image",
+        image,
+        env,
+        secrets,
+        ...(coreAccess !== undefined ? { coreAccess } : {}),
+      });
       continue;
     }
     if (hasDockerfile) {
@@ -58,6 +67,7 @@ export function discoverPlugins(configDir: string, config: QmConfig): { plugins:
         dockerfile: join(dir, "Dockerfile"),
         env,
         secrets,
+        ...(coreAccess !== undefined ? { coreAccess } : {}),
       });
       continue;
     }
diff --git a/cli/src/secrets.ts b/cli/src/secrets.ts
index ed061840..e30a7882 100644
--- a/cli/src/secrets.ts
+++ b/cli/src/secrets.ts
@@ -436,7 +436,8 @@ export function computedSecrets(config: QmConfig): ComputedSecret[] {
   }
   for (const plugin of config.plugins) {
     const signing = byName.get("CORE_SIGNING_SECRET");
-    if (signing && !signing.services.includes(plugin.name)) signing.services.push(plugin.name);
+    if (plugin.coreAccess !== false && signing && !signing.services.includes(plugin.name))
+      signing.services.push(plugin.name);
     for (const spec of plugin.secrets ?? []) {
       const required = spec.required !== false;
       const current = byName.get(spec.name);
diff --git a/cli/src/terraform.ts b/cli/src/terraform.ts
index 67d35f8e..17eb8efe 100644
--- a/cli/src/terraform.ts
+++ b/cli/src/terraform.ts
@@ -72,9 +72,79 @@ function hclString(source: string, name: string): string | undefined {
   }
 }
 
+function hclJson(source: string, name: string): unknown {
+  const assignment = hclAssignment(source, name);
+  const raw = assignment?.slice(assignment.indexOf("=") + 1).trim();
+  try {
+    return raw === undefined ? undefined : JSON.parse(raw);
+  } catch {
+    return undefined;
+  }
+}
+
+function assertSafeAssumeRoleTransition(config: QmConfig, existing: string): void {
+  if (!config.aws) return;
+  const previous = hclJson(existing, "services");
+  if (typeof previous !== "object" || previous === null || Array.isArray(previous)) return;
+  for (const [name, prior] of Object.entries(previous)) {
+    if (typeof prior !== "object" || prior === null || Array.isArray(prior)) continue;
+    const next = config.aws.services[name];
+    if (!next) continue;
+    const managedArn = `arn:aws:iam::${config.aws.accountId}:role/${config.aws.cluster}-${name === "core" ? "core" : name}-task`;
+    const priorRole = (prior as Record<string, unknown>)["task_role_arn"];
+    const priorManaged = (prior as Record<string, unknown>)["manage_task_role"] === true;
+    const currentRole = typeof priorRole === "string" ? priorRole : managedArn;
+    const nextRole =
+      next.taskRoleArn ??
+      (name === "core" || next.assumeRoleArns?.length
+        ? managedArn
+        : `arn:aws:iam::${config.aws.accountId}:role/${config.aws.cluster}-task`);
+    if (
+      name !== "core" &&
+      !priorManaged &&
+      priorRole === managedArn &&
+      !next.taskRoleArn &&
+      Boolean(next.assumeRoleArns?.length)
+    ) {
+      throw new CliError(
+        `aws.services.${name}.taskRoleArn cannot be removed because ${managedArn} is externally managed; keep the explicit ARN or import the role into Terraform first`,
+      );
+    }
+    if (name !== "core" && priorManaged && nextRole !== managedArn) {
+      throw new CliError(
+        `aws.services.${name}.taskRoleArn cannot replace the Terraform-managed role ${managedArn}; keep that role for this workload`,
+      );
+    }
+    const priorArns = (prior as Record<string, unknown>)["assume_role_arns"];
+    if (!Array.isArray(priorArns) || priorArns.length === 0) continue;
+    if (currentRole !== nextRole) {
+      throw new CliError(
+        `aws.services.${name}.taskRoleArn cannot change while its existing assumeRoleArns policy is active; first remove assumeRoleArns while keeping taskRoleArn set to ${currentRole}, apply and deploy, then change taskRoleArn in a later change`,
+      );
+    }
+  }
+}
+
+function managedTaskRoles(existing: string): Set<string> {
+  const services = hclJson(existing, "services");
+  if (typeof services !== "object" || services === null || Array.isArray(services)) return new Set();
+  return new Set(
+    Object.entries(services)
+      .filter(
+        ([, service]) =>
+          typeof service === "object" &&
+          service !== null &&
+          !Array.isArray(service) &&
+          (service as Record<string, unknown>)["manage_task_role"] === true,
+      )
+      .map(([name]) => name),
+  );
+}
+
 function derivedValues(
   config: QmConfig,
   declared: readonly string[],
+  managedRoles = new Set<string>(),
 ): { strings: Record<string, string>; json: Record<string, unknown> } {
   if (!config.aws) throw new CliError("terraform rendering requires target aws and an aws block");
   const aws = config.aws;
@@ -84,19 +154,26 @@ function derivedValues(
     );
   }
   const services = Object.fromEntries(
-    Object.entries(aws.services).map(([name, service]) => [
-      name,
-      {
-        ecr_repository: service!.ecrRepository,
-        ecs_service: service!.ecsService,
-        cpu: service!.cpu,
-        memory: service!.memory,
-        architecture: awsWorkloadArchitecture(config, name),
-        internal_port: isServiceName(name) ? serviceDef(name).docker.internalPort : 8080,
-        ...(service!.taskRoleArn ? { task_role_arn: service!.taskRoleArn } : {}),
-        ...(service!.executionRoleArn ? { execution_role_arn: service!.executionRoleArn } : {}),
-      },
-    ]),
+    Object.entries(aws.services).map(([name, service]) => {
+      const manageTaskRole =
+        name !== "core" &&
+        (managedRoles.has(name) || (!service!.taskRoleArn && Boolean(service!.assumeRoleArns?.length)));
+      return [
+        name,
+        {
+          ecr_repository: service!.ecrRepository,
+          ecs_service: service!.ecsService,
+          cpu: service!.cpu,
+          memory: service!.memory,
+          architecture: awsWorkloadArchitecture(config, name),
+          internal_port: isServiceName(name) ? serviceDef(name).docker.internalPort : 8080,
+          ...(service!.taskRoleArn ? { task_role_arn: service!.taskRoleArn } : {}),
+          ...(service!.executionRoleArn ? { execution_role_arn: service!.executionRoleArn } : {}),
+          ...(service!.assumeRoleArns !== undefined ? { assume_role_arns: service!.assumeRoleArns } : {}),
+          ...(manageTaskRole ? { manage_task_role: true } : {}),
+        },
+      ];
+    }),
   );
   const secrets = computedSecrets(config);
   return {
@@ -140,7 +217,8 @@ export function terraformVars(
   existing = "",
   declared: string[] = [...Object.keys(OPERATOR_DEFAULTS), "github_environment"],
 ): string {
-  const { strings, json } = derivedValues(config, declared);
+  assertSafeAssumeRoleTransition(config, existing);
+  const { strings, json } = derivedValues(config, declared, managedTaskRoles(existing));
   const line = (name: string, value: string): string => `${name.padEnd(19)} = ${value}`;
   const lines = Object.entries(strings).map(([name, value]) => line(name, JSON.stringify(value)));
   for (const name of new Set([...Object.keys(OPERATOR_DEFAULTS), ...declared])) {
@@ -159,21 +237,13 @@ export function terraformVarsDrift(
   existing: string,
   declared: string[] = ["github_environment"],
 ): string[] {
-  const { strings, json } = derivedValues(config, declared);
+  const { strings, json } = derivedValues(config, declared, managedTaskRoles(existing));
   const drift: string[] = [];
   for (const [name, value] of Object.entries(strings)) {
     if (hclString(existing, name) !== value) drift.push(name);
   }
   for (const [name, value] of Object.entries(json)) {
-    const assignment = hclAssignment(existing, name);
-    const raw = assignment?.slice(assignment.indexOf("=") + 1).trim();
-    let parsed: unknown;
-    try {
-      parsed = raw === undefined ? undefined : JSON.parse(raw);
-    } catch {
-      parsed = undefined;
-    }
-    if (canonicalJson(parsed) !== canonicalJson(value)) drift.push(name);
+    if (canonicalJson(hclJson(existing, name)) !== canonicalJson(value)) drift.push(name);
   }
   return drift;
 }
@@ -183,9 +253,29 @@ function declaredInDir(configDir: string): string[] | undefined {
   return existsSync(path) ? declaredVariables(readFileSync(path, "utf8")) : undefined;
 }
 
+export function assertTerraformScaffoldSupportsConfig(config: QmConfig, configDir: string): void {
+  if (!Object.values(config.aws?.services ?? {}).some((service) => service?.assumeRoleArns !== undefined)) return;
+  const tfvarsPath = join(configDir, "infra", "terraform.tfvars");
+  if (!existsSync(tfvarsPath)) return;
+  const variablesPath = join(configDir, "infra", "variables.tf");
+  const mainPath = join(configDir, "infra", "main.tf");
+  const variables = existsSync(variablesPath) ? readFileSync(variablesPath, "utf8") : "";
+  const main = existsSync(mainPath) ? readFileSync(mainPath, "utf8") : "";
+  if (
+    !/assume_role_arns\s*=\s*optional/.test(variables) ||
+    !/manage_task_role\s*=\s*optional/.test(variables) ||
+    !/qm_scaffold_version\s*=\s*3\b/.test(main)
+  ) {
+    throw new CliError(
+      "the vendored AWS scaffold predates aws.services.*.assumeRoleArns; update infra/variables.tf and infra/main.tf from the current scaffold before configuring it",
+    );
+  }
+}
+
 export function renderTerraformVars(config: QmConfig, configDir: string): void {
   const path = join(configDir, "infra", "terraform.tfvars");
   if (!existsSync(path)) throw new CliError(`${path} does not exist; scaffold it with qm init --target aws`);
+  assertTerraformScaffoldSupportsConfig(config, configDir);
   const existing = readFileSync(path, "utf8");
   const declared = declaredInDir(configDir);
   writeFileSync(path, terraformVars(config, existing, ...(declared ? [declared] : [])));
diff --git a/cli/templates/aws/main.tf b/cli/templates/aws/main.tf
index ea83f5e2..25fb760e 100644
--- a/cli/templates/aws/main.tf
+++ b/cli/templates/aws/main.tf
@@ -1,21 +1,36 @@
 locals {
-  tags                       = { Deployment = var.org_id, ManagedBy = "terraform" }
-  azs                        = length(data.aws_availability_zones.available.names) >= 2 ? slice(data.aws_availability_zones.available.names, 0, 2) : []
-  subnet_ids                 = values(aws_subnet.public)[*].id
-  vpc_id                     = aws_vpc.this.id
-  has_portal                 = contains(keys(var.services), "portal")
-  public_service_names       = local.has_portal ? ["portal"] : ["core"]
-  ingress_services           = { for name, service in var.services : name => service if contains(local.public_service_names, name) }
-  direct_path_services       = local.has_portal ? {} : { core = ["/v1/*"] }
-  alb_name                   = "${substr(var.cluster_name, 0, 23)}-${substr(sha1(var.cluster_name), 0, 8)}"
-  service_security_groups    = [aws_security_group.services.id]
-  default_task_role_arn      = aws_iam_role.task.arn
-  core_task_role_arn         = aws_iam_role.core_task.arn
+  qm_scaffold_version     = 3
+  tags                    = { Deployment = var.org_id, ManagedBy = "terraform" }
+  azs                     = length(data.aws_availability_zones.available.names) >= 2 ? slice(data.aws_availability_zones.available.names, 0, 2) : []
+  subnet_ids              = values(aws_subnet.public)[*].id
+  vpc_id                  = aws_vpc.this.id
+  has_portal              = contains(keys(var.services), "portal")
+  public_service_names    = local.has_portal ? ["portal"] : ["core"]
+  ingress_services        = { for name, service in var.services : name => service if contains(local.public_service_names, name) }
+  direct_path_services    = local.has_portal ? {} : { core = ["/v1/*"] }
+  alb_name                = "${substr(var.cluster_name, 0, 23)}-${substr(sha1(var.cluster_name), 0, 8)}"
+  service_security_groups = [aws_security_group.services.id]
+  default_task_role_arn   = aws_iam_role.task.arn
+  core_task_role_arn      = aws_iam_role.core_task.arn
+  assume_role_services    = { for name, service in var.services : name => service if try(length(service.assume_role_arns), 0) > 0 }
+  managed_assume_role_services = {
+    for name, service in var.services : name => service if service.manage_task_role
+  }
+  managed_assume_role_policy_services = {
+    for name, service in local.assume_role_services : name => service if service.manage_task_role
+  }
+  configured_assume_role_services = {
+    for name, service in local.assume_role_services : name => service if !service.manage_task_role
+  }
+  effective_task_role_arns = {
+    for name, service in var.services : name => coalesce(
+      service.task_role_arn,
+      try(service.manage_task_role ? aws_iam_role.assume_role_task[name].arn : null, null),
+      name == "core" ? local.core_task_role_arn : local.default_task_role_arn,
+    )
+  }
   default_execution_role_arn = aws_iam_role.task_execution.arn
-  task_role_arns = distinct(compact(concat(
-    [local.default_task_role_arn, local.core_task_role_arn],
-    [for service in values(var.services) : service.task_role_arn],
-  )))
+  task_role_arns             = distinct(values(local.effective_task_role_arns))
   execution_role_arns = distinct(compact(concat(
     [local.default_execution_role_arn],
     [for service in values(var.services) : service.execution_role_arn],
@@ -181,6 +196,32 @@ resource "aws_iam_role" "core_task" {
   tags               = local.tags
 }
 
+resource "aws_iam_role" "assume_role_task" {
+  for_each           = local.managed_assume_role_services
+  name               = "${var.cluster_name}-${each.key}-task"
+  assume_role_policy = jsonencode({ Version = "2012-10-17", Statement = [{ Effect = "Allow", Principal = { Service = "ecs-tasks.amazonaws.com" }, Action = "sts:AssumeRole" }] })
+  tags               = local.tags
+}
+
+resource "aws_iam_role_policy" "managed_service_assume_role" {
+  for_each = local.managed_assume_role_policy_services
+  role     = aws_iam_role.assume_role_task[each.key].id
+  policy = jsonencode({
+    Version   = "2012-10-17"
+    Statement = [{ Effect = "Allow", Action = ["sts:AssumeRole"], Resource = each.value.assume_role_arns }]
+  })
+}
+
+resource "aws_iam_role_policy" "configured_service_assume_role" {
+  for_each = local.configured_assume_role_services
+  role     = basename(local.effective_task_role_arns[each.key])
+  policy = jsonencode({
+    Version   = "2012-10-17"
+    Statement = [{ Effect = "Allow", Action = ["sts:AssumeRole"], Resource = each.value.assume_role_arns }]
+  })
+  lifecycle { create_before_destroy = true }
+}
+
 resource "aws_cloudwatch_log_group" "microvm" {
   name              = "/aws/lambda/microvms/${var.deploy_microvm_image}"
   retention_in_days = 30
@@ -353,10 +394,13 @@ resource "aws_iam_role_policy" "github_deploy" {
         Condition = { StringEquals = { "iam:PassedToService" = "ecs-tasks.amazonaws.com" } }
       },
       {
-        Sid      = "InspectDeployRoles"
-        Effect   = "Allow"
-        Action   = ["iam:GetRole"]
-        Resource = [aws_iam_role.github_deploy.arn, aws_iam_role.task_execution.arn, aws_iam_role.task.arn, aws_iam_role.core_task.arn, aws_iam_role.microvm_build.arn, var.deploy_microvm_execution_role_arn]
+        Sid    = "InspectDeployRoles"
+        Effect = "Allow"
+        Action = ["iam:GetRole"]
+        Resource = concat(
+          [aws_iam_role.github_deploy.arn, aws_iam_role.task_execution.arn, aws_iam_role.task.arn, aws_iam_role.core_task.arn, aws_iam_role.microvm_build.arn, var.deploy_microvm_execution_role_arn],
+          [for role in aws_iam_role.assume_role_task : role.arn],
+        )
       },
       {
         Sid    = "ManageStackMicrovmImage"
@@ -606,7 +650,7 @@ resource "aws_iam_role_policy" "task_objects" {
         Resource = aws_s3_bucket.objects.arn
       },
       {
-        Effect   = "Allow"
+        Effect = "Allow"
         # AbortMultipartUpload is its own action — PutObject covers Create/UploadPart/Complete but
         # not the abort, and without it a failed staging upload strands parts that bill silently.
         Action   = ["s3:GetObject", "s3:PutObject", "s3:DeleteObject", "s3:AbortMultipartUpload", "s3:ListMultipartUploadParts"]
@@ -783,7 +827,7 @@ resource "aws_ecs_task_definition" "bootstrap" {
   network_mode             = "awsvpc"
   requires_compatibilities = ["FARGATE"]
   execution_role_arn       = coalesce(each.value.execution_role_arn, local.default_execution_role_arn)
-  task_role_arn            = coalesce(each.value.task_role_arn, each.key == "core" ? local.core_task_role_arn : local.default_task_role_arn)
+  task_role_arn            = local.effective_task_role_arns[each.key]
   runtime_platform {
     operating_system_family = "LINUX"
     cpu_architecture        = each.value.architecture == "amd64" ? "X86_64" : "ARM64"
diff --git a/cli/templates/aws/variables.tf b/cli/templates/aws/variables.tf
index 2a06202f..7c2d2448 100644
--- a/cli/templates/aws/variables.tf
+++ b/cli/templates/aws/variables.tf
@@ -102,6 +102,8 @@ variable "services" {
     internal_port      = number
     task_role_arn      = optional(string)
     execution_role_arn = optional(string)
+    assume_role_arns   = optional(set(string))
+    manage_task_role   = optional(bool, false)
   }))
 }
 variable "secret_names" { type = set(string) }
diff --git a/cli/test/aws.test.ts b/cli/test/aws.test.ts
index 8f045e89..b99361ce 100644
--- a/cli/test/aws.test.ts
+++ b/cli/test/aws.test.ts
@@ -1004,9 +1004,9 @@ test("AWS up scales services to the configured desired count and live check flag
   const dockerBin = join(dir, "docker");
   writeFileSync(dockerBin, `#!/usr/bin/env node\nconsole.log("Digest: sha256:${"a".repeat(64)}");\n`);
   chmodSync(dockerBin, 0o755);
-  const scaled = (): QmConfig => {
+  const scaled = (desiredCount = 2): QmConfig => {
     const base = oneServiceConfig();
-    return { ...base, aws: { ...base.aws!, services: { core: { ...base.aws!.services.core!, desiredCount: 2 } } } };
+    return { ...base, aws: { ...base.aws!, services: { core: { ...base.aws!.services.core!, desiredCount } } } };
   };
   const fake = statefulAws(dir, scaled());
   const priorPath = process.env.PATH;
@@ -1017,8 +1017,16 @@ test("AWS up scales services to the configured desired count and live check flag
     assert.equal(state.services["acme-core"].desiredCount, 2);
     assert.match(readFileSync(fake.log, "utf8"), /ecs update-service .*--desired-count 2/);
     await assert.doesNotReject(() => awsCheckLive(scaled(), { report: false }));
-    state.services["acme-core"].desiredCount = 1;
-    writeFileSync(fake.state, JSON.stringify(state));
+    await awsUp(scaled(3), dir, { yes: true });
+    const rescaled = JSON.parse(readFileSync(fake.state, "utf8"));
+    const currentId = rescaled.dynamo["deployment/current"].manifestId.S;
+    const current = JSON.parse(rescaled.dynamo[`deployment/manifest/${currentId}`].manifest.S);
+    assert.deepEqual(current.counts, { core: 3 });
+    await awsRollback(scaled(3));
+    const rolledBack = JSON.parse(readFileSync(fake.state, "utf8"));
+    assert.equal(rolledBack.services["acme-core"].desiredCount, 2);
+    rolledBack.services["acme-core"].desiredCount = 1;
+    writeFileSync(fake.state, JSON.stringify(rolledBack));
     await assert.rejects(
       () => awsCheckLive(scaled(), { report: false }),
       /core: runtime is ACTIVE with 1\/1 running, expected 2/,
@@ -1945,7 +1953,14 @@ test("AWS renders third-party plugins as private ECS workloads with scoped secre
       ...config.aws!,
       services: {
         ...config.aws!.services,
-        linear: { ecrRepository: "qm-linear", ecsService: "acme-linear", cpu: 256, memory: 512, architecture: "amd64" },
+        linear: {
+          ecrRepository: "qm-linear",
+          ecsService: "acme-linear",
+          cpu: 256,
+          memory: 512,
+          architecture: "amd64",
+          assumeRoleArns: ["arn:aws:iam::111122223333:role/model-gateway"],
+        },
       },
     },
   };
@@ -1955,6 +1970,7 @@ test("AWS renders third-party plugins as private ECS workloads with scoped secre
     LINEAR_TOKEN: "arn:linear-token",
   });
   const container = task.containerDefinitions[0]!;
+  assert.equal(task.taskRoleArn, "arn:aws:iam::123456789012:role/acme-qm-linear-task");
   assert.equal(container.name, "linear");
   assert.deepEqual(
     Object.fromEntries(
@@ -1973,6 +1989,37 @@ test("AWS renders third-party plugins as private ECS workloads with scoped secre
   ]);
 });
 
+test("AWS coreless workloads receive no core endpoint or source-auth secret", () => {
+  const pluginConfig: QmConfig = {
+    ...config,
+    plugins: [{ name: "signer", image: "ghcr.io/acme/signer:1", coreAccess: false }],
+    aws: {
+      ...config.aws!,
+      services: {
+        ...config.aws!.services,
+        signer: {
+          ecrRepository: "qm-signer",
+          ecsService: "acme-signer",
+          cpu: 256,
+          memory: 512,
+          architecture: "arm64",
+        },
+      },
+    },
+  };
+  const image = `123456789012.dkr.ecr.us-west-2.amazonaws.com/qm-signer@sha256:${"c".repeat(64)}`;
+  const task = renderTaskDefinition(pluginConfig, "signer", image, {
+    CORE_SIGNING_SECRET: "arn:core-signing",
+  });
+  const container = task.containerDefinitions[0]!;
+  const environment = Object.fromEntries(
+    (container.environment as Array<{ name: string; value: string }>).map(({ name, value }) => [name, value]),
+  );
+  assert.equal(environment.CORE_API_URL, undefined);
+  assert.equal(environment.CORE_ORG_ID, "acme");
+  assert.deepEqual(container.secrets, []);
+});
+
 test("AWS fixes third-party plugin PORT to its ECS port mapping", () => {
   const pluginConfig: QmConfig = {
     ...config,
@@ -2220,7 +2267,10 @@ test("AWS secret upload registers and records a task revision for a newly suppli
     required.map((secret) => [secret.name, "arn:aws:secretsmanager:us-west-2:123456789012:secret:test-AbCdEf"]),
   );
   state.definitions[oldTask] = renderTaskDefinition(secretsConfig, "core", image, arns);
-  state.dynamo = manifestItems([{ id: "current", imageLabel: "release", tasks: { core: oldTask } }], "current");
+  state.dynamo = manifestItems(
+    [{ id: "current", imageLabel: "release", tasks: { core: oldTask }, counts: { core: 0 } }],
+    "current",
+  );
   writeFileSync(fake.state, JSON.stringify(state));
   try {
     await awsSecretsPush(secretsConfig, dir);
@@ -2233,6 +2283,7 @@ test("AWS secret upload registers and records a task revision for a newly suppli
     assert.notEqual(currentId, "current");
     const manifest = JSON.parse(after.dynamo[`deployment/manifest/${currentId}`].manifest.S);
     assert.notEqual(manifest.tasks.core, oldTask);
+    assert.deepEqual(manifest.counts, { core: 0 });
     const names = after.definitions[manifest.tasks.core].containerDefinitions[0].secrets.map(
       (secret: { name: string }) => secret.name,
     );
@@ -2332,6 +2383,7 @@ function manifestItems(
     sandboxImage?: string;
     dbSnapshot?: string;
     tasks: Record<string, string>;
+    counts?: Record<string, number>;
     imageProvenance?: Record<
       string,
       | { kind: "configured"; source: string }
@@ -3446,6 +3498,70 @@ test("AWS up requires a complete trusted baseline before a partial deployment",
   }
 });
 
+test("AWS up can introduce a selected workload onto a trusted deployment baseline", async () => {
+  const dir = mkdtempSync(join(tmpdir(), "qm-aws-partial-add-"));
+  const multi = twoServiceConfig();
+  const coreTask = "arn:aws:ecs:us-west-2:123456789012:task-definition/acme-core:1";
+  const baseline = statefulAws(
+    dir,
+    multi,
+    manifestItems(
+      [
+        {
+          id: "baseline",
+          imageLabel: "previous",
+          tasks: { core: coreTask },
+          imageProvenance: { core: { kind: "configured", source: "ghcr.io/acme/qm-core:0.1.0" } },
+        },
+      ],
+      "baseline",
+    ),
+  );
+  const state = JSON.parse(readFileSync(baseline.state, "utf8"));
+  state.definitions[coreTask] = {
+    containerDefinitions: [
+      {
+        name: "core",
+        image: `123456789012.dkr.ecr.us-west-2.amazonaws.com/qm-core@sha256:${"d".repeat(64)}`,
+      },
+    ],
+  };
+  writeFileSync(baseline.state, JSON.stringify(state));
+  await assert.rejects(
+    () => awsUp(multi, dir, { dryRun: true, only: ["web-ui"] }),
+    /selected workload web-ui .* must be scaled to zero before adoption/,
+  );
+  state.services["acme-web-ui"].desiredCount = 0;
+  writeFileSync(baseline.state, JSON.stringify(state));
+  const dockerBin = join(dir, "docker");
+  writeFileSync(dockerBin, "#!/bin/sh\nexit 0\n");
+  chmodSync(dockerBin, 0o755);
+  const priorPath = process.env.PATH;
+  process.env.PATH = `${dir}:${priorPath}`;
+  try {
+    await awsUp(multi, dir, { yes: true, only: ["web-ui"] });
+    const after = JSON.parse(readFileSync(baseline.state, "utf8"));
+    const currentId = after.dynamo["deployment/current"].manifestId.S;
+    const current = JSON.parse(after.dynamo[`deployment/manifest/${currentId}`].manifest.S);
+    assert.equal(current.tasks.core, coreTask);
+    assert.match(current.tasks["web-ui"], /task-definition\/acme-web-ui:2$/);
+    const previous = JSON.parse(after.dynamo["deployment/manifest/baseline"].manifest.S);
+    assert.match(previous.tasks["web-ui"], /task-definition\/acme-web-ui:1$/);
+    assert.equal(previous.counts["web-ui"], 0);
+    const calls = readFileSync(baseline.log, "utf8");
+    assert.doesNotMatch(calls, /ecs update-service .*--service acme-core/);
+    assert.match(calls, /ecs update-service .*--service acme-web-ui/);
+    await awsRollback(multi);
+    const rolledBack = JSON.parse(readFileSync(baseline.state, "utf8"));
+    assert.match(rolledBack.services["acme-web-ui"].taskDefinition, /task-definition\/acme-web-ui:1$/);
+    assert.equal(rolledBack.services["acme-web-ui"].desiredCount, 0);
+  } finally {
+    process.env.PATH = priorPath;
+    baseline.restore();
+    rmSync(dir, { recursive: true, force: true });
+  }
+});
+
 test("AWS up cleans staging tags when ECS deployment fails", async () => {
   const dir = mkdtempSync(join(tmpdir(), "qm-aws-up-cleanup-"));
   const dockerBin = join(dir, "docker");
@@ -3727,7 +3843,14 @@ test("sandbox pin repoints a live core through a re-rendered task and records th
     dir,
     single,
     manifestItems(
-      [{ id: "baseline", imageLabel: "previous", sandboxImage: OLD_PIN, tasks: { core: CORE_TASK } }],
+      [
+        {
+          id: "baseline",
+          imageLabel: "previous",
+          sandboxImage: OLD_PIN,
+          tasks: { core: CORE_TASK },
+        },
+      ],
       "baseline",
     ),
     { failTransactionPuts: 2 },
@@ -3764,7 +3887,15 @@ test("sandbox pin on a scaled-to-zero core records a carried manifest without to
     dir,
     single,
     manifestItems(
-      [{ id: "baseline", imageLabel: "previous", sandboxImage: OLD_PIN, tasks: { core: CORE_TASK } }],
+      [
+        {
+          id: "baseline",
+          imageLabel: "previous",
+          sandboxImage: OLD_PIN,
+          tasks: { core: CORE_TASK },
+          counts: { core: 0 },
+        },
+      ],
       "baseline",
     ),
   );
@@ -3782,6 +3913,7 @@ test("sandbox pin on a scaled-to-zero core records a carried manifest without to
       "the pin is carried on the recorded tasks; it takes effect on the next up",
     );
     assert.equal(manifest.imageLabel, "previous");
+    assert.deepEqual(manifest.counts, { core: 0 });
     assert.doesNotMatch(readFileSync(fake.log, "utf8"), /ecs update-service|ecs register-task-definition/);
   } finally {
     fake.restore();
diff --git a/cli/test/config.test.ts b/cli/test/config.test.ts
index 7d62298d..f4c2c8fa 100644
--- a/cli/test/config.test.ts
+++ b/cli/test/config.test.ts
@@ -101,6 +101,20 @@ test("plugins: image is OPTIONAL (source plugins); env attaches to either; bad i
   withConfig({ plugins: [{ name: "linear", image: "ghcr.io/acme/linear:1" }] }, ({ path }) => {
     assert.equal(loadConfigAt(path).config.plugins[0]!.image, "ghcr.io/acme/linear:1");
   });
+  withConfig({ plugins: [{ name: "signer", image: "ghcr.io/acme/signer:1", coreAccess: false }] }, ({ path }) => {
+    assert.equal(loadConfigAt(path).config.plugins[0]!.coreAccess, false);
+  });
+  withConfig({ plugins: [{ name: "signer", coreAccess: "no" }] }, ({ path }) =>
+    assert.throws(() => loadConfigAt(path), /coreAccess must be a boolean/),
+  );
+  for (const name of ["CORE_API_URL", "CORE_SIGNING_SECRET"]) {
+    withConfig({ plugins: [{ name: "signer", coreAccess: false, env: { [name]: "forbidden" } }] }, ({ path }) =>
+      assert.throws(() => loadConfigAt(path), new RegExp(`cannot declare ${name} when coreAccess is false`)),
+    );
+    withConfig({ plugins: [{ name: "signer", coreAccess: false, secrets: [{ name }] }] }, ({ path }) =>
+      assert.throws(() => loadConfigAt(path), new RegExp(`cannot declare ${name} when coreAccess is false`)),
+    );
+  }
   withConfig({ plugins: [{ name: "x", image: "" }] }, ({ path }) =>
     assert.throws(() => loadConfigAt(path), /image must be a non-empty string/),
   );
@@ -194,6 +208,31 @@ test("AWS workload architecture accepts arm64 or amd64 only", () => {
   withConfig({ target: "aws", aws }, ({ path }) => {
     assert.equal(loadConfigAt(path).config.aws!.services.core!.architecture, "amd64");
   });
+  withConfig(
+    {
+      target: "aws",
+      aws: { ...aws, services: { core: { ...aws.services.core, assumeRoleArns: [] } } },
+    },
+    ({ path }) => assert.throws(() => loadConfigAt(path), /assumeRoleArns.*must contain at least one IAM role ARN/),
+  );
+  withConfig(
+    {
+      target: "aws",
+      services: ["core", "web-ui"],
+      aws: {
+        ...aws,
+        services: {
+          core: {
+            ...aws.services.core,
+            taskRoleArn: "arn:aws:iam::123456789012:role/acme-task",
+            assumeRoleArns: ["arn:aws:iam::111122223333:role/model-gateway"],
+          },
+          "web-ui": { ecrRepository: "web-ui", ecsService: "acme-web-ui", cpu: 512, memory: 1024 },
+        },
+      },
+    },
+    ({ path }) => assert.throws(() => loadConfigAt(path), /taskRoleArn.*must be unique.*also used by web-ui/),
+  );
   withConfig(
     { target: "aws", aws: { ...aws, services: { core: { ...aws.services.core, architecture: "ppc64" } } } },
     ({ path }) => {
@@ -223,6 +262,70 @@ test("AWS workload architecture accepts arm64 or amd64 only", () => {
   );
 });
 
+test("AWS workloads accept scoped commercial IAM assume-role targets", () => {
+  const aws = {
+    accountId: "123456789012",
+    region: "us-west-2",
+    cluster: "acme",
+    deployRoleArn: "arn:aws:iam::123456789012:role/deploy",
+    secretsPrefix: "acme/",
+    imageLabel: "release",
+    networking: { cloudMapNamespace: "acme.internal" },
+    services: {
+      core: {
+        ecrRepository: "core",
+        ecsService: "acme-core",
+        cpu: 512,
+        memory: 1024,
+        assumeRoleArns: [
+          "arn:aws:iam::111122223333:role/model-gateway",
+          "arn:aws:iam::111122223333:role/model-gateway",
+        ],
+      },
+    },
+  };
+  withConfig({ target: "aws", aws }, ({ path }) => {
+    assert.deepEqual(loadConfigAt(path).config.aws!.services.core!.assumeRoleArns, [
+      "arn:aws:iam::111122223333:role/model-gateway",
+    ]);
+  });
+  withConfig(
+    {
+      target: "aws",
+      aws: {
+        ...aws,
+        services: {
+          core: {
+            ...aws.services.core,
+            assumeRoleArns: ["arn:aws-us-gov:iam::111122223333:role/model-gateway"],
+          },
+        },
+      },
+    },
+    ({ path }) => assert.throws(() => loadConfigAt(path), /commercial AWS IAM role ARNs/),
+  );
+  withConfig(
+    {
+      target: "aws",
+      aws: {
+        ...aws,
+        cluster: "a".repeat(49),
+        services: {
+          "model-gateway-signer": {
+            ...aws.services.core,
+            assumeRoleArns: ["arn:aws:iam::111122223333:role/model-gateway"],
+          },
+        },
+      },
+    },
+    ({ path }) =>
+      assert.throws(
+        () => loadConfigAt(path),
+        /requires an explicit taskRoleArn because the derived IAM role name exceeds 64 characters/,
+      ),
+  );
+});
+
 test("AWS config rejects public surfaces without the HTTPS portal and real harnesses over HTTP", () => {
   const service = (name: string) => ({
     ecrRepository: `qm-${name}`,
diff --git a/cli/test/docker-secrets.test.ts b/cli/test/docker-secrets.test.ts
index 98a34a9a..1da545e2 100644
--- a/cli/test/docker-secrets.test.ts
+++ b/cli/test/docker-secrets.test.ts
@@ -85,6 +85,7 @@ test("docker up delivers secrets via a 0600 env-file, never on the docker argv",
             env: { LINEAR_REGION: "us", PLUG_TOKEN: "config-placeholder" },
             secrets: [{ name: "PLUG_TOKEN" }, { name: "EMPTY_TOKEN", required: false }],
           },
+          { name: "signer", image: "ghcr.io/acme/signer:1", coreAccess: false },
         ],
         sandbox: {
           app: "sekrit-sandboxes",
@@ -152,6 +153,14 @@ test("docker up delivers secrets via a 0600 env-file, never on the docker argv",
     assert.ok(argv.includes("FLY_SANDBOX_APP_NAME=sekrit-sandboxes"), "non-secret env still flows as -e");
     assert.ok(argv.includes("FLY_RESIDENT_ENV_TZ=UTC"), "sandbox.env literals are not secrets");
     assert.ok(argv.includes("LINEAR_REGION=us"), "undeclared plugin env still flows as -e");
+    const signerArgs = argv
+      .split("\n")
+      .filter(Boolean)
+      .map((line) => JSON.parse(line) as string[])
+      .find((args) => args.includes("qm-sekrit-signer"));
+    assert.ok(signerArgs, "coreless plugin starts");
+    assert.ok(!signerArgs.includes("CORE_API_URL=http://core:8080"), "coreless plugin gets no core endpoint");
+    assert.ok(!signerArgs.includes("--env-file"), "coreless plugin gets no source-auth secret");
     assert.ok(argv.includes("SECURITY_SCREEN_BACKEND=proxy"));
     assert.ok(argv.includes("SECURITY_SCREEN_PROXY_PROVIDER=example-screen"));
     assert.ok(argv.includes("SECURITY_SCREEN_PROXY_ENDPOINT=https://screen.example.test/classify"));
diff --git a/cli/test/doctor.test.ts b/cli/test/doctor.test.ts
index b94130b0..dd60f101 100644
--- a/cli/test/doctor.test.ts
+++ b/cli/test/doctor.test.ts
@@ -141,6 +141,43 @@ if (app === "acme-core") process.stdout.write("CAPABILITY_SECRET\\nCONNECTOR_SEC
   }
 });
 
+test("Fly doctor rejects persisted core access on coreless plugins", async () => {
+  const dir = mkdtempSync(join(tmpdir(), "qm-fly-doctor-coreless-"));
+  const bin = join(dir, "fake-fly.cjs");
+  const prior = process.env.FLY_BIN;
+  writeFileSync(
+    bin,
+    `#!/usr/bin/env node
+const args = process.argv.slice(2);
+const app = args[args.indexOf("-a") + 1];
+if (app === "acme-core") process.stdout.write("CAPABILITY_SECRET\\nCONNECTOR_SECRET_KEY\\nCORE_SIGNING_SECRET\\nPORTAL_IDENTITY_SECRET\\nSKILL_SIGNING_SECRET\\nFLY_API_TOKEN\\n");
+if (app === "acme-signer") process.stdout.write("CORE_API_URL\\nCORE_SIGNING_SECRET\\n");
+`,
+  );
+  chmodSync(bin, 0o755);
+  process.env.FLY_BIN = bin;
+  try {
+    await assert.rejects(
+      flyDoctor(
+        {
+          ...config,
+          target: "fly",
+          appPrefix: "acme",
+          region: "sjc",
+          flyOrg: "personal",
+          plugins: [{ name: "signer", image: "ghcr.io/acme/signer:1", coreAccess: false }],
+        },
+        dir,
+      ),
+      /unexpected CORE_API_URL[\s\S]*unexpected CORE_SIGNING_SECRET/,
+    );
+  } finally {
+    if (prior === undefined) delete process.env.FLY_BIN;
+    else process.env.FLY_BIN = prior;
+    rmSync(dir, { recursive: true, force: true });
+  }
+});
+
 test("Fly doctor demands the plain name too for a dual-role (core + sandbox) secret", async () => {
   const dir = mkdtempSync(join(tmpdir(), "qm-fly-doctor-dual-"));
   const bin = join(dir, "fake-fly.cjs");
diff --git a/cli/test/fly-sandbox.test.ts b/cli/test/fly-sandbox.test.ts
index 4d1d7f28..eddffab6 100644
--- a/cli/test/fly-sandbox.test.ts
+++ b/cli/test/fly-sandbox.test.ts
@@ -126,6 +126,33 @@ test("a plugin's entry env can override the injected wiring (entry env wins)", (
   assert.match(derivedPluginTomlFor(config, plugin), /CORE_API_URL = "http:\/\/elsewhere:9000"/);
 });
 
+test("a coreless plugin gets no core endpoint", () => {
+  const config: QmConfig = {
+    contract: 1,
+    orgId: "acme",
+    publicUrl: "https://acme.example.com",
+    target: "fly",
+    appPrefix: "qm",
+    region: "sjc",
+    flyOrg: "personal",
+    services: ["core"],
+    plugins: [],
+    skills: [],
+    env: {},
+    imageOverrides: {},
+  };
+  const plugin: ResolvedPlugin = {
+    name: "signer",
+    kind: "image",
+    image: "ghcr.io/x:1",
+    env: {},
+    coreAccess: false,
+  };
+  const toml = derivedPluginTomlFor(config, plugin);
+  assert.doesNotMatch(toml, /CORE_API_URL/);
+  assert.match(toml, /CORE_ORG_ID = "acme"/);
+});
+
 test("a plugin env value with quotes/backslashes is escaped into valid TOML", () => {
   const config: QmConfig = {
     contract: 1,
@@ -235,7 +262,7 @@ test("fly secrets push stages a dual-role secret under BOTH names on the core ap
     region: "sjc",
     flyOrg: "personal",
     services: ["core", "slack"],
-    plugins: [],
+    plugins: [{ name: "signer", image: "ghcr.io/acme/signer:1", coreAccess: false }],
     skills: [],
     env: { core: { HARNESS: "pi" } },
     imageOverrides: {},
@@ -261,7 +288,7 @@ test("fly secrets push stages a dual-role secret under BOTH names on the core ap
   );
   const fake = fakeFly(
     dir,
-    `const v = fs.readFileSync(0, "utf8"); fs.appendFileSync(${JSON.stringify(join(dir, "fly.log"))}, "value:" + v + "\\n");`,
+    `if (a === "secrets list -a acme-signer") console.log("CORE_API_URL digest\\nCORE_SIGNING_SECRET digest"); const v = fs.readFileSync(0, "utf8"); fs.appendFileSync(${JSON.stringify(join(dir, "fly.log"))}, "value:" + v + "\\n");`,
   );
   const priorAnthropic = process.env.ANTHROPIC_API_KEY;
   process.env.ANTHROPIC_API_KEY = "proc-wins";
@@ -290,6 +317,15 @@ test("fly secrets push stages a dual-role secret under BOTH names on the core ap
     );
     assert.ok(calls.includes("apps create acme-core --org personal"), "the service app exists before secret staging");
     assert.ok(calls.includes("apps create acme-srcplug --org personal"), "source plugin apps are created too");
+    assert.ok(calls.includes("apps create acme-signer --org personal"), "coreless plugin apps are created too");
+    assert.ok(
+      calls.includes("secrets unset --stage -a acme-signer CORE_API_URL CORE_SIGNING_SECRET"),
+      "coreless plugins lose previously stored core access",
+    );
+    assert.ok(
+      !calls.includes("secrets set --stage -a acme-signer CORE_SIGNING_SECRET"),
+      "coreless plugins do not get the signing secret",
+    );
     assert.ok(
       !calls.includes("apps create acme-sb --org personal"),
       "secret delivery never adopts or creates the separately managed sandbox registry app",
diff --git a/cli/test/terraform.test.ts b/cli/test/terraform.test.ts
index bb52ea45..e66800a1 100644
--- a/cli/test/terraform.test.ts
+++ b/cli/test/terraform.test.ts
@@ -1,7 +1,15 @@
 import { test } from "node:test";
 import assert from "node:assert/strict";
-import { readFileSync } from "node:fs";
-import { awsObjectStoreBucket, declaredVariables, terraformVars, terraformVarsDrift } from "../src/terraform.ts";
+import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import {
+  awsObjectStoreBucket,
+  declaredVariables,
+  renderTerraformVars,
+  terraformVars,
+  terraformVarsDrift,
+} from "../src/terraform.ts";
 import type { QmConfig } from "../src/config.ts";
 
 const DEPLOY_IMAGE = "acme-qm-sandbox";
@@ -65,6 +73,231 @@ test("new S3 buckets use AWS's default public-access block without a separate mu
   assert.doesNotMatch(mainTf, /aws_s3_bucket_public_access_block/);
 });
 
+test("each AWS workload can assume configured roles through an isolated task role", () => {
+  assert.match(
+    mainTf,
+    /assume_role_services\s*=\s*\{ for name, service in var\.services : name => service if try\(length\(service\.assume_role_arns\), 0\) > 0 \}/,
+  );
+  assert.match(mainTf, /managed_assume_role_services\s*=\s*\{[\s\S]*service\.manage_task_role[\s\S]*\}/);
+  assert.match(mainTf, /qm_scaffold_version\s*=\s*3/);
+  assert.match(
+    readFileSync(new URL("../templates/aws/variables.tf", import.meta.url), "utf8"),
+    /manage_task_role\s*=\s*optional\(bool, false\)/,
+  );
+  assert.match(mainTf, /resource "aws_iam_role" "assume_role_task"/);
+  assert.match(mainTf, /for_each\s*=\s*local\.managed_assume_role_services/);
+  assert.match(mainTf, /name\s*=\s*"\$\{var\.cluster_name\}-\$\{each\.key\}-task"/);
+  assert.match(mainTf, /resource "aws_iam_role_policy" "managed_service_assume_role"/);
+  assert.match(mainTf, /role\s*=\s*aws_iam_role\.assume_role_task\[each\.key\]\.id/);
+  assert.match(mainTf, /resource "aws_iam_role_policy" "configured_service_assume_role"/);
+  assert.match(mainTf, /role\s*=\s*basename\(local\.effective_task_role_arns\[each\.key\]\)/);
+  assert.match(mainTf, /lifecycle \{ create_before_destroy = true \}/);
+  assert.match(mainTf, /Action\s*=\s*\["sts:AssumeRole"\]/);
+  assert.match(mainTf, /Resource\s*=\s*each\.value\.assume_role_arns/);
+  assert.match(mainTf, /task_role_arn\s*=\s*local\.effective_task_role_arns\[each\.key\]/);
+  const rendered = terraformVars(
+    {
+      ...config,
+      aws: {
+        ...config.aws!,
+        services: {
+          ...config.aws!.services,
+          core: {
+            ...config.aws!.services.core!,
+            taskRoleArn: "arn:aws:iam::123456789012:role/existing-core",
+            assumeRoleArns: ["arn:aws:iam::111122223333:role/model-gateway"],
+          },
+        },
+      },
+    },
+    "",
+    declared,
+  );
+  assert.match(rendered, /"assume_role_arns": \[/);
+  assert.match(rendered, /arn:aws:iam::111122223333:role\/model-gateway/);
+  assert.match(rendered, /"task_role_arn": "arn:aws:iam::123456789012:role\/existing-core"/);
+  assert.doesNotMatch(terraformVars(config, "", declared), /assume_role_arns/);
+});
+
+test("an active assume-role policy blocks task-role replacement", () => {
+  const service = { ecrRepository: "qm-signer", ecsService: "acme-signer", cpu: 256, memory: 512 };
+  const role = "arn:aws:iam::111122223333:role/model-gateway";
+  const configured: QmConfig = {
+    ...config,
+    aws: {
+      ...config.aws!,
+      services: { ...config.aws!.services, signer: { ...service, assumeRoleArns: [role] } },
+    },
+  };
+  const rendered = terraformVars(configured, "", declared);
+  assert.match(rendered, /"manage_task_role": true/);
+  const managedRole = "arn:aws:iam::123456789012:role/acme-qm-signer-task";
+  const replacementRole = "arn:aws:iam::123456789012:role/acme-qm-task";
+  const removed: QmConfig = {
+    ...configured,
+    aws: {
+      ...configured.aws!,
+      services: { ...configured.aws!.services, signer: { ...service, taskRoleArn: replacementRole } },
+    },
+  };
+  assert.throws(() => terraformVars(removed, rendered, declared), /cannot replace the Terraform-managed role/);
+  const staged: QmConfig = {
+    ...configured,
+    aws: {
+      ...configured.aws!,
+      services: {
+        ...configured.aws!.services,
+        signer: { ...service, taskRoleArn: managedRole },
+      },
+    },
+  };
+  const stagedVars = terraformVars(staged, rendered, declared);
+  assert.match(stagedVars, /"manage_task_role": true/);
+  assert.match(stagedVars, /"task_role_arn": "arn:aws:iam::123456789012:role\/acme-qm-signer-task"/);
+  assert.throws(() => terraformVars(removed, stagedVars, declared), /cannot replace the Terraform-managed role/);
+
+  const explicit: QmConfig = {
+    ...configured,
+    aws: {
+      ...configured.aws!,
+      services: {
+        ...configured.aws!.services,
+        signer: { ...service, taskRoleArn: replacementRole, assumeRoleArns: [role] },
+      },
+    },
+  };
+  const explicitVars = terraformVars(explicit, "", declared);
+  assert.doesNotMatch(explicitVars, /manage_task_role/);
+  const conventionallyNamedExternal: QmConfig = {
+    ...configured,
+    aws: {
+      ...configured.aws!,
+      services: {
+        ...configured.aws!.services,
+        signer: { ...service, taskRoleArn: managedRole, assumeRoleArns: [role] },
+      },
+    },
+  };
+  const conventionallyNamedExternalVars = terraformVars(conventionallyNamedExternal, "", declared);
+  assert.doesNotMatch(conventionallyNamedExternalVars, /manage_task_role/);
+  assert.throws(
+    () => terraformVars(configured, conventionallyNamedExternalVars, declared),
+    /cannot be removed because .* is externally managed/,
+  );
+  assert.throws(
+    () =>
+      terraformVars(
+        {
+          ...explicit,
+          aws: {
+            ...explicit.aws!,
+            services: {
+              ...explicit.aws!.services,
+              signer: { ...service, taskRoleArn: managedRole, assumeRoleArns: [role] },
+            },
+          },
+        },
+        explicitVars,
+        declared,
+      ),
+    /taskRoleArn cannot change/,
+  );
+  const deauthorized: QmConfig = {
+    ...explicit,
+    aws: {
+      ...explicit.aws!,
+      services: {
+        ...explicit.aws!.services,
+        signer: { ...service, taskRoleArn: replacementRole },
+      },
+    },
+  };
+  const deauthorizedVars = terraformVars(deauthorized, explicitVars, declared);
+  assert.doesNotThrow(() =>
+    terraformVars(
+      {
+        ...deauthorized,
+        aws: {
+          ...deauthorized.aws!,
+          services: {
+            ...deauthorized.aws!.services,
+            signer: { ...service, taskRoleArn: managedRole },
+          },
+        },
+      },
+      deauthorizedVars,
+      declared,
+    ),
+  );
+
+  const coreRole = "arn:aws:iam::123456789012:role/existing-core";
+  const configuredCore: QmConfig = {
+    ...config,
+    aws: {
+      ...config.aws!,
+      services: {
+        core: { ...config.aws!.services.core!, taskRoleArn: coreRole, assumeRoleArns: [role] },
+      },
+    },
+  };
+  const configuredCoreVars = terraformVars(configuredCore, "", declared);
+  assert.throws(
+    () =>
+      terraformVars(
+        {
+          ...configuredCore,
+          aws: {
+            ...configuredCore.aws!,
+            services: {
+              core: {
+                ...configuredCore.aws!.services.core!,
+                taskRoleArn: "arn:aws:iam::123456789012:role/replacement-core",
+              },
+            },
+          },
+        },
+        configuredCoreVars,
+        declared,
+      ),
+    /taskRoleArn cannot change/,
+  );
+});
+
+test("assume-role config rejects vendored AWS scaffolds that predate workload roles", () => {
+  const dir = mkdtempSync(join(tmpdir(), "qm-legacy-terraform-"));
+  try {
+    const infra = join(dir, "infra");
+    mkdirSync(infra);
+    writeFileSync(join(infra, "terraform.tfvars"), "services = {}\n");
+    writeFileSync(
+      join(infra, "variables.tf"),
+      'variable "services" { type = map(object({ assume_role_arns = optional(set(string)) })) }\n',
+    );
+    writeFileSync(
+      join(infra, "main.tf"),
+      'resource "aws_iam_role" "assume_role_task" {}\nresource "aws_iam_role_policy" "managed_service_assume_role" {}\nresource "aws_iam_role_policy" "configured_service_assume_role" {}\n',
+    );
+    const configured: QmConfig = {
+      ...config,
+      aws: {
+        ...config.aws!,
+        services: {
+          core: {
+            ...config.aws!.services.core!,
+            assumeRoleArns: ["arn:aws:iam::111122223333:role/model-gateway"],
+          },
+        },
+      },
+    };
+    assert.throws(
+      () => renderTerraformVars(configured, dir),
+      /AWS scaffold predates aws\.services\.\*\.assumeRoleArns[\s\S]*variables\.tf[\s\S]*main\.tf/,
+    );
+  } finally {
+    rmSync(dir, { recursive: true, force: true });
+  }
+});
+
 test("the deploy role registers task definitions only for configured ECS families", () => {
   const policy = mainTf.match(/resource "aws_iam_role_policy" "github_deploy" \{([\s\S]*?)\n\}/)?.[1] ?? "";
   const management = policy.match(/Sid\s*= "ManageStackTaskDefinitions"([\s\S]*?)\n\s*\},/)?.[1] ?? "";
@@ -283,7 +516,7 @@ test("AWS module reuses account OIDC, guards account and passes configured task
   assert.match(versions, /allowed_account_ids\s*= \[var\.account_id\]/);
   assert.match(mainTf, /concat\(local\.execution_role_arns, local\.task_role_arns\)/);
   assert.match(mainTf, /coalesce\(each\.value\.execution_role_arn, local\.default_execution_role_arn\)/);
-  assert.match(mainTf, /each\.key == "core" \? local\.core_task_role_arn : local\.default_task_role_arn/);
+  assert.match(mainTf, /effective_task_role_arns/);
   assert.match(mainTf, /role = aws_iam_role\.core_task\.id/);
   assert.match(mainTf, /dynamodb:ConditionCheckItem/);
   assert.match(mainTf, /dynamodb:DescribeTable/);