diff --git a/packages/amico-run/src/index.ts b/packages/amico-run/src/index.ts
index dcd0ab44..03c732c2 100644
--- a/packages/amico-run/src/index.ts
+++ b/packages/amico-run/src/index.ts
@@ -5,3 +5,5 @@ export * from "./schemas.js";
 export * from "./event_queue.js";
 export * from "./local_executor.js";
 export * from "./scheduler.js";
+export * from "./remote_config.js";
+export * from "./remote_executor.js";
diff --git a/packages/amico-run/src/launch.ts b/packages/amico-run/src/launch.ts
index 8758eb7d..d00c8081 100644
--- a/packages/amico-run/src/launch.ts
+++ b/packages/amico-run/src/launch.ts
@@ -9,7 +9,8 @@ import { existsSync, readFileSync } from "node:fs";
 import { join } from "node:path";
 import { parse as parseToml } from "smol-toml";
 import { LocalExecutor } from "./local_executor.js";
-import { ConfigError, type Finished, type SubmitOpts } from "./types.js";
+import { RemoteExecutor } from "./remote_executor.js";
+import { ConfigError, type Executor, type Finished, type SubmitOpts } from "./types.js";
 import { readAuthoring } from "./authoring.js";
 import { runGate } from "./gate.js";
 import { runVerification } from "./verify.js";
@@ -23,7 +24,7 @@ function readTomlSafe(fp: string): Record<string, unknown> | undefined {
   }
 }
 
-const USAGE = `usage: amico-run <script.jl> [--executor local] [--lab <id-or-path>]
+const USAGE = `usage: amico-run <script.jl> [--executor local|remote] [--lab <id-or-path>]
                  [--runs-root <path>] [--julia <path>] [--project <path>] [--sysimage <path>]
                  [--spec <solvespec.json>]   (spec C: validate + gate before launch)
        amico-run resolve --platform <p> --kind <k> --size <n>   (tier resolution → JSON)
@@ -99,10 +100,20 @@ export async function launch(argv: string[]): Promise<number> {
     console.error(`amico-run: no script given\n${USAGE}`);
     return 64;
   }
-  if (executor !== "local") {
-    console.error(`amico-run: only --executor local is supported in β`);
+  if (executor !== "local" && executor !== "remote") {
+    console.error(`amico-run: unknown --executor ${executor} (supported: local, remote)`);
     return 64;
   }
+  if (executor === "remote" && specPath !== undefined) {
+    // Named seam: the gate could run pre-submit, but free-tier re-rollout
+    // verification (runVerification) replays LOCAL artifacts the mirror
+    // doesn't have. Reject loudly rather than half-verify.
+    console.error(`amico-run: --spec with --executor remote is not supported yet (verification is local-only)`);
+    return 64;
+  }
+  if (executor === "remote" && (opts.julia!.julia || opts.julia!.project || opts.julia!.sysimage)) {
+    console.error(`amico-run: --julia/--project/--sysimage are ignored with --executor remote (the runner image owns the environment)`);
+  }
 
   // ── spec C: the launch gate. Failures leave NO run dir and exit 64. ──
   if (specPath) {
@@ -152,7 +163,8 @@ export async function launch(argv: string[]): Promise<number> {
 
   let handle;
   try {
-    handle = await new LocalExecutor().submit(script, opts);
+    const exec: Executor = executor === "remote" ? new RemoteExecutor() : new LocalExecutor();
+    handle = await exec.submit(script, opts);
   } catch (e) {
     if (e instanceof ConfigError) {
       console.error(`amico-run: ${e.message}`);
diff --git a/packages/amico-run/src/local_executor.ts b/packages/amico-run/src/local_executor.ts
index 39da67e3..f8f278e0 100644
--- a/packages/amico-run/src/local_executor.ts
+++ b/packages/amico-run/src/local_executor.ts
@@ -167,10 +167,3 @@ export class LocalExecutor implements Executor {
     return { runId, runDir, events, finished, abort };
   }
 }
-
-/** Spec §3: interface seam only — implementation is post-β. */
-export class RemoteExecutor implements Executor {
-  submit(): Promise<RunHandle> {
-    return Promise.reject(new Error("RemoteExecutor: not implemented in β (D9 plan, Phase 2+)"));
-  }
-}
diff --git a/packages/amico-run/src/remote_config.ts b/packages/amico-run/src/remote_config.ts
new file mode 100644
index 00000000..95ed9c12
--- /dev/null
+++ b/packages/amico-run/src/remote_config.ts
@@ -0,0 +1,48 @@
+// packages/amico-run/src/remote_config.ts
+// Cloud solve-service config (Δ8) — the authoring.ts idiom (env override →
+// ~/.amico file). SECURITY: the token value must never appear in an error
+// message or log line (llm_creds.mjs stance — the secret never enters
+// amico's surfaces beyond the Authorization header itself).
+import { existsSync, readFileSync } from "node:fs";
+import { homedir } from "node:os";
+import { join } from "node:path";
+import { ConfigError } from "./types.js";
+
+export interface RemoteConfig {
+  baseUrl: string; // e.g. https://solves.staging.harmoniqs.co (no trailing slash)
+  token: string; // per-user Δ2 credential
+}
+
+/** $AMICO_CLOUD_FILE overrides the path (tests) — authoring.ts:37-41 idiom. */
+export function cloudConfigFile(env: NodeJS.ProcessEnv = process.env): string {
+  const v = env.AMICO_CLOUD_FILE;
+  if (v && v.trim() !== "") return v;
+  return join(homedir(), ".amico", "cloud.json");
+}
+
+/** Resolution order: AMICO_CLOUD_URL+AMICO_CLOUD_TOKEN env pair → cloud.json.
+ *  Any failure is ConfigError — exit-64 class (types.ts:50): nothing ran. */
+export function readRemoteConfig(env: NodeJS.ProcessEnv = process.env): RemoteConfig {
+  const url = env.AMICO_CLOUD_URL;
+  const token = env.AMICO_CLOUD_TOKEN;
+  if (url && token) return { baseUrl: url.replace(/\/+$/, ""), token };
+  if (url || token)
+    throw new ConfigError(
+      "cloud config: set BOTH AMICO_CLOUD_URL and AMICO_CLOUD_TOKEN (or neither, to use cloud.json)",
+    );
+  const file = cloudConfigFile(env);
+  if (!existsSync(file))
+    throw new ConfigError(
+      `cloud config not found: ${file} (write {"base_url","token"} or set AMICO_CLOUD_URL/AMICO_CLOUD_TOKEN)`,
+    );
+  let raw: unknown;
+  try {
+    raw = JSON.parse(readFileSync(file, "utf8"));
+  } catch {
+    throw new ConfigError(`malformed cloud config at ${file}`);
+  }
+  const d = (typeof raw === "object" && raw !== null ? raw : {}) as Record<string, unknown>;
+  if (typeof d.base_url !== "string" || d.base_url === "" || typeof d.token !== "string" || d.token === "")
+    throw new ConfigError(`cloud config at ${file} needs non-empty string keys "base_url" and "token"`);
+  return { baseUrl: d.base_url.replace(/\/+$/, ""), token: d.token };
+}
diff --git a/packages/amico-run/src/remote_executor.ts b/packages/amico-run/src/remote_executor.ts
new file mode 100644
index 00000000..82b39f90
--- /dev/null
+++ b/packages/amico-run/src/remote_executor.ts
@@ -0,0 +1,284 @@
+// packages/amico-run/src/remote_executor.ts
+// Δ8 (#32): the cloud executor. Submits via Δ2, then MIRRORS the cloud run
+// into a contract-conforming local run dir fed by the Δ4 poll endpoints —
+// so every downstream reader (Scheduler S12, RunsManager index tail, run.log
+// tail + poll backstop, FINISHED watch, stopPlan) consumes remote runs with
+// ZERO executor branches. Locked resolutions: (a) frames best-effort,
+// (b) abort()=request, (c) executor-owned warming budget, (d) terminal via
+// status poll — FINISHED authoritative, instance-gone → inferred terminal.
+import { appendFileSync, existsSync, mkdirSync, readFileSync, renameSync, utimesSync, writeFileSync } from "node:fs";
+import { basename, join, resolve } from "node:path";
+import { EventQueue } from "./event_queue.js";
+import { classifyLine } from "./telemetry.js";
+import {
+  appendIndex,
+  atomicWriteFile,
+  defaultRunsRoot,
+  deriveLabId,
+  generateRunId,
+  updateLatest,
+  writeFinished,
+  writeManifest,
+} from "./run_dir.js";
+import { readRemoteConfig, type RemoteConfig } from "./remote_config.js";
+import {
+  ConfigError,
+  type Executor,
+  type Finished,
+  type RunEvent,
+  type RunHandle,
+  type RunStatus,
+  type SubmitOpts,
+} from "./types.js";
+import pkg from "../package.json" with { type: "json" };
+
+const ORCHESTRATOR_VERSION = pkg.version;
+
+/** The remote status payload carries no exit code; mirror launch.ts's
+ *  status→exit lanes. Inferred terminals (resolution (d)) use 255. */
+const EXIT: Record<RunStatus, number> = { completed: 0, failed: 1, aborted: 130 };
+export const EXIT_INFERRED = 255;
+
+export interface RemoteExecutorOpts {
+  /** Endpoint + credential; default readRemoteConfig() (env pair → ~/.amico/cloud.json). */
+  config?: RemoteConfig;
+  /** Poll cadence, default 2000ms. Test knob (the graceMs idiom, types.ts:13) — NOT exposed in the CLI. */
+  pollMs?: number;
+  /** Resolution (c): per-executor warming budget — no sign of life (Running
+   *  status or a first iter) within this window → inferred terminal.
+   *  Default 15 min: remote cold-start ≫ local seconds. */
+  warmingBudgetMs?: number;
+  /** Client half of resolution (d): no SUCCESSFUL status poll for this long →
+   *  inferred terminal (observability lost; S3 keeps the cloud truth).
+   *  Default 10 min — deliberately the inspector's STALL_AFTER_MS. */
+  lostAfterMs?: number;
+  /** Δ2 pass-through: overridable wall-clock cap (seconds). */
+  maxWallclock?: number;
+}
+
+export class RemoteExecutor implements Executor {
+  private readonly cfgOverride?: RemoteConfig;
+  private readonly pollMs: number;
+  private readonly warmingBudgetMs: number;
+  private readonly lostAfterMs: number;
+  private readonly maxWallclock?: number;
+
+  constructor(opts: RemoteExecutorOpts = {}) {
+    this.cfgOverride = opts.config;
+    this.pollMs = opts.pollMs ?? 2000;
+    this.warmingBudgetMs = opts.warmingBudgetMs ?? 15 * 60 * 1000;
+    this.lostAfterMs = opts.lostAfterMs ?? 10 * 60 * 1000;
+    this.maxWallclock = opts.maxWallclock;
+  }
+
+  async submit(scriptPath: string, opts: SubmitOpts = {}): Promise<RunHandle> {
+    // ---- step 1 (LocalExecutor §5 parity): validate config; NO run dir on failure ----
+    const script = resolve(scriptPath);
+    if (!existsSync(script)) throw new ConfigError(`script not found: ${script}`);
+    const cfg = this.cfgOverride ?? readRemoteConfig();
+    const lab = opts.lab ?? "default";
+    const labId = deriveLabId(lab);
+    const runsRoot = opts.runsRoot ?? defaultRunsRoot(labId);
+
+    // ---- step 2: Δ2 submit — still no run dir; a rejected submit ran nothing ----
+    const payload: Record<string, unknown> = { script: readFileSync(script, "utf8"), filename: basename(script) };
+    if (this.maxWallclock !== undefined) payload.max_wallclock = this.maxWallclock;
+    let res: Response;
+    try {
+      res = await fetch(`${cfg.baseUrl}/solves`, {
+        method: "POST",
+        headers: { authorization: `Bearer ${cfg.token}`, "content-type": "application/json" },
+        body: JSON.stringify(payload),
+      });
+    } catch (e) {
+      throw new Error(`cloud submit failed: ${(e as Error).message}`);
+    }
+    // 401 is config-class (bad credential): the exit-64 lane, like a bad --julia path.
+    if (res.status === 401)
+      throw new ConfigError("cloud credential rejected (401) — check AMICO_CLOUD_TOKEN / cloud.json");
+    if (res.status !== 202) throw new Error(`cloud submit: unexpected HTTP ${res.status}`);
+    const taskId = String(((await res.json()) as { task_id?: unknown }).task_id ?? "");
+    if (taskId === "") throw new Error("cloud submit: 202 without task_id");
+
+    // ---- step 3: local MIRROR run dir — the same contract files LocalExecutor
+    // writes (its steps 2–5), so downstream needs zero executor branches (S12).
+    try {
+      mkdirSync(runsRoot, { recursive: true });
+    } catch (e) {
+      throw new ConfigError(`runs root not writable: ${runsRoot} (${(e as Error).message})`);
+    }
+    const runId = generateRunId(runsRoot);
+    const runDir = join(runsRoot, runId);
+    mkdirSync(runDir);
+    const createdAt = new Date().toISOString();
+    writeManifest(runDir, {
+      schema_version: "1",
+      run_id: runId,
+      script_path: script,
+      lab,
+      lab_id: labId,
+      created_at: createdAt,
+      orchestrator_version: ORCHESTRATOR_VERSION,
+      julia: { binary: "cloud" }, // the runner image owns the real binary; run.schema.json requires the key
+    });
+    // task_id lives in a SIDECAR: run.schema.json is additionalProperties:false (frozen contract).
+    atomicWriteFile(runDir, "remote.json", JSON.stringify({ task_id: taskId, base_url: cfg.baseUrl }) + "\n");
+    writeFileSync(join(runDir, "run.log"), ""); // exists from t0 — stall logic keys off its mtime
+    appendIndex(runsRoot, runId, createdAt, script);
+    updateLatest(runsRoot, runId);
+
+    // ---- step 4: the poll pump feeds the RunHandle ----
+    const events = new EventQueue<RunEvent>();
+    let resolveFinished!: (f: Finished) => void;
+    const finished = new Promise<Finished>((r) => {
+      resolveFinished = r;
+    });
+    let settled = false;
+    let sawLife = false; // Running status or a first iter observed (warming budget clock)
+    let iterHigh = -1; // stats high-water: Δ4 re-serves history each poll; dedup here
+    let frameHigh = -1; // frames high-water
+    let lastOkPollAt = Date.now(); // resolution (d) client half: observability clock
+    const startedAt = Date.now();
+
+    /** run.log line + event — byte-for-byte the LocalExecutor onLine path
+     *  (local_executor.ts:139-145): log first, then classifyLine → push. */
+    const emitLine = (line: string): void => {
+      appendFileSync(join(runDir, "run.log"), line + "\n");
+      events.push(classifyLine(line, "stdout"));
+    };
+
+    const settle = (status: RunStatus, exitCode: number): void => {
+      if (settled) return;
+      settled = true;
+      try {
+        writeFinished(runDir, status, exitCode); // atomic; the mirror's authoritative verdict
+      } catch (e) {
+        process.stderr.write(`amico-run: failed to write FINISHED: ${(e as Error).message}\n`);
+      }
+      events.push({ kind: "finished", status, exitCode });
+      events.close();
+      resolveFinished({ status, exitCode });
+    };
+
+    const get = (path: string): Promise<Response> =>
+      fetch(`${cfg.baseUrl}/solves/${taskId}/${path}`, { headers: { authorization: `Bearer ${cfg.token}` } });
+
+    let abortPosted = false;
+    const postAbort = async (): Promise<void> => {
+      if (abortPosted) return; // idempotent, like LocalExecutor's settled-guard
+      abortPosted = true;
+      try {
+        await fetch(`${cfg.baseUrl}/solves/${taskId}/abort`, {
+          method: "POST",
+          headers: { authorization: `Bearer ${cfg.token}` },
+        });
+      } catch {
+        /* the request is best-effort; the status poll still owns the terminal */
+      }
+    };
+
+    const pollOnce = async (): Promise<void> => {
+      // status — the authoritative terminal lane (resolution (d))
+      const sres = await get("status");
+      if (!sres.ok) throw new Error(`status HTTP ${sres.status}`);
+      const s = (await sres.json()) as {
+        task_status?: string;
+        finished?: { status?: string };
+        liveness?: string;
+      };
+      lastOkPollAt = Date.now();
+      if (s.task_status === "Running") sawLife = true;
+      // stats → synthesized AMICODE_ITER lines: run.log + events, the exact
+      // local delivery path, so tail/backstop consumers can't tell the difference.
+      try {
+        const r = await get("stats");
+        if (r.ok) {
+          const stats = (await r.json()) as { iters?: Array<Record<string, unknown>> };
+          for (const it of stats.iters ?? []) {
+            const n = Number(it.iter);
+            if (!Number.isFinite(n) || n <= iterHigh) continue; // Δ4 re-serves history: dedup on high-water
+            iterHigh = n;
+            sawLife = true;
+            emitLine(`AMICODE_ITER iter=${it.iter} f=${it.f} inf_pr=${it.inf_pr} inf_du=${it.inf_du}`);
+          }
+        }
+      } catch {
+        /* stats are advisory — status stays the authoritative lane */
+      }
+      // frames — resolution (a): best-effort; ANY failure is swallowed
+      try {
+        const r = await get("frames");
+        if (r.ok && r.status !== 204) {
+          const fr = (await r.json()) as { iter?: number; png_base64?: string };
+          if (typeof fr.iter === "number" && fr.iter > frameHigh && typeof fr.png_base64 === "string") {
+            frameHigh = fr.iter;
+            const name = `iter_${String(fr.iter).padStart(3, "0")}.png`; // the S3 layout's iter_*.png
+            const tmp = join(runDir, `.${name}.tmp`);
+            writeFileSync(tmp, Buffer.from(fr.png_base64, "base64"));
+            renameSync(tmp, join(runDir, name)); // atomic: no reader sees a torn png
+          }
+        }
+      } catch {
+        /* frames are best-effort by contract */
+      }
+      // heartbeat: a successful status poll proves the cloud channel is alive.
+      // Mirror that into run.log's MTIME (content untouched) so the inspector's
+      // disk-keyed stall logic (liveStatus/stopPlan, STALL_AFTER_MS) measures
+      // CLOUD silence — remote warming stays the executor's budget (resolution (c)).
+      try {
+        const now = new Date();
+        utimesSync(join(runDir, "run.log"), now, now);
+      } catch {
+        /* mirror deleted underneath us — the terminal lanes still settle */
+      }
+      // resolution (d): instance gone without FINISHED → inferred terminal
+      if (s.liveness === "gone" && s.finished === undefined) {
+        emitLine(`AMICODE_REMOTE_LOST instance gone without FINISHED (task ${taskId})`);
+        settle("failed", EXIT_INFERRED);
+        return;
+      }
+      const f = s.finished?.status;
+      if (f === "completed" || f === "failed" || f === "aborted") settle(f, EXIT[f]);
+    };
+
+    const pump = async (): Promise<void> => {
+      while (!settled) {
+        try {
+          await pollOnce();
+        } catch {
+          // transient poll failure: skip this tick. run.log's mtime is NOT
+          // advanced, so a sustained outage honestly reads "stalled" downstream.
+        }
+        if (settled) return;
+        // resolution (c): the executor OWNS its warming budget — no Scheduler
+        // timer exists to do this (pinned by scheduler.test.ts:285).
+        if (!sawLife && Date.now() - startedAt > this.warmingBudgetMs) {
+          emitLine(`AMICODE_REMOTE_LOST warming budget exhausted (${this.warmingBudgetMs}ms, task ${taskId})`);
+          void postAbort(); // best-effort: stop paying for the instance
+          settle("failed", EXIT_INFERRED);
+          return;
+        }
+        // resolution (d) client half: observability lost. S3 keeps the cloud
+        // truth; the mirror records an inferred verdict + breadcrumb rather
+        // than polling a dead endpoint forever (bounded pump).
+        if (Date.now() - lastOkPollAt > this.lostAfterMs) {
+          emitLine(`AMICODE_REMOTE_LOST poll endpoint unreachable for ${this.lostAfterMs}ms (task ${taskId})`);
+          settle("failed", EXIT_INFERRED);
+          return;
+        }
+        await new Promise((r) => setTimeout(r, this.pollMs));
+      }
+    };
+    void pump();
+
+    // resolution (b): abort REQUESTS termination; the run is live until the
+    // status poll delivers the real terminal. Idempotent; never rejects.
+    const abort = async (): Promise<void> => {
+      if (settled) return;
+      await postAbort();
+      await finished;
+    };
+
+    return { runId, runDir, events, finished, abort };
+  }
+}
diff --git a/packages/amico-run/test/cli.test.ts b/packages/amico-run/test/cli.test.ts
index b814297f..113aade4 100644
--- a/packages/amico-run/test/cli.test.ts
+++ b/packages/amico-run/test/cli.test.ts
@@ -3,6 +3,7 @@ import { execFileSync, execFile } from "node:child_process";
 import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs";
 import { join } from "node:path";
 import { tmpRoot, fakeJulia, readToml } from "./helpers.js";
+import { FakeCloud } from "./fake_cloud.js";
 
 const BUNDLE = join(__dirname, "..", "dist", "amico-run.js");
 beforeAll(() => {
@@ -53,11 +54,71 @@ describe("amico-run CLI", () => {
     expect(r.code).toBe(64);
     expect(r.stderr).toMatch(/unknown flag/);
   });
-  it("--executor remote → 64 (only local in β)", () => {
+  it("--executor remote without cloud config → 64 (config-class), no run dir", () => {
     const root = tmpRoot();
-    const r = run([fakeJulia(root, "s.jl", ""), "--executor", "remote"]);
+    const r = run([fakeJulia(root, "s.jl", ""), "--executor", "remote", "--runs-root", join(root, "runs")], {
+      AMICO_CLOUD_FILE: join(root, "no-such-cloud.json"), // hermetic: ignore any real ~/.amico/cloud.json
+      AMICO_CLOUD_URL: "",
+      AMICO_CLOUD_TOKEN: "",
+    });
+    expect(r.code).toBe(64);
+    expect(r.stderr).toMatch(/cloud config/);
+    expect(existsSync(join(root, "runs"))).toBe(false);
+  });
+
+  it("--executor bogus → 64 naming the supported set", () => {
+    const root = tmpRoot();
+    const r = run([fakeJulia(root, "s.jl", ""), "--executor", "bogus"]);
     expect(r.code).toBe(64);
+    expect(r.stderr).toMatch(/local, remote/);
   });
+
+  it("--spec with --executor remote → 64 (named seam: verification is local-only)", () => {
+    const root = tmpRoot();
+    writeFileSync(join(root, "spec.json"), "{}");
+    const r = run(
+      [fakeJulia(root, "s.jl", ""), "--executor", "remote", "--spec", join(root, "spec.json")],
+      { AMICO_CLOUD_URL: "http://127.0.0.1:1", AMICO_CLOUD_TOKEN: "t" },
+    );
+    expect(r.code).toBe(64);
+    expect(r.stderr).toMatch(/--spec .*remote/);
+  });
+
+  it("--executor remote full lane through the bundle (fake cloud): iter relay + AMICODE_FINISHED, exit 0", async () => {
+    const fake = new FakeCloud();
+    await fake.start();
+    fake.state = {
+      task_status: "Running",
+      liveness: "alive",
+      iters: [{ iter: 1, f: "1.0e-2", inf_pr: "1e-8", inf_du: "1e-6" }],
+      finished: { status: "completed" },
+    };
+    try {
+      const root = tmpRoot();
+      const script = fakeJulia(root, "s.jl", "");
+      // ASYNC lane (the SIGTERM-test pattern, cli.test.ts:191-196) — NEVER the
+      // sync run() helper here: FakeCloud runs IN this test process, and
+      // execFileSync would block the event loop, so the fake could never answer
+      // the child CLI's HTTP requests (deadlock until undici's timeout).
+      const r = await new Promise<{ code: number; stdout: string }>((resolveP) => {
+        let stdout = "";
+        const child = execFile(
+          "node",
+          [BUNDLE, script, "--executor", "remote", "--runs-root", join(root, "runs")],
+          { env: { ...process.env, AMICO_CLOUD_URL: fake.base, AMICO_CLOUD_TOKEN: fake.token } },
+        );
+        child.stdout!.on("data", (d: string) => {
+          stdout += d;
+        });
+        child.on("exit", (c) => resolveP({ code: c ?? -1, stdout }));
+      });
+      expect(r.code).toBe(0);
+      expect(r.stdout).toContain("AMICODE_ITER iter=1 f=1.0e-2");
+      expect(r.stdout).toMatch(/AMICODE_FINISHED status=completed exitCode=0 runDir=.+/);
+    } finally {
+      await fake.stop();
+    }
+  }, 15000);
   it("--spec: gate failure → 64, one-line stderr reason, NO run dir (spec C)", () => {
     const root = tmpRoot();
     const script = fakeJulia(root, "s.jl", "");
diff --git a/packages/amico-run/test/executor_parity.test.ts b/packages/amico-run/test/executor_parity.test.ts
new file mode 100644
index 00000000..2bc5af01
--- /dev/null
+++ b/packages/amico-run/test/executor_parity.test.ts
@@ -0,0 +1,127 @@
+// packages/amico-run/test/executor_parity.test.ts
+// S7/S11 snapshot (Δ8 AC): the SAME solve through LocalExecutor and
+// RemoteExecutor yields IDENTICAL downstream-observable state — event
+// sequence (iter/finished), FINISHED bytes, AMICODE_ITER run.log lines.
+// Frames are best-effort (S11): a broken frames endpoint changes nothing.
+import { describe, it, expect } from "vitest";
+import { existsSync, readFileSync } from "node:fs";
+import { join } from "node:path";
+import { fileURLToPath } from "node:url";
+import { tmpRoot, fakeJulia, readToml } from "./helpers.js";
+import { FakeCloud } from "./fake_cloud.js";
+import { LocalExecutor } from "../src/local_executor.js";
+import { RemoteExecutor } from "../src/remote_executor.js";
+import { validateManifest, validateFinished } from "../src/schemas.js";
+import type { RunEvent, RunHandle } from "../src/types.js";
+
+const ITERS = [
+  { iter: 1, f: "1.0e-2", inf_pr: "1e-8", inf_du: "1e-6" },
+  { iter: 2, f: "3.0e-4", inf_pr: "1e-9", inf_du: "1e-7" },
+];
+const LOCAL_BODY = ITERS.map(
+  (i) => `console.log('AMICODE_ITER iter=${i.iter} f=${i.f} inf_pr=${i.inf_pr} inf_du=${i.inf_du}')`,
+).join("\n");
+
+async function collect(h: RunHandle): Promise<RunEvent[]> {
+  const out: RunEvent[] = [];
+  for await (const e of h.events) out.push(e);
+  return out;
+}
+
+/** What downstream actually consumes: iter/finished events + contract files. */
+function observable(evs: RunEvent[], runDir: string) {
+  return {
+    events: evs
+      .filter((e) => e.kind === "iter" || e.kind === "finished")
+      .map((e) =>
+        e.kind === "iter"
+          ? { kind: e.kind, fields: e.fields }
+          : { kind: e.kind, status: e.status, exitCode: e.exitCode },
+      ),
+    finishedFile: readFileSync(join(runDir, "FINISHED"), "utf8"),
+    iterLines: readFileSync(join(runDir, "run.log"), "utf8")
+      .split("\n")
+      .filter((l) => l.startsWith("AMICODE_ITER")),
+  };
+}
+
+describe("S7/S11 — Local vs Remote over the same solve", () => {
+  it("terminal state, FINISHED bytes, run.log iter lines, event sequence: IDENTICAL", async () => {
+    const lroot = tmpRoot();
+    const lh = await new LocalExecutor().submit(fakeJulia(lroot, "s.jl", ""), {
+      runsRoot: join(lroot, "runs"),
+      julia: { julia: fakeJulia(lroot, "julia", LOCAL_BODY) },
+    });
+    const lobs = observable(await collect(lh), lh.runDir);
+
+    const fake = new FakeCloud();
+    await fake.start();
+    fake.state = { task_status: "Running", liveness: "alive", iters: ITERS, finished: { status: "completed" } };
+    try {
+      const rroot = tmpRoot();
+      const rh = await new RemoteExecutor({
+        config: { baseUrl: fake.base, token: fake.token },
+        pollMs: 10,
+      }).submit(fakeJulia(rroot, "s.jl", ""), { runsRoot: join(rroot, "runs") });
+      const robs = observable(await collect(rh), rh.runDir);
+      expect(robs).toEqual(lobs); // S7: identical, not merely similar
+      expect(validateManifest(readToml(join(lh.runDir, "run.toml"))).ok).toBe(true);
+      expect(validateManifest(readToml(join(rh.runDir, "run.toml"))).ok).toBe(true);
+      expect(validateFinished(readToml(join(rh.runDir, "FINISHED"))).ok).toBe(true);
+    } finally {
+      await fake.stop();
+    }
+  });
+
+  it("S11: framesBroken changes NOTHING about the terminal state (frames best-effort)", async () => {
+    const fake = new FakeCloud();
+    await fake.start();
+    fake.state = {
+      task_status: "Running",
+      liveness: "alive",
+      iters: ITERS,
+      finished: { status: "completed" },
+      framesBroken: true,
+    };
+    try {
+      const root = tmpRoot();
+      const h = await new RemoteExecutor({
+        config: { baseUrl: fake.base, token: fake.token },
+        pollMs: 10,
+      }).submit(fakeJulia(root, "s.jl", ""), { runsRoot: join(root, "runs") });
+      expect(await h.finished).toEqual({ status: "completed", exitCode: 0 });
+      expect(existsSync(join(h.runDir, "FINISHED"))).toBe(true);
+    } finally {
+      await fake.stop();
+    }
+  });
+
+  it("S12 (amico-run side): scheduler.ts CODE names no executor type (scheduler.test.ts:285 idiom)", () => {
+    const src = readFileSync(fileURLToPath(new URL("../src/scheduler.ts", import.meta.url)), "utf8")
+      .replace(/\/\/.*$/gm, "") // comments may NAME executors (the header prose does)
+      .replace(/\/\*[\s\S]*?\*\//g, "");
+    expect(src).not.toMatch(/LocalExecutor|RemoteExecutor|instanceof\s+\w*Executor/);
+  });
+
+  it("NAMED DIVERGENCE (Deviations #6): the remote failed lane pins exit_code 1 — local passes julia's real rc through", async () => {
+    // Local: FINISHED{failed, <rc>} carries julia's ACTUAL exit code (pinned
+    // by cli.test.ts:33-44, "julia rc 7 passes through as exit 7"). The Δ4
+    // status shape carries NO exit code, so the remote mirror maps failed→1.
+    // FINISHED bytes therefore DIFFER Local vs Remote on failure, by design —
+    // this pin keeps the divergence intentional and guarded, not accidental.
+    const fake = new FakeCloud();
+    await fake.start();
+    fake.state = { task_status: "Running", liveness: "alive", iters: [], finished: { status: "failed" } };
+    try {
+      const root = tmpRoot();
+      const h = await new RemoteExecutor({
+        config: { baseUrl: fake.base, token: fake.token },
+        pollMs: 10,
+      }).submit(fakeJulia(root, "s.jl", ""), { runsRoot: join(root, "runs") });
+      expect(await h.finished).toEqual({ status: "failed", exitCode: 1 });
+      expect(readFileSync(join(h.runDir, "FINISHED"), "utf8")).toBe('status = "failed"\nexit_code = 1\n');
+    } finally {
+      await fake.stop();
+    }
+  });
+});
diff --git a/packages/amico-run/test/fake_cloud.ts b/packages/amico-run/test/fake_cloud.ts
new file mode 100644
index 00000000..62e45964
--- /dev/null
+++ b/packages/amico-run/test/fake_cloud.ts
@@ -0,0 +1,91 @@
+// packages/amico-run/test/fake_cloud.ts
+// Hermetic Δ2/Δ4-shaped server (SHAPE ONLY — Δ4 is not deployed; this fake is
+// the executable contract the client is built against; revisit at the live
+// smoke, Task 11). One task at a time; mutate `state` mid-test to script the
+// run's lifecycle. Also imported (relative, Bundler-style) by the extension's
+// Δ9 tests — keep it vscode-free and dependency-free.
+import { createServer, type IncomingMessage, type Server, type ServerResponse } from "node:http";
+import type { AddressInfo } from "node:net";
+
+export interface FakeIter {
+  iter: number;
+  f: string;
+  inf_pr: string;
+  inf_du: string;
+}
+
+export interface FakeState {
+  task_status: "Pending" | "Running";
+  finished?: { status: "completed" | "failed" | "aborted" };
+  liveness: "alive" | "gone";
+  iters: FakeIter[]; // Δ4 stats: full history each poll (client dedups on high-water)
+  frame?: { iter: number; png_base64: string }; // Δ4 frames: newest only
+  framesBroken?: boolean; // 500 the frames endpoint — resolution (a) lane
+}
+
+export class FakeCloud {
+  readonly token = "test-token-abc";
+  readonly taskId = "task-0001";
+  state: FakeState = { task_status: "Pending", liveness: "alive", iters: [] };
+  submits: Array<{ auth: string | undefined; body: Record<string, unknown> }> = [];
+  submitStatus = 202; // override to 500 etc. for failure lanes (401 comes from a bad token)
+  aborts = 0;
+  statusPolls = 0;
+  base = "";
+  private server?: Server;
+
+  async start(): Promise<void> {
+    this.server = createServer((req, res) => void this.route(req, res));
+    await new Promise<void>((r) => this.server!.listen(0, "127.0.0.1", r));
+    this.base = `http://127.0.0.1:${(this.server!.address() as AddressInfo).port}`;
+  }
+
+  async stop(): Promise<void> {
+    if (!this.server) return;
+    const s = this.server;
+    this.server = undefined;
+    await new Promise<void>((r) => s.close(() => r()));
+  }
+
+  /** Deterministic sequencing: resolves once the client has status-polled ≥ n times. */
+  async waitForPolls(n: number): Promise<void> {
+    while (this.statusPolls < n) await new Promise((r) => setTimeout(r, 5));
+  }
+
+  private async route(req: IncomingMessage, res: ServerResponse): Promise<void> {
+    const chunks: Buffer[] = [];
+    for await (const c of req) chunks.push(c as Buffer);
+    const url = req.url ?? "";
+    const send = (code: number, body?: unknown): void => {
+      res.writeHead(code, { "content-type": "application/json" });
+      res.end(body === undefined ? "" : JSON.stringify(body));
+    };
+    const authed = req.headers.authorization === `Bearer ${this.token}`;
+    if (req.method === "POST" && url === "/solves") {
+      this.submits.push({
+        auth: req.headers.authorization,
+        body: JSON.parse(Buffer.concat(chunks).toString() || "{}") as Record<string, unknown>,
+      });
+      if (!authed) return send(401, { error: "bad credential" });
+      if (this.submitStatus !== 202) return send(this.submitStatus, { error: "boom" });
+      return send(202, { task_id: this.taskId, status: "Pending" });
+    }
+    if (!authed) return send(401, { error: "bad credential" });
+    if (url === `/solves/${this.taskId}/status`) {
+      this.statusPolls++;
+      const { task_status, finished, liveness } = this.state;
+      return send(200, finished ? { task_status, finished, liveness } : { task_status, liveness });
+    }
+    if (url === `/solves/${this.taskId}/stats`) return send(200, { iters: this.state.iters });
+    if (url === `/solves/${this.taskId}/frames`) {
+      if (this.state.framesBroken) return send(500, { error: "frames unavailable" });
+      if (!this.state.frame) return send(204);
+      return send(200, this.state.frame);
+    }
+    if (req.method === "POST" && url === `/solves/${this.taskId}/abort`) {
+      this.aborts++;
+      return send(202, { status: "aborting" });
+    }
+    return send(404, { error: `no route ${req.method} ${url}` });
+  }
+}
diff --git a/packages/amico-run/test/remote_config.test.ts b/packages/amico-run/test/remote_config.test.ts
new file mode 100644
index 00000000..42211680
--- /dev/null
+++ b/packages/amico-run/test/remote_config.test.ts
@@ -0,0 +1,59 @@
+// packages/amico-run/test/remote_config.test.ts
+import { describe, it, expect } from "vitest";
+import { mkdtempSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { cloudConfigFile, readRemoteConfig } from "../src/remote_config.js";
+import { ConfigError } from "../src/types.js";
+
+// Hermetic env: never let the runner's real AMICO_CLOUD_* leak in.
+const noEnv = {} as NodeJS.ProcessEnv;
+
+describe("remote_config — endpoint + credential resolution (authoring.ts idiom)", () => {
+  it("env pair wins over any file; trailing slash trimmed", () => {
+    const c = readRemoteConfig({
+      AMICO_CLOUD_URL: "https://x.example/",
+      AMICO_CLOUD_TOKEN: "t1",
+      AMICO_CLOUD_FILE: "/no/such/file.json",
+    } as NodeJS.ProcessEnv);
+    expect(c).toEqual({ baseUrl: "https://x.example", token: "t1" });
+  });
+  it("partial env pair → ConfigError naming both vars", () => {
+    expect(() => readRemoteConfig({ AMICO_CLOUD_URL: "https://x" } as NodeJS.ProcessEnv)).toThrow(ConfigError);
+    expect(() => readRemoteConfig({ AMICO_CLOUD_TOKEN: "t" } as NodeJS.ProcessEnv)).toThrow(/BOTH/);
+  });
+  it("falls back to $AMICO_CLOUD_FILE json", () => {
+    const f = join(mkdtempSync(join(tmpdir(), "cloud-")), "cloud.json");
+    writeFileSync(f, JSON.stringify({ base_url: "https://api.example/", token: "sekret" }));
+    expect(readRemoteConfig({ ...noEnv, AMICO_CLOUD_FILE: f })).toEqual({
+      baseUrl: "https://api.example",
+      token: "sekret",
+    });
+  });
+  it("missing file → ConfigError naming the path (exit-64 class: nothing ran)", () => {
+    const f = join(mkdtempSync(join(tmpdir(), "cloud-")), "absent.json");
+    expect(() => readRemoteConfig({ ...noEnv, AMICO_CLOUD_FILE: f })).toThrow(ConfigError);
+    expect(() => readRemoteConfig({ ...noEnv, AMICO_CLOUD_FILE: f })).toThrow(f);
+  });
+  it("wrong-shape file → ConfigError; the token value NEVER appears in the message", () => {
+    const f = join(mkdtempSync(join(tmpdir(), "cloud-")), "cloud.json");
+    writeFileSync(f, JSON.stringify({ base_url: 42, token: "sekret-token-value" }));
+    let msg = "";
+    try {
+      readRemoteConfig({ ...noEnv, AMICO_CLOUD_FILE: f });
+    } catch (e) {
+      expect(e).toBeInstanceOf(ConfigError);
+      msg = (e as Error).message;
+    }
+    expect(msg).toMatch(/base_url/);
+    expect(msg).not.toContain("sekret-token-value"); // llm_creds.mjs security stance
+  });
+  it("malformed JSON → ConfigError (never a JSON.parse throw)", () => {
+    const f = join(mkdtempSync(join(tmpdir(), "cloud-")), "cloud.json");
+    writeFileSync(f, "{nope");
+    expect(() => readRemoteConfig({ ...noEnv, AMICO_CLOUD_FILE: f })).toThrow(ConfigError);
+  });
+  it("default path is ~/.amico/cloud.json", () => {
+    expect(cloudConfigFile(noEnv)).toMatch(/\.amico\/cloud\.json$/);
+  });
+});
diff --git a/packages/amico-run/test/remote_executor.test.ts b/packages/amico-run/test/remote_executor.test.ts
new file mode 100644
index 00000000..2ee454b1
--- /dev/null
+++ b/packages/amico-run/test/remote_executor.test.ts
@@ -0,0 +1,263 @@
+// packages/amico-run/test/remote_executor.test.ts
+import { describe, it, expect } from "vitest";
+import { existsSync, readFileSync, readdirSync, statSync, utimesSync } from "node:fs";
+import { join } from "node:path";
+import { tmpRoot, fakeJulia, readToml } from "./helpers.js";
+import { FakeCloud } from "./fake_cloud.js";
+import { RemoteExecutor, EXIT_INFERRED, type RemoteExecutorOpts } from "../src/remote_executor.js";
+import { validateManifest, validateFinished } from "../src/schemas.js";
+import { ConfigError, type RunEvent } from "../src/types.js";
+
+async function withCloud(fn: (fake: FakeCloud) => Promise<void>): Promise<void> {
+  const fake = new FakeCloud();
+  await fake.start();
+  try {
+    await fn(fake);
+  } finally {
+    await fake.stop();
+  }
+}
+const ex = (fake: FakeCloud, knobs: Partial<RemoteExecutorOpts> = {}): RemoteExecutor =>
+  new RemoteExecutor({ config: { baseUrl: fake.base, token: fake.token }, pollMs: 10, ...knobs });
+
+async function collect(events: AsyncIterable<RunEvent>): Promise<RunEvent[]> {
+  const out: RunEvent[] = [];
+  for await (const e of events) out.push(e);
+  return out;
+}
+
+describe("RemoteExecutor.submit — Δ2 wire shape + local mirror", () => {
+  it("POSTs script CONTENT + filename with the Bearer credential; 202 → conforming mirror run dir", async () => {
+    await withCloud(async (fake) => {
+      fake.state.finished = { status: "completed" }; // settle on first poll
+      const root = tmpRoot();
+      const script = fakeJulia(root, "solve.jl", "// julia body");
+      const h = await ex(fake).submit(script, { lab: "testlab", runsRoot: join(root, "runs") });
+      await h.finished;
+      // Δ2 wire shape
+      expect(fake.submits).toHaveLength(1);
+      expect(fake.submits[0].auth).toBe(`Bearer ${fake.token}`);
+      expect(String(fake.submits[0].body.script)).toContain("// julia body"); // content, not a path
+      expect(fake.submits[0].body.filename).toBe("solve.jl");
+      // mirror run dir — same contract files LocalExecutor writes
+      const manifest = readToml(join(h.runDir, "run.toml"));
+      expect(validateManifest(manifest).ok).toBe(true);
+      expect(manifest.lab_id).toBe("testlab");
+      expect((manifest.julia as { binary: string }).binary).toBe("cloud");
+      // sidecar (run.toml is additionalProperties:false — frozen), index, latest, run.log from t0
+      expect(JSON.parse(readFileSync(join(h.runDir, "remote.json"), "utf8"))).toEqual({
+        task_id: fake.taskId,
+        base_url: fake.base,
+      });
+      expect(readFileSync(join(root, "runs", "index"), "utf8")).toContain(h.runId);
+      expect(existsSync(join(root, "runs", "latest"))).toBe(true);
+      expect(existsSync(join(h.runDir, "run.log"))).toBe(true);
+    });
+  });
+
+  it("401 → ConfigError (config-class credential fault), NO run dir, NO index", async () => {
+    await withCloud(async (fake) => {
+      const root = tmpRoot();
+      const script = fakeJulia(root, "s.jl", "");
+      const bad = new RemoteExecutor({ config: { baseUrl: fake.base, token: "wrong" }, pollMs: 10 });
+      await expect(bad.submit(script, { runsRoot: join(root, "runs") })).rejects.toThrow(ConfigError);
+      expect(existsSync(join(root, "runs"))).toBe(false); // step-1 parity: nothing ran, no run dir
+    });
+  });
+
+  it("non-202 (500) → plain Error (not config-class), NO run dir", async () => {
+    await withCloud(async (fake) => {
+      fake.submitStatus = 500;
+      const root = tmpRoot();
+      const script = fakeJulia(root, "s.jl", "");
+      const p = ex(fake).submit(script, { runsRoot: join(root, "runs") });
+      await expect(p).rejects.toThrow(/HTTP 500/);
+      await expect(p).rejects.not.toThrow(ConfigError);
+      expect(existsSync(join(root, "runs"))).toBe(false);
+    });
+  });
+
+  it("missing script → ConfigError BEFORE any network call", async () => {
+    await withCloud(async (fake) => {
+      const root = tmpRoot();
+      await expect(ex(fake).submit(join(root, "nope.jl"), { runsRoot: join(root, "runs") })).rejects.toThrow(
+        /script not found/,
+      );
+      expect(fake.submits).toHaveLength(0);
+    });
+  });
+});
+
+describe("terminal resolution (d) — status poll is authoritative, FINISHED mirrored locally", () => {
+  for (const [status, exitCode] of [
+    ["completed", 0],
+    ["failed", 1],
+    ["aborted", 130],
+  ] as const) {
+    it(`finished{${status}} → FINISHED{${status}, ${exitCode}}, events terminate ON the finished event`, async () => {
+      await withCloud(async (fake) => {
+        fake.state.finished = { status };
+        const root = tmpRoot();
+        const h = await ex(fake).submit(fakeJulia(root, "s.jl", ""), { runsRoot: join(root, "runs") });
+        const evs = await collect(h.events);
+        expect(evs.at(-1)).toEqual({ kind: "finished", status, exitCode });
+        expect(await h.finished).toEqual({ status, exitCode });
+        const fin = readToml(join(h.runDir, "FINISHED"));
+        expect(validateFinished(fin).ok).toBe(true);
+        expect(fin).toEqual({ status, exit_code: exitCode });
+      });
+    });
+  }
+});
+
+describe("poll streaming — stats/frames fed into the mirror (Δ4)", () => {
+  it("stats become AMICODE_ITER lines in run.log AND iter events; re-served history is deduped", async () => {
+    await withCloud(async (fake) => {
+      fake.state.task_status = "Running";
+      fake.state.iters = [
+        { iter: 1, f: "1.0e-2", inf_pr: "1e-8", inf_du: "1e-6" },
+        { iter: 2, f: "3.0e-4", inf_pr: "1e-9", inf_du: "1e-7" },
+      ];
+      const root = tmpRoot();
+      const h = await ex(fake).submit(fakeJulia(root, "s.jl", ""), { runsRoot: join(root, "runs") });
+      await fake.waitForPolls(4); // several polls over the SAME stats — dedup must hold
+      fake.state.finished = { status: "completed" };
+      const evs = await collect(h.events);
+      const iters = evs.filter((e) => e.kind === "iter");
+      expect(iters).toHaveLength(2); // not 2 × polls
+      expect(iters[0]).toMatchObject({ fields: { iter: "1", f: "1.0e-2", inf_pr: "1e-8", inf_du: "1e-6" } });
+      expect(evs.at(-1)!.kind).toBe("finished");
+      // the mirror's run.log carries the exact synthesized lines (tail/backstop food)
+      const log = readFileSync(join(h.runDir, "run.log"), "utf8");
+      expect(log).toContain("AMICODE_ITER iter=1 f=1.0e-2 inf_pr=1e-8 inf_du=1e-6");
+      expect(log.match(/AMICODE_ITER iter=1 /g)).toHaveLength(1);
+    });
+  });
+
+  it("frames land as iter_NNN.png (S3 layout name); newest-wins high-water", async () => {
+    await withCloud(async (fake) => {
+      fake.state.task_status = "Running";
+      fake.state.frame = { iter: 7, png_base64: Buffer.from("png-bytes-7").toString("base64") };
+      const root = tmpRoot();
+      const h = await ex(fake).submit(fakeJulia(root, "s.jl", ""), { runsRoot: join(root, "runs") });
+      await fake.waitForPolls(2);
+      fake.state.finished = { status: "completed" };
+      await h.finished;
+      expect(readFileSync(join(h.runDir, "iter_007.png"), "utf8")).toBe("png-bytes-7");
+      expect(readdirSync(h.runDir).filter((f) => f.endsWith(".tmp"))).toHaveLength(0); // atomic write
+    });
+  });
+
+  it("resolution (a): a 500ing frames endpoint changes NOTHING — run completes, no png, no error event", async () => {
+    await withCloud(async (fake) => {
+      fake.state.task_status = "Running";
+      fake.state.framesBroken = true;
+      fake.state.finished = { status: "completed" };
+      const root = tmpRoot();
+      const h = await ex(fake).submit(fakeJulia(root, "s.jl", ""), { runsRoot: join(root, "runs") });
+      expect(await h.finished).toEqual({ status: "completed", exitCode: 0 });
+      expect(readdirSync(h.runDir).filter((f) => f.endsWith(".png"))).toHaveLength(0);
+    });
+  });
+});
+
+describe("liveness lanes — resolutions (c) and (d)", () => {
+  it("heartbeat: a successful status poll re-touches run.log's mtime (cloud liveness → disk signal)", async () => {
+    await withCloud(async (fake) => {
+      const root = tmpRoot(); // Pending, alive, no iters: pure warming
+      const h = await ex(fake).submit(fakeJulia(root, "s.jl", ""), { runsRoot: join(root, "runs") });
+      await fake.waitForPolls(2);
+      const cold = new Date(Date.now() - 11 * 60 * 1000); // age past the inspector's 10-min knob
+      utimesSync(join(h.runDir, "run.log"), cold, cold);
+      const seen = fake.statusPolls;
+      await fake.waitForPolls(seen + 2); // ≥1 full poll after aging
+      expect(Date.now() - statSync(join(h.runDir, "run.log")).mtimeMs).toBeLessThan(60_000);
+      fake.state.finished = { status: "completed" };
+      await h.finished;
+    });
+  });
+
+  it("resolution (d): liveness=gone without FINISHED → inferred terminal failed/255 + breadcrumb", async () => {
+    await withCloud(async (fake) => {
+      fake.state.task_status = "Running";
+      fake.state.liveness = "gone";
+      const root = tmpRoot();
+      const h = await ex(fake).submit(fakeJulia(root, "s.jl", ""), { runsRoot: join(root, "runs") });
+      expect(await h.finished).toEqual({ status: "failed", exitCode: EXIT_INFERRED });
+      const fin = readToml(join(h.runDir, "FINISHED"));
+      expect(fin).toEqual({ status: "failed", exit_code: EXIT_INFERRED });
+      expect(readFileSync(join(h.runDir, "run.log"), "utf8")).toContain(
+        "AMICODE_REMOTE_LOST instance gone without FINISHED",
+      );
+    });
+  });
+
+  it("resolution (c): warming budget exhausted (Pending forever) → inferred terminal + best-effort abort", async () => {
+    await withCloud(async (fake) => {
+      const root = tmpRoot(); // stays Pending with no iters — never shows life
+      const h = await ex(fake, { warmingBudgetMs: 150 }).submit(fakeJulia(root, "s.jl", ""), {
+        runsRoot: join(root, "runs"),
+      });
+      expect(await h.finished).toEqual({ status: "failed", exitCode: EXIT_INFERRED });
+      expect(readFileSync(join(h.runDir, "run.log"), "utf8")).toContain("warming budget exhausted");
+      // best-effort abort fired is asserted in Task 6 (postAbort is a no-op until then)
+    });
+  });
+
+  it("resolution (d) client half: endpoint gone after life was seen → bounded inferred terminal", async () => {
+    const fake = new FakeCloud();
+    await fake.start();
+    fake.state.task_status = "Running";
+    fake.state.iters = [{ iter: 1, f: "1e-2", inf_pr: "1e-8", inf_du: "1e-6" }];
+    const root = tmpRoot();
+    const h = await ex(fake, { lostAfterMs: 200 }).submit(fakeJulia(root, "s.jl", ""), {
+      runsRoot: join(root, "runs"),
+    });
+    await fake.waitForPolls(2); // life seen — the warming budget is out of play
+    await fake.stop(); // the endpoint disappears
+    expect(await h.finished).toEqual({ status: "failed", exitCode: EXIT_INFERRED }); // bounded, no forever-pump
+    expect(readFileSync(join(h.runDir, "run.log"), "utf8")).toContain("poll endpoint unreachable");
+  });
+});
+
+describe("abort() — resolution (b): a REQUEST; the run is live until the real terminal", () => {
+  it("posts …/abort once (idempotent), keeps streaming, settles ONLY when the poll reports aborted", async () => {
+    await withCloud(async (fake) => {
+      fake.state.task_status = "Running";
+      const root = tmpRoot();
+      const h = await ex(fake).submit(fakeJulia(root, "s.jl", ""), { runsRoot: join(root, "runs") });
+      await fake.waitForPolls(1);
+      const aborting = h.abort(); // request…
+      void h.abort(); // …idempotent: no second POST
+      await fake.waitForPolls(fake.statusPolls + 2); // the pump is STILL polling post-abort
+      expect(fake.aborts).toBe(1);
+      expect(existsSync(join(h.runDir, "FINISHED"))).toBe(false); // not terminal yet — request ≠ kill
+      // the run keeps streaming after the abort request (still live)
+      fake.state.iters = [{ iter: 3, f: "5e-3", inf_pr: "1e-8", inf_du: "1e-6" }];
+      const seen = fake.statusPolls;
+      await fake.waitForPolls(seen + 2);
+      expect(readFileSync(join(h.runDir, "run.log"), "utf8")).toContain("AMICODE_ITER iter=3");
+      // the cloud finally reports the terminal — NOW everything settles
+      fake.state.finished = { status: "aborted" };
+      await aborting; // abort() resolves with the terminal, like LocalExecutor's
+      expect(await h.finished).toEqual({ status: "aborted", exitCode: 130 });
+      expect(readToml(join(h.runDir, "FINISHED"))).toEqual({ status: "aborted", exit_code: 130 });
+    });
+  });
+
+  it("warming-budget exhaustion fires the best-effort abort request (Task 5 leftover)", async () => {
+    await withCloud(async (fake) => {
+      const root = tmpRoot();
+      const h = await ex(fake, { warmingBudgetMs: 120 }).submit(fakeJulia(root, "s.jl", ""), {
+        runsRoot: join(root, "runs"),
+      });
+      await h.finished;
+      // the abort POST is fire-and-forget (`void postAbort()` — settle must not
+      // depend on it), so the in-flight request can land AFTER finished resolves;
+      // spin until it arrives (the FakeCloud.waitForPolls idiom; vitest's 5s
+      // timeout bounds a never-fired request as a failure).
+      while (fake.aborts < 1) await new Promise((r) => setTimeout(r, 5));
+      expect(fake.aborts).toBe(1);
+    });
+  });
+});
diff --git a/packages/amico-run/test/s31.test.ts b/packages/amico-run/test/s31.test.ts
index 89f6a08b..38d1998c 100644
--- a/packages/amico-run/test/s31.test.ts
+++ b/packages/amico-run/test/s31.test.ts
@@ -9,10 +9,18 @@ import { join } from "node:path";
 // NOT a physics knob; all physics stays in the script.)
 const FORBIDDEN = [/--gate\b/, /--system\b/, /--pulse\b/, /modelcontextprotocol/i, /node:https?\b/, /\bfetch\s*\(/];
 
+// Δ8 (amicode#32, spec-20260628 cloud-solve-service): RemoteExecutor is the
+// ONE sanctioned network edge in amico-run — submit→poll over the Δ2/Δ4 API
+// is its entire job. The S31 ban (no ambient HTTP in the orchestrator) stays
+// for everything else; this exemption is scoped to the single module whose
+// ratified contract IS HTTP. Same lift mechanism as the spec-C SolveSpec ban.
+const EXEMPT = new Set(["remote_executor.ts"]);
+
 describe("S31 grep rule", () => {
   it("src/ contains no forbidden tool-layer patterns", () => {
     const srcDir = join(__dirname, "..", "src");
     for (const f of readdirSync(srcDir)) {
+      if (EXEMPT.has(f)) continue;
       const text = readFileSync(join(srcDir, f), "utf8");
       for (const re of FORBIDDEN) {
         expect(text, `${f} matches forbidden ${re}`).not.toMatch(re);
diff --git a/packages/amico-run/test/slow/remote_live_smoke.test.ts b/packages/amico-run/test/slow/remote_live_smoke.test.ts
new file mode 100644
index 00000000..187dac99
--- /dev/null
+++ b/packages/amico-run/test/slow/remote_live_smoke.test.ts
@@ -0,0 +1,34 @@
+// packages/amico-run/test/slow/remote_live_smoke.test.ts
+// ⛔ SKIP-UNTIL-DEPLOYED — the ONLY non-hermetic Δ8 test. Gate:
+//   aws-infra#166 + aws-infra#167 merged, Δ4 deployed on STAGING, then run:
+//   AMICO_CLOUD_SMOKE=1 AMICO_CLOUD_URL=<staging> AMICO_CLOUD_TOKEN=<cred> \
+//     pnpm exec vitest run test/slow/remote_live_smoke.test.ts
+// Purpose: validate the SHAPE-ONLY assumptions the fakes encode (frames
+// response encoding, liveness field values, abort route) against the real
+// service. Any mismatch → fix FakeCloud first, then the client, keeping the
+// hermetic suite the source of truth.
+import { describe, it, expect } from "vitest";
+import { existsSync, mkdtempSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { RemoteExecutor } from "../../src/remote_executor.js";
+
+const gated =
+  process.env.AMICO_CLOUD_SMOKE === "1" && !!process.env.AMICO_CLOUD_URL && !!process.env.AMICO_CLOUD_TOKEN;
+
+describe.skipIf(!gated)("remote live smoke — STAGING", () => {
+  it(
+    "submits a trivial script, polls to a REAL terminal, mirror conforms",
+    async () => {
+      const root = mkdtempSync(join(tmpdir(), "smoke-"));
+      const script = join(root, "smoke.jl");
+      writeFileSync(script, 'println("AMICODE_ITER iter=1 f=1.0e-3 inf_pr=1e-9 inf_du=1e-7")\n');
+      const h = await new RemoteExecutor({ pollMs: 5000 }).submit(script, { runsRoot: join(root, "runs") });
+      const fin = await h.finished; // staging wall time: minutes (cold EC2), not ms
+      expect(["completed", "failed"]).toContain(fin.status); // a real terminal — never a hang
+      expect(existsSync(join(h.runDir, "FINISHED"))).toBe(true);
+      expect(existsSync(join(h.runDir, "remote.json"))).toBe(true);
+    },
+    20 * 60 * 1000, // 20-min vitest timeout: covers the 15-min warming budget
+  );
+});
diff --git a/packages/extension/test/executor_branches.test.ts b/packages/extension/test/executor_branches.test.ts
new file mode 100644
index 00000000..d503a32e
--- /dev/null
+++ b/packages/extension/test/executor_branches.test.ts
@@ -0,0 +1,27 @@
+// packages/extension/test/executor_branches.test.ts
+// S12 structural pin (Track C locked decisions, Δ8 AC): downstream consumes
+// ONLY the RunHandle / run-dir contract — no executor-type branches anywhere
+// in the runs/inspector pipeline. Comments stripped: prose may name them.
+import { describe, it, expect } from "vitest";
+import { readFileSync } from "node:fs";
+import { fileURLToPath } from "node:url";
+
+const FILES = [
+  "runs_manager.ts",
+  "run_registry.ts",
+  "run_inspector.ts",
+  "run_dir_reader.ts",
+  "run_controls.ts",
+  "status_bar.ts",
+];
+
+describe("S12 — no executor-type branches in the runs/inspector pipeline", () => {
+  for (const f of FILES) {
+    it(`${f} names no executor type in code`, () => {
+      const src = readFileSync(fileURLToPath(new URL(`../src/${f}`, import.meta.url)), "utf8")
+        .replace(/\/\/.*$/gm, "")
+        .replace(/\/\*[\s\S]*?\*\//g, "");
+      expect(src).not.toMatch(/LocalExecutor|RemoteExecutor|instanceof\s+\w*Executor/);
+    });
+  }
+});
diff --git a/packages/extension/test/remote_statemachine.test.ts b/packages/extension/test/remote_statemachine.test.ts
new file mode 100644
index 00000000..0eabfcf6
--- /dev/null
+++ b/packages/extension/test/remote_statemachine.test.ts
@@ -0,0 +1,110 @@
+// packages/extension/test/remote_statemachine.test.ts
+// Δ9 (#33): a REMOTE run consumed through the SAME state machine as local —
+// index-tail discovery, warming, run.log tail via the poll backstop, and
+// FINISHED-keyed completion. Asserts the same inspector calls as the local
+// flow in runs_manager.test.ts:90-112. ZERO extension production code changed.
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { existsSync, mkdtempSync, readFileSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+const { inspector } = vi.hoisted(() => ({
+  inspector: {
+    setWarmingUp: vi.fn(),
+    postTiming: vi.fn(),
+    postCompletion: vi.fn(),
+    postIterationRecord: vi.fn(),
+    postPulse: vi.fn(),
+    setRunLabel: vi.fn(),
+    activate: vi.fn(),
+    reveal: vi.fn(),
+  },
+}));
+vi.mock("../src/run_inspector", () => ({ getInspector: () => inspector }));
+
+import { RunsManager } from "../src/runs_manager";
+import { RemoteExecutor, Scheduler } from "@amicode/amico-run";
+import { FakeCloud } from "../../amico-run/test/fake_cloud";
+
+const channel = { appendLine() {}, append() {} } as never;
+const tick = (m: RunsManager): void => (m as unknown as { tick(): void }).tick();
+const until = async (pred: () => boolean, ms = 3000): Promise<void> => {
+  const t0 = Date.now();
+  while (!pred()) {
+    if (Date.now() - t0 > ms) throw new Error("condition not reached in time");
+    await new Promise((r) => setTimeout(r, 10));
+  }
+};
+
+describe("Δ9 — remote run through the SAME inspector state machine", () => {
+  beforeEach(() => {
+    for (const f of Object.values(inspector)) f.mockClear();
+  });
+
+  it("warming → poll-delivered iters/frames → completion: same inspector calls as a local run", async () => {
+    const fake = new FakeCloud(); // Pending, alive, no iters: warming
+    await fake.start();
+    const root = mkdtempSync(join(tmpdir(), "runs-"));
+    const m = new RunsManager({ runsRoot: root, channel });
+    m.start();
+    const ex = new RemoteExecutor({ config: { baseUrl: fake.base, token: fake.token }, pollMs: 10 });
+    try {
+      const script = join(root, "s.jl");
+      writeFileSync(script, "// content posted to the cloud\n");
+      const h = await ex.submit(script, { runsRoot: root }); // mirror dir + index line land NOW
+      tick(m); // poll backstop drains the index → discovery (runs_manager.ts:170)
+
+      // 1. warming — identical to the local fresh-run lane (runs_manager.test.ts:98-101)
+      expect(inspector.setWarmingUp).toHaveBeenCalledWith(h.runId);
+      expect(inspector.setRunLabel).toHaveBeenCalledWith(h.runId, h.runId);
+      expect(inspector.activate).toHaveBeenCalledWith(h.runId);
+      expect(m.selectedRun).toBe(h.runId);
+
+      // 2. poll-delivered iter + frame (Δ4 → mirror → run.log tail via the backstop)
+      fake.state.task_status = "Running";
+      fake.state.iters = [{ iter: 7, f: "1.0e-2", inf_pr: "1e-8", inf_du: "1e-6" }];
+      fake.state.frame = { iter: 7, png_base64: Buffer.from("png-bytes").toString("base64") };
+      await until(() => readFileSync(join(h.runDir, "run.log"), "utf8").includes("iter=7"));
+      tick(m); // backstop re-pokes the tail — the "no new paradigm" hinge
+      expect(inspector.postIterationRecord).toHaveBeenCalledWith(h.runId, expect.objectContaining({ iter: 7 }));
+      await until(() => existsSync(join(h.runDir, "iter_007.png"))); // frames mirrored best-effort
+
+      // 3. completion — FINISHED authoritative via the status poll (resolution (d))
+      fake.state.finished = { status: "completed" };
+      await h.finished;
+      tick(m); // same idempotent FINISHED re-check as local (checkFinished)
+      expect(inspector.postCompletion).toHaveBeenCalledWith(h.runId, "completed", undefined);
+      // fidelity undefined: result.toml mirroring is the NAMED Δ4 seam (status
+      // shape doesn't carry it yet) — the CALL SHAPE is identical to local.
+    } finally {
+      m.dispose();
+      await fake.stop();
+    }
+  });
+
+  it("scheduler lane: remote run via Scheduler.enqueue registers on `started` and completes (S12 passthrough)", async () => {
+    const fake = new FakeCloud();
+    await fake.start();
+    fake.state = { task_status: "Running", liveness: "alive", iters: [], finished: { status: "completed" } };
+    const root = mkdtempSync(join(tmpdir(), "runs-"));
+    const m = new RunsManager({ runsRoot: root, channel });
+    m.start();
+    try {
+      const s = new Scheduler(
+        new RemoteExecutor({ config: { baseUrl: fake.base, token: fake.token }, pollMs: 10 }),
+      );
+      m.attachScheduler(s);
+      const script = join(root, "s.jl");
+      writeFileSync(script, "//\n");
+      const r = s.enqueue({ scriptPath: script, opts: { runsRoot: root } });
+      const h = await r.handle; // the RunHandle IS the executor's (scheduler.test.ts:78)
+      expect(m.selectedRun).toBe(h.runId); // `started` registered it — no index wait
+      await h.finished;
+      tick(m);
+      expect(inspector.postCompletion).toHaveBeenCalledWith(h.runId, "completed", undefined);
+    } finally {
+      m.dispose();
+      await fake.stop();
+    }
+  });
+});
diff --git a/packages/extension/test/remote_warming.test.ts b/packages/extension/test/remote_warming.test.ts
new file mode 100644
index 00000000..046549c9
--- /dev/null
+++ b/packages/extension/test/remote_warming.test.ts
@@ -0,0 +1,64 @@
+// packages/extension/test/remote_warming.test.ts
+// Δ9 / resolution (c): remote warming must NOT trip the local 10-min stall
+// machinery — the executor's heartbeat mirrors cloud liveness into run.log's
+// mtime, so stopPlan/liveStatus measure CLOUD silence. An actual poll outage
+// then reads honestly (stalled → bounded inferred terminal).
+import { describe, it, expect } from "vitest";
+import { mkdtempSync, readFileSync, statSync, utimesSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { RemoteExecutor, EXIT_INFERRED } from "@amicode/amico-run";
+import { FakeCloud } from "../../amico-run/test/fake_cloud";
+import { stopPlan, STALL_AFTER_MS } from "../src/run_controls";
+
+describe("Δ9 — remote warming budget is the executor's, not the inspector's", () => {
+  it("warming past the LOCAL stall threshold stays 'cooperative': the heartbeat keeps run.log's mtime live", async () => {
+    const fake = new FakeCloud(); // Pending, alive, no iters — pure warming
+    await fake.start();
+    try {
+      const root = mkdtempSync(join(tmpdir(), "runs-"));
+      const script = join(root, "s.jl");
+      writeFileSync(script, "//\n");
+      const ex = new RemoteExecutor({ config: { baseUrl: fake.base, token: fake.token }, pollMs: 10 });
+      const h = await ex.submit(script, { runsRoot: root });
+      await fake.waitForPolls(2);
+      // simulate a long warming on the local clock: age the log past the knob…
+      const cold = new Date(Date.now() - STALL_AFTER_MS - 60_000);
+      utimesSync(join(h.runDir, "run.log"), cold, cold);
+      const seen = fake.statusPolls;
+      await fake.waitForPolls(seen + 2); // …then let ≥1 full successful poll land
+      // the heartbeat re-touched mtime: inspector-side logic reads "alive"
+      expect(Date.now() - statSync(join(h.runDir, "run.log")).mtimeMs).toBeLessThan(STALL_AFTER_MS);
+      expect(stopPlan(h.runDir)).toBe("cooperative"); // NOT "force" — no zombie verdict for a warming remote run
+      fake.state.finished = { status: "completed" }; // settle so no pump leaks past the test
+      await h.finished;
+    } finally {
+      await fake.stop();
+    }
+  });
+
+  it("a sustained poll OUTAGE reads honestly: no heartbeat, then a bounded inferred terminal", async () => {
+    const fake = new FakeCloud();
+    await fake.start();
+    fake.state = {
+      task_status: "Running",
+      liveness: "alive",
+      iters: [{ iter: 1, f: "1e-2", inf_pr: "1e-8", inf_du: "1e-6" }],
+    };
+    const root = mkdtempSync(join(tmpdir(), "runs-"));
+    const script = join(root, "s.jl");
+    writeFileSync(script, "//\n");
+    const ex = new RemoteExecutor({
+      config: { baseUrl: fake.base, token: fake.token },
+      pollMs: 10,
+      lostAfterMs: 200, // test knob — the 10-min default in fast-forward
+    });
+    const h = await ex.submit(script, { runsRoot: root });
+    await fake.waitForPolls(2); // life seen — warming budget out of play
+    await fake.stop(); // the endpoint disappears
+    const fin = await h.finished; // bounded by lostAfterMs (resolution (d) client half)
+    expect(fin).toEqual({ status: "failed", exitCode: EXIT_INFERRED });
+    expect(readFileSync(join(h.runDir, "run.log"), "utf8")).toContain("AMICODE_REMOTE_LOST poll endpoint unreachable");
+    expect(stopPlan(h.runDir)).toBe("already-finished"); // FINISHED written — Stop converges, never wedges
+  });
+});