diff --git a/.github/workflows/frontstage-pages.yml b/.github/workflows/frontstage-pages.yml
index 8aa34405e0..80e1783366 100644
--- a/.github/workflows/frontstage-pages.yml
+++ b/.github/workflows/frontstage-pages.yml
@@ -48,6 +48,7 @@ on:
       - "scripts/render-star-history.py"
       - "scripts/fetch-github-stargazers.sh"
       - "scripts/install-from-github.sh"
+      - "loopx/cli_commands/update_artifact.py"
   push:
     branches:
       - main
@@ -91,6 +92,7 @@ on:
       - "scripts/render-star-history.py"
       - "scripts/fetch-github-stargazers.sh"
       - "scripts/install-from-github.sh"
+      - "loopx/cli_commands/update_artifact.py"
 
 permissions:
   actions: read
@@ -193,6 +195,14 @@ jobs:
           LOOPX_SITE_RESOLVE_DESKTOP_RELEASE: "1"
         run: npm run export:frontstage-share -- --base /loopx/ --out-dir ../../../output/frontstage-pages
 
+      - name: Require stable runtime before publishing its bootstrap
+        if: github.event_name != 'pull_request'
+        run: |
+          python - <<'PY'
+          from loopx.cli_commands.update_artifact import resolve_artifact
+          resolve_artifact("loopx-project/loopx", "stable")
+          PY
+
       - name: Fetch complete GitHub stargazer history
         if: github.event_name != 'pull_request'
         env:
diff --git a/.github/workflows/release-artifacts.yml b/.github/workflows/release-artifacts.yml
index 57ba65102a..a97c0738cb 100644
--- a/.github/workflows/release-artifacts.yml
+++ b/.github/workflows/release-artifacts.yml
@@ -16,6 +16,8 @@ on:
       - "scripts/verify_installed_chat.py"
       - "README.md"
       - "scripts/release_artifacts.py"
+      - "scripts/runtime_artifact.py"
+      - "scripts/qualify_runtime_artifact.py"
       - "tsconfig.control-plane.json"
   release:
     types: [published]
@@ -390,6 +392,11 @@ jobs:
           identity.write_text(json.dumps(record, indent=2) + "\n")
           PY
 
+      - name: Build and qualify the stable snapshot runtime
+        run: |
+          python scripts/runtime_artifact.py --channel stable
+          python scripts/qualify_runtime_artifact.py dist/runtime > dist/runtime/qualification.json
+
       - name: Attest release packages and checksum manifest
         if: github.event_name != 'pull_request'
         uses: actions/attest@v4
@@ -398,6 +405,7 @@ jobs:
             dist/packages/*
             dist/SHA256SUMS
             dist/RELEASE-SOURCES.json
+            dist/runtime/*
 
       - name: Upload validated release bundle
         uses: actions/upload-artifact@v7
@@ -443,6 +451,7 @@ jobs:
             dist/packages/* \
             dist/SHA256SUMS \
             dist/RELEASE-SOURCES.json \
+            dist/runtime/* \
             --repo "${GITHUB_REPOSITORY}"
 
   publish-pypi:
diff --git a/.github/workflows/runtime-snapshots.yml b/.github/workflows/runtime-snapshots.yml
new file mode 100644
index 0000000000..c0c462411c
--- /dev/null
+++ b/.github/workflows/runtime-snapshots.yml
@@ -0,0 +1,119 @@
+name: Runtime snapshots
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    paths:
+      - ".github/workflows/runtime-snapshots.yml"
+      - "scripts/runtime_artifact.py"
+      - "scripts/qualify_runtime_artifact.py"
+      - "scripts/install*.sh"
+      - "loopx/cli_commands/update*.py"
+      - "loopx/cli_commands/support_control_update*.py"
+      - "loopx/self_update*.py"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: runtime-snapshot-${{ github.event_name == 'pull_request' && github.ref || 'main' }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+
+jobs:
+  build:
+    if: github.repository == 'loopx-project/loopx'
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    permissions:
+      contents: read
+      id-token: write
+      attestations: write
+    steps:
+      - uses: actions/checkout@v7
+        with:
+          fetch-depth: 0
+      - uses: actions/setup-python@v6
+        with:
+          python-version: "3.11"
+      - uses: actions/setup-node@v6
+        with:
+          node-version: "24"
+      - name: Prepare test environment
+        run: python -m pip install -e '.[test]'
+      - name: Qualify update transport and ownership
+        run: python -m pytest -q tests/test_update_artifact.py tests/test_update_operation.py tests/test_self_update_runtime_activation.py tests/test_archive_installer_commit_response.py
+      - name: Build qualified frontend
+        run: python scripts/chat_bundle.py ensure
+      - name: Build immutable runtime and resource inventory
+        run: python scripts/runtime_artifact.py --channel main
+      - name: Qualify clean installation and upgrade
+        run: python scripts/qualify_runtime_artifact.py dist/runtime > dist/runtime/qualification.json
+      - name: Freeze revision-addressed metadata
+        shell: bash
+        run: |
+          set -euo pipefail
+          revision="$(python -c 'import json; print(json.load(open("dist/runtime/loopx-runtime.json"))["source_revision"])')"
+          cp dist/runtime/loopx-runtime.json "dist/runtime/loopx-runtime-${revision}.json"
+          mv dist/runtime/runtime-inventory.json "dist/runtime/runtime-inventory-${revision}.json"
+          mv dist/runtime/qualification.json "dist/runtime/qualification-${revision}.json"
+      - name: Attest exact runtime bytes and inventory
+        if: github.event_name != 'pull_request'
+        uses: actions/attest@v4
+        with:
+          subject-path: dist/runtime/*
+      - uses: actions/upload-artifact@v7
+        with:
+          name: loopx-runtime
+          path: dist/runtime
+          if-no-files-found: error
+
+  publish:
+    if: github.event_name != 'pull_request' && github.ref == 'refs/heads/main'
+    needs: build
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    permissions:
+      contents: write
+    steps:
+      - uses: actions/download-artifact@v7
+        with:
+          name: loopx-runtime
+          path: runtime
+      - name: Publish immutable snapshot then channel pointer
+        env:
+          GH_TOKEN: ${{ github.token }}
+        shell: bash
+        run: |
+          set -euo pipefail
+          revision="$(python -c 'import json; print(json.load(open("runtime/loopx-runtime.json"))["source_revision"])')"
+          # One dedicated channel release keeps developer snapshots out of
+          # product release history and bounded stable-predecessor discovery.
+          tag=runtime-main
+          if ! gh release view "$tag" --repo "$GITHUB_REPOSITORY" >/dev/null 2>&1; then
+            gh release create "$tag" --repo "$GITHUB_REPOSITORY" --target "$revision" --prerelease --title "Latest qualified main runtime" --notes "Mutable channel metadata; revision-addressed archives and manifests are immutable. Stable users continue with PyPI or the stable runtime."
+          fi
+          # A retry reuses immutable bytes. Never clobber an immutable asset.
+          mkdir -p existing
+          for file in runtime/*; do
+            name="$(basename "$file")"
+            if [[ "$name" == loopx-runtime.json ]]; then continue; fi
+            if gh release download "$tag" --repo "$GITHUB_REPOSITORY" --pattern "$name" --dir existing >/dev/null 2>&1; then
+              cmp "$file" "existing/$name"
+            else
+              gh release upload "$tag" "$file" --repo "$GITHUB_REPOSITORY"
+            fi
+          done
+          # Publication is serialized. A qualified ancestor is usable while a
+          # newer main is still building; never publish a non-main revision.
+          relation="$(gh api "repos/${GITHUB_REPOSITORY}/compare/${revision}...main" --jq .status)"
+          if [[ "$relation" != ahead && "$relation" != identical ]]; then exit 1; fi
+          # Rerunning an older successful workflow may retain its immutable
+          # assets, but must not move the qualified main channel backwards.
+          current="$(gh api "repos/${GITHUB_REPOSITORY}/commits/runtime-main" --jq .sha)"
+          promotion="$(gh api "repos/${GITHUB_REPOSITORY}/compare/${current}...${revision}" --jq .status)"
+          if [[ "$promotion" == behind ]]; then exit 0; fi
+          if [[ "$promotion" != ahead && "$promotion" != identical ]]; then exit 1; fi
+          gh api --method PATCH "repos/${GITHUB_REPOSITORY}/git/refs/tags/runtime-main" -f sha="$revision" -F force=false >/dev/null
+          gh release upload runtime-main runtime/loopx-runtime.json --repo "$GITHUB_REPOSITORY" --clobber
diff --git a/apps/presentation/dashboard/src/features/personal-workspace/desktop-update.tsx b/apps/presentation/dashboard/src/features/personal-workspace/desktop-update.tsx
index d4ad70327a..032f35af33 100644
--- a/apps/presentation/dashboard/src/features/personal-workspace/desktop-update.tsx
+++ b/apps/presentation/dashboard/src/features/personal-workspace/desktop-update.tsx
@@ -1,6 +1,7 @@
 import { ChevronUp, Download, RefreshCw, X } from "lucide-react";
 import { useEffect, useRef, useState } from "react";
 import { useWorkspaceI18n } from "./i18n";
+import { observeTransfer, transferText, type TransferSample } from "./update-progress";
 
 type Phase = "idle" | "runtime_required" | "runtime_pairing_required" | "connecting" | "service_error" | "checking" | "available" | "up_to_date" | "downloading" | "installing_app" | "installing_runtime" | "restart_required" | "ready" | "error";
 type UpdateState = { phase: Phase; details?: { version?: string; channel?: string; received?: number; total?: number; code?: string; background?: boolean; prepared?: boolean } };
@@ -45,6 +46,8 @@ export function DesktopUpdate() {
   const [rollbackAvailable, setRollbackAvailable] = useState(false);
   const [runtimeSelection, setRuntimeSelection] = useState<RuntimeSelection>();
   const busy = useRef(false);
+  const [transfer, setTransfer] = useState<TransferSample | null>(null);
+  const [now, setNow] = useState(Date.now);
   const statusGeneration = useRef(0);
   const selectedChannel = useRef<string | null>(null);
   const invoke = (window as DesktopWindow).__TAURI__?.core.invoke;
@@ -53,6 +56,15 @@ export function DesktopUpdate() {
   const repairAvailable = runtimeSelection?.explicit !== true && runtimeSelection?.bundled_repair_available !== false;
   const restoreInPlace = state.phase === "error" && state.details?.code === "app_install_incomplete" && rollbackAvailable;
 
+  useEffect(() => {
+    setTransfer((previous) => observeTransfer(previous, state.phase, state.details?.received, Date.now()));
+  }, [state.phase, state.details?.received]);
+  useEffect(() => {
+    if (!observing) return;
+    const timer = window.setInterval(() => setNow(Date.now()), 1000);
+    return () => window.clearInterval(timer);
+  }, [observing]);
+
   function readStatus(value: UpdateStatus, includeState = true) {
     setVersion(value.app_version);
     setPlatform(value.platform ?? "");
@@ -195,6 +207,7 @@ export function DesktopUpdate() {
       {state.details?.version ? <p>{zh ? "目标版本：" : "Target: "}{state.details.version}</p> : null}
       <p role="status" aria-live="polite">{observing ? <RefreshCw className="is-spinning" size={14} aria-hidden="true" /> : null}{message[state.phase]}</p>
       {state.phase === "downloading" && state.details?.total ? <progress aria-label={zh ? "下载进度" : "Download progress"} max={state.details.total} value={state.details.received ?? 0} /> : null}
+      {state.phase === "downloading" && transfer ? <p className="personal-update-transfer" role="status">{transferText(transfer, state.details?.total, now, zh)}</p> : null}
       {invoke ? <div className="personal-update-actions">
         {state.phase === "restart_required" ? <button type="button" onClick={() => void run("restart")}>{zh ? "重启完成更新" : "Restart to finish"}</button> : <>
           {restoreInPlace ? <button type="button" onClick={() => void run("rollback")}>{zh ? "恢复上个版本" : "Restore previous version"}</button> : null}
diff --git a/apps/presentation/dashboard/src/features/personal-workspace/update-progress.ts b/apps/presentation/dashboard/src/features/personal-workspace/update-progress.ts
new file mode 100644
index 0000000000..e310297b7d
--- /dev/null
+++ b/apps/presentation/dashboard/src/features/personal-workspace/update-progress.ts
@@ -0,0 +1,26 @@
+// Presentation-only observations. The updater remains the authority for bytes,
+// phase and outcome; missing totals never become a fabricated percentage.
+export type TransferSample = {
+  phase: string; received: number | null; startedAt: number; changedAt: number;
+  sampledAt: number; bytesPerSecond: number | null;
+};
+
+export function observeTransfer(previous: TransferSample | null, phase: string, bytes: number | undefined, now: number): TransferSample {
+  const received = typeof bytes === "number" && Number.isFinite(bytes) && bytes >= 0 ? bytes : null;
+  if (!previous || previous.phase !== phase || (received !== null && previous.received !== null && received < previous.received)) {
+    return { phase, received, startedAt: now, changedAt: now, sampledAt: now, bytesPerSecond: null };
+  }
+  if (received === previous.received) return previous;
+  const elapsed = (now - previous.sampledAt) / 1000;
+  return { ...previous, received, sampledAt: now, changedAt: now,
+    bytesPerSecond: received !== null && previous.received !== null && elapsed > 0 ? (received - previous.received) / elapsed : null };
+}
+
+export function transferText(sample: TransferSample, total: number | undefined, now: number, zh: boolean): string {
+  const amount = sample.received === null ? (zh ? "等待下载数据" : "Waiting for download data") : `${(sample.received / 1048576).toFixed(1)} MiB`;
+  const length = typeof total === "number" && total > 0 ? `${(total / 1048576).toFixed(1)} MiB` : (zh ? "总量未知" : "size unknown");
+  const rate = sample.bytesPerSecond === null ? "" : ` · ${((now - sample.changedAt >= 5_000 ? 0 : sample.bytesPerSecond) / 1024).toFixed(0)} KiB/s`;
+  const elapsed = Math.max(0, Math.floor((now - sample.startedAt) / 1000));
+  const waiting = now - sample.changedAt >= 10_000 ? (zh ? " · 10 秒内无新进展" : " · no new progress for 10s") : "";
+  return `${amount} / ${length}${rate} · ${elapsed}s${waiting}`;
+}
diff --git a/docs/guides/installing-loopx.md b/docs/guides/installing-loopx.md
index c8ecdcfeea..f32463889e 100644
--- a/docs/guides/installing-loopx.md
+++ b/docs/guides/installing-loopx.md
@@ -18,12 +18,110 @@ Choose one installation owner and keep it authoritative:
 | Isolated CLI on an externally managed machine | `pipx` | `pipx install loopx` | `pipx upgrade loopx`, then refresh LoopX host material |
 | Existing uv tool environment | `uv` | `uv tool install loopx` | `uv tool upgrade loopx`, then refresh LoopX host material |
 | Contributor or source qualification | Git checkout | clone/fetch plus `scripts/install-local.sh` | update the checkout explicitly, rerun the installer, validate `loopx-canary` before promotion |
-| No-clone recovery fallback | LoopX archive snapshot | published archive installer | `loopx update apply` |
+| No-clone CLI or main preview | LoopX runtime snapshot | published installer with a prebuilt runtime | `loopx update apply` |
 
 PyPI is the canonical source for normal releases. A source checkout is a
 development and qualification surface, not a second implicit package channel.
 The archive snapshot remains a recovery path rather than a competing default.
 
+### Stable, main and source builds
+
+`loopx update apply` and `loopx upgrade` preserve the installation owner. pip
+and pipx continue to use their own environment and configured package index;
+they do not become archive installs. Stable PyPI wheels already contain Chat's
+frontend. Node is needed for the TypeScript runtime described below, but npm
+and a local frontend build are not part of a normal package upgrade.
+
+Managed snapshots inherit their recorded repo and ref, including a pinned SHA.
+New snapshots default to stable. Stable snapshots use the runtime attached to
+the latest stable release; main uses the latest **qualified** main snapshot,
+which can lag the branch while CI runs. The plan records its exact revision,
+version, size and SHA-256 before downloading. Missing metadata or a missing
+artifact stops the update without starting a source build. An older, diverged
+or unverified target requires explicit `--allow-downgrade` intent.
+
+```bash
+loopx update plan
+loopx upgrade --ref main             # snapshot installs only
+loopx update apply --ref stable --allow-downgrade
+loopx upgrade --ref main --source    # explicit source download + local build
+```
+
+To acquire a main snapshot without a checkout, use the published installer
+with `LOOPX_REF=main`. To build another ref, set
+`LOOPX_BUILD_FROM_SOURCE=1 LOOPX_REF=<ref>` before running that installer, or
+use the existing checkout installer. `LOOPX_ARCHIVE_URL` remains an explicit
+developer archive path. These choices never silently replace a pip/pipx or
+live-checkout owner. App updates continue through the native signed updater;
+a CLI-only installation does not install a Desktop App.
+
+Updates run in an owned background worker even when the caller waits for the
+result. A terminal/Agent disconnect does not cancel it. Repeating apply while
+that installation's update is running observes the existing operation. To
+return immediately and read back later:
+
+```bash
+loopx --format json upgrade --background
+loopx --format json update status
+loopx --format json update status --operation-id <returned-id>
+```
+
+Windows console launchers retain their existing self-uninstall protection in
+the original process. Follow the printed `python -m loopx.entrypoint` retry
+to update safely and use the durable worker; a worker never hides a locked
+console executable from that guard.
+
+JSON stdout contains the final result (or background operation descriptor).
+Progress is NDJSON on stderr: operation id, phase, downloaded bytes, total
+(`null` when unknown), observed rate, elapsed time and last progress time.
+The worker refreshes observation every five seconds and marks ten seconds
+without progress explicitly; a heartbeat is not a new byte or completed stage.
+Native App downloads show bytes, unknown totals, rate and stalled observation
+in **LoopX updates**, with existing check/install/recovery actions unchanged.
+
+Runtime downloads cache hash-verified bytes across invocations. Partial data
+is appended only with a matching strong ETag, origin, exact Content-Range and
+length; ignored/rejected Range restarts within the bounded budget. A complete
+cache hit is verified again before use. `LOOPX_UPDATE_CACHE` selects the cache;
+`LOOPX_UPDATE_STATE_DIR` selects private operation readback. Failed or killed
+workers remain failed/interrupted, and status reads never replay installation.
+Normal HTTP(S) proxy environment variables remain authoritative. Optional
+`LOOPX_DOWNLOAD_MIRRORS` is a comma-separated list of operator-trusted HTTPS
+base URLs serving identical filenames; publisher-listed sources are also
+supported. Timeouts or sustained low throughput can switch sources, always
+with the same expected revision/hash/size. This is not a claim that regional
+mirrors have been deployed or that any country's latency has been measured.
+
+The prebuilt runtime retains CLI/runtime code, skills and references, text
+documentation, developer smokes and the verified frontend. Named presentation
+videos/screenshots and frontend development sources are in the accompanying
+optional resource archive; illustrations remain available in versioned online
+docs. The release inventory lists every retained
+and optional file. Installers validate a candidate before activation and restore
+owned executable links on activation failure. Data-format migrations retain
+their independent receipts/backups; restoring an executable does not reverse
+migrated data. Existing managed-service ownership and pending-work restart
+guards still apply.
+
+中文：普通 stable 用户继续使用 PyPI；pip/pipx 保留原环境和索引。snapshot 沿用
+既有 main/stable/固定 SHA，默认取带前端的预构建包。main 指最新通过 CI 的快照；
+源码构建须显式 `--source`，不会静默降级到整仓下载。旧版本或无法核实的目标须
+明确 `--allow-downgrade`。`--background` 返回 operation id，断线后用
+`update status` 回读原升级；JSON 结果在 stdout，持续进度在 stderr。下载缓存、
+续传和可选可信镜像都校验同一制品；代理沿用用户配置。恢复旧可执行文件不回滚
+已迁移的数据。无 Node 时无需构建前端，但运行控制平面仍需下面要求的 Node。
+
+Release rollout: publish and qualify the stable runtime before deploying the
+new public bootstrap. The Pages lane checks this dependency and retains the
+previous deployed site if it is unavailable. `runtime-snapshots.yml` builds
+main artifacts; `release-artifacts.yml` reuses the release frontend/wheel lane
+for stable. Main archives and revision-addressed metadata share one dedicated
+`runtime-main` release, keeping product release history and stable-predecessor
+discovery intact; only the latest-channel metadata is replaced. Clean
+install/upgrade, package contents, byte size, provenance and
+phase timings are release evidence. A 20 MiB runtime is an evaluation target,
+not a waived dependency check or a global p50/p95 promise.
+
 LoopX's Effect Program core runs in a managed, idle-exiting TypeScript runtime
 and requires Node.js 22.22.3 or later; Node.js 24 LTS is the recommended primary
 runtime. LoopX starts and reuses that local runtime automatically; users do not
diff --git a/examples/codex-cli-packaged-install-smoke.py b/examples/codex-cli-packaged-install-smoke.py
index d0896d6535..fc989ae38b 100755
--- a/examples/codex-cli-packaged-install-smoke.py
+++ b/examples/codex-cli-packaged-install-smoke.py
@@ -131,6 +131,13 @@ def main() -> None:
 
         assert not (home / ".local" / "bin" / "loopx-canary").exists()
         assert not (home / ".local" / "bin" / "goal-harness-canary").exists()
+        # Retire this disposable installed runtime before its HOME is removed;
+        # an idle server can otherwise publish an exit-time cache during rmtree.
+        subprocess.run([sys.executable, "-c",
+            "from loopx.control_plane.effect_runtime import restart_effect_runtime; "
+            "restart_effect_runtime(wait_for_exit=True)"],
+            env={**env, "PYTHONPATH": str(release_root)}, cwd=release_root,
+            check=True, timeout=15, stdout=subprocess.DEVNULL)
 
 
 if __name__ == "__main__":
diff --git a/examples/desktop-update-browser-smoke.mjs b/examples/desktop-update-browser-smoke.mjs
index cc1ff0bcc1..b47a06356a 100644
--- a/examples/desktop-update-browser-smoke.mjs
+++ b/examples/desktop-update-browser-smoke.mjs
@@ -614,7 +614,27 @@ try {
   await page.getByRole("button", { name: "Update prepared" }).click();
   await page.getByText(/It installs on the next open/).waitFor();
   assert.deepEqual(calls.slice(beforeBackground).map((call) => call.args?.action), ["check"]);
-  console.log("desktop-update-browser-smoke: passed (confirmation, failure redaction, mobile, missing assets + reload, startup motion states, startup recovery, immediate terminal recovery, automatic startup presentation)");
+  // Reopening observes a transfer; it must neither restart it nor invent a
+  // total/percentage when the native source has not reported Content-Length.
+  const beforeTransfer = calls.length;
+  nativeState = { phase: "downloading", details: { channel: "stable", received: 1048576, total: 4194304 } };
+  await page.setViewportSize({ width: 1280, height: 900 });
+  await page.reload();
+  await page.getByRole("button", { name: "Updating…" }).click();
+  await page.getByText(/1.0 MiB \/ 4.0 MiB/).waitFor();
+  assert.equal(await page.getByRole("progressbar").getAttribute("value"), "1048576");
+  await page.screenshot({ path: resolve(output, "transfer-progress-desktop.png") });
+  nativeState = { phase: "downloading", details: { channel: "stable", received: 2097152 } };
+  await page.getByText(/2.0 MiB \/ size unknown/).waitFor();
+  assert.equal(await page.getByRole("progressbar").count(), 0);
+  await page.setViewportSize({ width: 390, height: 844 });
+  await page.getByRole("button", { name: "Open Goal navigation" }).click();
+  if (!await page.locator(".personal-update-panel").isVisible()) await page.getByRole("button", { name: "Updating…" }).click();
+  await page.getByText(/2.0 MiB \/ size unknown/).waitFor();
+  assert.ok(await page.evaluate(() => document.documentElement.scrollWidth <= innerWidth));
+  await page.screenshot({ path: resolve(output, "transfer-progress-mobile.png") });
+  assert.equal(calls.length, beforeTransfer, "observing transfer metrics must never start an installation");
+  console.log("desktop-update-browser-smoke: passed (recovery, channel ownership, transfer bytes/unknown totals, bilingual narrow/desktop)");
 } finally {
   await browser.close();
   await new Promise((done) => server.close(done));
diff --git a/examples/export-frontstage-share-bundle.mjs b/examples/export-frontstage-share-bundle.mjs
index 78c578a3a7..5e67d4eb95 100644
--- a/examples/export-frontstage-share-bundle.mjs
+++ b/examples/export-frontstage-share-bundle.mjs
@@ -131,6 +131,7 @@ async function copyHomepage(siteDir, base) {
   await cp(buildDir, siteDir, { force: true, recursive: true });
   await rm(buildDir, { force: true, recursive: true });
   await copyFile(resolve(repoRoot, installerScriptPath), resolve(siteDir, "install.sh"));
+  await copyFile(resolve(repoRoot, "loopx/cli_commands/update_artifact.py"), resolve(siteDir, "update-artifact.py"));
 }
 
 async function copyPublicSiteRoutes(siteDir) {
diff --git a/examples/loopx-update-smoke.py b/examples/loopx-update-smoke.py
index a68428fea5..5863fddd84 100644
--- a/examples/loopx-update-smoke.py
+++ b/examples/loopx-update-smoke.py
@@ -192,14 +192,14 @@ def test_module_plan() -> None:
     assert f"Manifest package version tag: `{release_version_tag()}`" in rendered, rendered
 
 
-def test_default_source_uses_stable_ref() -> None:
+def test_installed_stable_channel_is_inherited() -> None:
     with mock.patch.dict(os.environ, {}, clear=True):
         payload = build_update_plan(doctor_payload=fake_doctor_payload())
     assert payload["source"]["ref"] == DEFAULT_UPDATE_REF == "stable", payload
     assert payload["source"]["channel"] == "github_archive_stable", payload
-    assert payload["source"]["ref_source"] == "default_stable", payload
+    assert payload["source"]["ref_source"] == "installed_manifest", payload
     assert "/tar.gz/stable" in payload["source"]["archive_url"], payload
-    assert payload["plan"]["install_command"] == "loopx update apply", payload
+    assert payload["plan"]["install_command"] == "loopx update apply --repo example/loopx --ref stable", payload
     assert "--archive-url" not in payload["plan"]["install_command"], payload
     default_env = _installer_env_for_source(
         payload["source"],
@@ -331,6 +331,7 @@ def test_execute_update_uses_persisted_release_python() -> None:
         (release_root / ".loopx-python").write_text(f"{missing_python}\n", encoding="utf-8")
         payload = build_update_plan(
             execute=True,
+            build_from_source=True,
             doctor_payload=fake_doctor_payload_for_release(release_root, home=home),
         )
         failed = subprocess.CompletedProcess(
@@ -385,6 +386,7 @@ def read(self, limit: int) -> bytes:
 def test_check_compares_selected_source_version() -> None:
     with mock.patch("loopx.self_update.urlopen", return_value=FakeVersionResponse("9.9.9")) as fetch:
         payload = build_update_plan(
+            build_from_source=True,
             repo="example/loopx",
             ref="stable",
             check_only=True,
@@ -403,6 +405,7 @@ def test_check_compares_selected_source_version() -> None:
 
     with mock.patch("loopx.self_update.urlopen", return_value=FakeVersionResponse(__version__)):
         current_payload = build_update_plan(
+            build_from_source=True,
             repo="example/loopx",
             ref="stable",
             check_only=True,
@@ -416,6 +419,7 @@ def test_check_compares_selected_source_version() -> None:
 def test_check_degrades_when_source_version_is_unavailable() -> None:
     with mock.patch("loopx.self_update.urlopen", side_effect=OSError("offline")):
         payload = build_update_plan(
+            build_from_source=True,
             repo="example/loopx",
             ref="stable",
             check_only=True,
@@ -446,6 +450,7 @@ def test_immutable_source_ref_qualifies_activation() -> None:
     ):
         activated = build_update_plan(
             repo="example/loopx",
+            build_from_source=True,
             ref=pinned,
             check_only=True,
             doctor_payload=immutable_fresh_doctor_payload(pinned),
@@ -460,6 +465,7 @@ def test_immutable_source_ref_qualifies_activation() -> None:
     other = TARGET_COMMIT
     unproven = build_update_plan(
         repo="example/loopx",
+        build_from_source=True,
         ref=other,
         check_only=True,
         doctor_payload=immutable_fresh_doctor_payload(other),
@@ -583,7 +589,7 @@ def test_cli_rollback_previous_with_temp_home() -> None:
 
 def main() -> int:
     test_module_plan()
-    test_default_source_uses_stable_ref()
+    test_installed_stable_channel_is_inherited()
     test_explicit_archive_url_reaches_installer()
     test_update_preview_uses_managed_download_path()
     test_execute_update_propagates_installer_download_failure()
diff --git a/loopx/cli_commands/support_control_update.py b/loopx/cli_commands/support_control_update.py
index b4e7949f2c..e230c9cd30 100644
--- a/loopx/cli_commands/support_control_update.py
+++ b/loopx/cli_commands/support_control_update.py
@@ -18,6 +18,9 @@
 
 import argparse
 import json
+import os
+import sys
+import zipimport
 from collections.abc import Callable
 from pathlib import Path
 from functools import partial
@@ -65,7 +68,7 @@ def register_update_command(
     update_parser.add_argument(
         "update_action",
         nargs="?",
-        choices=tuple(action.value for action in UpdateAction),
+        choices=(*tuple(action.value for action in UpdateAction), "status"),
         help="Explicit intent: check (read only), plan (read only), or apply (mutating).",
     )
     update_mode = update_parser.add_mutually_exclusive_group()
@@ -95,8 +98,12 @@ def register_update_command(
     )
     update_parser.add_argument(
         "--ref",
-        help="Git ref used by the installer archive. Defaults to LOOPX_REF or stable.",
+        help="Update channel/tag/SHA. Inherits the installed snapshot's choice; new installs use stable.",
     )
+    update_parser.add_argument("--source", action="store_true", help="Explicitly download and build a source snapshot instead of a prebuilt runtime.")
+    update_parser.add_argument("--allow-downgrade", action="store_true", help="Explicitly accept an older, diverged or unverified snapshot target.")
+    update_parser.add_argument("--background", action="store_true", help="Start or observe one durable update and return its operation id immediately.")
+    update_parser.add_argument("--operation-id", help="Read a previous update with `update status --operation-id ID`.")
     update_parser.add_argument(
         "--archive-url",
         help="Explicit tarball URL passed to the installer as LOOPX_ARCHIVE_URL.",
@@ -131,8 +138,16 @@ def handle_update_command(
     upgrade = args.command == "upgrade"
     from .support_control_update_progress import UpdateProgress
 
-    progress = UpdateProgress(output_format(args) != "json")
+    progress = UpdateProgress(json_output=output_format(args) == "json", pending=os.environ.get("LOOPX_UPDATE_WORKER") != "1")
     try:
+        if args.update_action == "status":
+            if any((args.check, args.dry_run, args.execute, args.rollback, args.repo, args.ref, args.archive_url,
+                    getattr(args, "source", False), getattr(args, "allow_downgrade", False), getattr(args, "background", False))):
+                raise ValueError("update status is read only; use only --operation-id to select an operation")
+            from .update_operation import read_operation
+            payload = read_operation(getattr(args, "operation_id", None))
+            print_payload(payload, output_format(args), lambda value: f"Update {value.get('operation_id') or '—'}: {value['status']}")
+            return 0 if payload.get("ok") else 1
         update_action = resolve_update_action(
             args.update_action,
             check=args.check,
@@ -145,6 +160,10 @@ def handle_update_command(
             progress.report("Checking installation owner and update source")
         if args.timeout_seconds <= 0:
             raise ValueError("--timeout-seconds must be positive")
+        if getattr(args, "operation_id", None):
+            raise ValueError("--operation-id requires `update status`")
+        if getattr(args, "background", False) and update_action is not UpdateAction.APPLY:
+            raise ValueError("--background requires update apply or upgrade")
         if upgrade and args.rollback:
             raise ValueError("CLI rollback uses `loopx update --rollback`; App rollback uses Recovery & updates")
         if args.rollback and args.update_action:
@@ -153,6 +172,28 @@ def handle_update_command(
             )
         if args.installed_doctor_json and update_action is not UpdateAction.CHECK:
             raise ValueError("--installed-doctor-json requires `loopx update check`")
+        # Keep the original console's owning-entry refusal visible. Moving it
+        # to python first would hide the still-running Windows .exe from that
+        # guard and let pip uninstall the observer's locked launcher.
+        windows_console = sys.platform == "win32" and isinstance(
+            getattr(sys.modules["__main__"], "__loader__", None), zipimport.zipimporter)
+        if update_action is UpdateAction.APPLY and not args.rollback and os.environ.get("LOOPX_UPDATE_WORKER") != "1" and not windows_console:
+            from .update_operation import start_or_observe, observe
+            progress.report("Starting or reconnecting to update")
+            operation = start_or_observe(args, registry_path, registry_was_supplied=registry_was_supplied)
+            progress.operation_id = operation["operation_id"]
+            progress.stop()
+            payload = operation if getattr(args, "background", False) else observe(
+                operation["operation_id"], json_output=output_format(args) == "json")
+            if getattr(args, "background", False):
+                renderer = lambda value: f"Update {value['operation_id']}: {value['status']}\nRead back with loopx update status --operation-id {value['operation_id']}"
+            elif upgrade:
+                from .support_control_upgrade import render_upgrade_markdown
+                renderer = render_upgrade_markdown
+            else:
+                renderer = render_update_plan_markdown
+            print_payload(payload, output_format(args), renderer)
+            return 0 if payload.get("ok") else 1
         if args.rollback:
             payload = build_rollback_plan(release_id=args.rollback)
             payload = execute_rollback_plan(
@@ -175,6 +216,8 @@ def handle_update_command(
                     repo=args.repo, ref=args.ref, archive_url=args.archive_url,
                     action=update_action, doctor_payload=doctor_payload,
                     registry_path=registry_path, runtime_root=args.runtime_root,
+                    build_from_source=getattr(args, "source", False),
+                    allow_downgrade=getattr(args, "allow_downgrade", False),
                 )
             else:
                 payload = build_update_plan(
@@ -185,10 +228,24 @@ def handle_update_command(
                     doctor_payload=doctor_payload,
                     registry_path=registry_path,
                     runtime_root=args.runtime_root,
+                    build_from_source=getattr(args, "source", False),
+                    allow_downgrade=getattr(args, "allow_downgrade", False),
                 )
             payload["installed_doctor_source"] = (
                 "explicit_json" if doctor_payload is not None else "current_runtime"
             )
+            if update_action is UpdateAction.APPLY:
+                selected = payload.get("source", {})
+                lifecycle = payload.get("install_lifecycle", {})
+                target = {
+                    "channel": selected.get("ref"), "revision": selected.get("resolved_revision"),
+                    "version": selected.get("artifact", {}).get("package_version"),
+                    "distribution": selected.get("distribution"),
+                } if lifecycle.get("owner") == "loopx_release_snapshot" else {
+                    "owner": lifecycle.get("owner"), "driver": lifecycle.get("execution_driver"),
+                    "resolution": "installation_owner",
+                }
+                progress.report({"stage": "target", "target": target})
             if update_action is UpdateAction.APPLY and payload.get("plan", {}).get(
                 "apply_supported"
             ):
@@ -229,6 +286,8 @@ def handle_update_command(
         }
     finally:
         progress.stop()
+    if update_action is UpdateAction.APPLY:
+        payload["update_progress"] = dict(progress.snapshot(), status="succeeded" if payload.get("ok") else "failed")
     renderer = render_update_plan_markdown
     if upgrade:
         from .support_control_upgrade import render_upgrade_markdown
diff --git a/loopx/cli_commands/support_control_update_progress.py b/loopx/cli_commands/support_control_update_progress.py
index 274f3cc732..04f2eb236a 100644
--- a/loopx/cli_commands/support_control_update_progress.py
+++ b/loopx/cli_commands/support_control_update_progress.py
@@ -1,39 +1,110 @@
-"""Bounded, public-safe progress on stderr; JSON stdout stays machine readable."""
+"""Observable update progress; stdout remains a single final JSON document."""
 
 from __future__ import annotations
 
+import json
+import os
+from pathlib import Path
 import sys
 import threading
 import time
+from uuid import uuid4
+
+from .update_artifact import atomic_json
 
 
 class UpdateProgress:
-    def __init__(self, enabled: bool) -> None:
+    def __init__(self, enabled: bool = True, *, json_output: bool = False,
+                 operation_id: str | None = None, status_path: Path | None = None,
+                 pending: bool = False) -> None:
         self.enabled = enabled
+        self.json_output = json_output
+        self.operation_id = None if pending else (operation_id or os.environ.get("LOOPX_UPDATE_OPERATION_ID") or uuid4().hex)
+        self.status_path = status_path or (Path(os.environ["LOOPX_UPDATE_STATUS_PATH"])
+            if os.environ.get("LOOPX_UPDATE_STATUS_PATH") else None)
         self.started = time.monotonic()
-        self.stage = "Checking installation"
+        self.changed = self.started
+        self.stage_started = self.started
+        self.stage = "resolve"
+        self.received: int | None = None
+        self.total: int | None = None
+        self.rate = 0.0
+        self.last_sample = (self.started, 0)
+        self.durations: dict[str, float] = {}
+        self.target: dict | None = None
         self.stopped = threading.Event()
+        self.lock = threading.RLock()
         self.thread: threading.Thread | None = None
 
-    def report(self, stage: str) -> None:
-        if not self.enabled:
-            return
-        self.stage = stage
-        self._write()
-        if self.thread is None:
-            self.thread = threading.Thread(target=self._tick, daemon=True)
-            self.thread.start()
+    def report(self, event: str | dict) -> None:
+        with self.lock:
+            now = time.monotonic()
+            details = event if isinstance(event, dict) else {"stage": event}
+            if isinstance(details.get("target"), dict):
+                self.target = details["target"]
+            stage = str(details.get("stage", self.stage))
+            if stage != self.stage:
+                self.durations[self.stage] = self.durations.get(self.stage, 0) + now - self.stage_started
+                self.stage, self.stage_started = stage, now
+                self.received, self.total, self.rate = None, None, 0
+                self.last_sample = (now, 0)
+                self.changed = now
+            received = details.get("received")
+            if type(received) is int and received >= 0:
+                if received != self.received:
+                    self.changed = now
+                previous_time, previous_bytes = self.last_sample
+                self.rate = max(0, received - previous_bytes) / max(.001, now - previous_time)
+                self.received = received
+                self.last_sample = (now, received)
+            total = details.get("total")
+            if type(total) is int and total > 0:
+                self.total = total
+            self._write()
+            if self.thread is None and not self.stopped.is_set():
+                self.thread = threading.Thread(target=self._tick, daemon=True)
+                self.thread.start()
+
+    def snapshot(self) -> dict:
+        now = time.monotonic()
+        return {
+            "schema_version": "loopx_update_progress_v1", "operation_id": self.operation_id,
+            "pid": os.getpid(), "stage": self.stage, "status": "running",
+            "target": self.target,
+            "received": self.received, "total": self.total,
+            "bytes_per_second": round(self.rate, 1) if now - self.last_sample[0] < 5 else 0,
+            "elapsed_seconds": round(now - self.started, 1),
+            "stage_seconds": round(now - self.stage_started, 1),
+            "last_progress_at": round(time.time() - (now - self.changed), 3),
+            "updated_at": time.time(), "stalled": now - self.changed >= 10,
+            "stage_durations": {key: round(value, 3) for key, value in {
+                **self.durations, self.stage: self.durations.get(self.stage, 0) + now - self.stage_started}.items()},
+        }
 
     def _write(self) -> None:
-        print(
-            f"LoopX: {self.stage} ({int(time.monotonic() - self.started)}s elapsed)",
-            file=sys.stderr,
-            flush=True,
-        )
+        event = self.snapshot()
+        if self.status_path:
+            atomic_json(self.status_path, event)
+        if not self.enabled:
+            return
+        try:
+            if self.json_output:
+                print(json.dumps(event), file=sys.stderr, flush=True)
+            else:
+                amount = ""
+                if self.received is not None:
+                    total = f"{self.total / 1048576:.1f} MiB" if self.total else "unknown total"
+                    amount = f" · {self.received / 1048576:.1f}/{total} · {event['bytes_per_second'] / 1024:.0f} KiB/s"
+                waiting = " · no new progress for 10s" if event["stalled"] else ""
+                identity = f" [{self.operation_id[:8]}]" if self.operation_id else ""
+                print(f"LoopX{identity}: {self.stage}{amount} ({event['elapsed_seconds']}s){waiting}", file=sys.stderr, flush=True)
+        except (BrokenPipeError, OSError):
+            self.enabled = False
 
     def _tick(self) -> None:
-        while not self.stopped.wait(10):
-            self._write()
+        while not self.stopped.wait(5):
+            with self.lock:
+                self._write()
 
     def stop(self) -> None:
         self.stopped.set()
diff --git a/loopx/cli_commands/support_control_upgrade.py b/loopx/cli_commands/support_control_upgrade.py
index 84f579fd31..feb1c3dc82 100644
--- a/loopx/cli_commands/support_control_upgrade.py
+++ b/loopx/cli_commands/support_control_upgrade.py
@@ -86,6 +86,8 @@ def build_upgrade_plan(
     doctor_payload: dict[str, Any] | None = None,
     registry_path: Path | None = None,
     runtime_root: str | None = None,
+    build_from_source: bool = False,
+    allow_downgrade: bool = False,
 ) -> dict[str, Any]:
     doctor = (
         doctor_payload
@@ -111,8 +113,11 @@ def build_upgrade_plan(
         doctor_payload=doctor,
         registry_path=registry_path,
         runtime_root=runtime_root,
+        build_from_source=build_from_source,
+        allow_downgrade=allow_downgrade,
+        resolve_remote=doctor_payload is None,
     )
-    desktop = desktop_upgrade_plan(selected_ref)
+    desktop = desktop_upgrade_plan(payload.get("source", {}).get("ref") if selected_ref else selected_ref)
     payload["desktop_upgrade"] = desktop
     next_action = payload.get("next_action", {})
     command = next_action.get("command")
diff --git a/loopx/cli_commands/update_artifact.py b/loopx/cli_commands/update_artifact.py
new file mode 100644
index 0000000000..a657afa609
--- /dev/null
+++ b/loopx/cli_commands/update_artifact.py
@@ -0,0 +1,400 @@
+"""Bootstrap IO for prebuilt snapshots (stdlib only, including before Node exists).
+
+This is also the website installer's Python bootstrap. It deliberately reuses
+install-local.sh for candidate validation, migrations and activation. It is not
+a second control-plane decision owner or a package-manager replacement.
+"""
+
+from __future__ import annotations
+
+import argparse
+from contextlib import contextmanager
+import hashlib
+from http.client import HTTPException
+import json
+import os
+from pathlib import Path, PurePosixPath
+import re
+import subprocess
+import sys
+import tarfile
+import tempfile
+import time
+from typing import Any
+from urllib.error import HTTPError
+from urllib.parse import urlsplit
+from urllib.request import HTTPRedirectHandler, Request, build_opener
+
+RUNTIME_ARTIFACT_SCHEMA_VERSION = "loopx_runtime_artifact_v1"
+SHA = re.compile(r"[0-9a-f]{40}")
+HASH = re.compile(r"[0-9a-f]{64}")
+MAX_MANIFEST = 256 * 1024
+MAX_ARCHIVE = 256 * 1024 * 1024
+MAX_EXPANDED = 1024 * 1024 * 1024
+
+
+def safe_url(url: str) -> str:
+    parsed = urlsplit(url)
+    # HTTP is only useful for disposable loopback qualification servers.
+    if (parsed.scheme != "https" and not (
+        parsed.scheme == "http" and parsed.hostname in {"127.0.0.1", "::1", "localhost"}
+    )) or not parsed.hostname or parsed.username or parsed.password or parsed.fragment:
+        raise ValueError("Update sources require HTTPS without embedded credentials")
+    return url
+
+
+class UpdateRedirectHandler(HTTPRedirectHandler):
+    def redirect_request(self, request, response, code, message, headers, new_url):
+        safe_url(new_url)
+        if request.type == "https" and urlsplit(new_url).scheme != "https":
+            raise ValueError("Update redirects must preserve TLS")
+        return super().redirect_request(request, response, code, message, headers, new_url)
+
+
+def read_json(url: str) -> dict[str, Any]:
+    request = Request(safe_url(url), headers={"User-Agent": "LoopX-update", "Accept": "application/json"})
+    try:
+        with build_opener(UpdateRedirectHandler()).open(request, timeout=10) as response:
+            raw = response.read(MAX_MANIFEST + 1)
+        if len(raw) > MAX_MANIFEST:
+            raise ValueError("Update metadata exceeds its size limit")
+        result = json.loads(raw)
+    except (OSError, ValueError, HTTPException) as exc:
+        # Do not echo proxy credentials, signed URLs or response bodies.
+        raise ValueError("Update metadata unavailable; retry online or explicitly use --source") from exc
+    if not isinstance(result, dict):
+        raise ValueError("Update metadata must be an object")
+    return result
+
+
+def validate_manifest(value: dict[str, Any], *, revision: str | None = None) -> dict[str, Any]:
+    if value.get("schema_version") != RUNTIME_ARTIFACT_SCHEMA_VERSION:
+        raise ValueError("Unsupported runtime artifact metadata")
+    if not SHA.fullmatch(str(value.get("source_revision", ""))):
+        raise ValueError("Runtime artifact has no immutable source revision")
+    if revision and value["source_revision"] != revision:
+        raise ValueError("Runtime artifact belongs to another revision")
+    if not HASH.fullmatch(str(value.get("sha256", ""))):
+        raise ValueError("Runtime artifact has no SHA-256")
+    if type(value.get("size")) is not int or not 0 < value["size"] <= MAX_ARCHIVE:
+        raise ValueError("Runtime artifact has an invalid byte length")
+    if not isinstance(value.get("package_version"), str) or not value["package_version"]:
+        raise ValueError("Runtime artifact has no package version")
+    urls = value.get("urls")
+    if not isinstance(urls, list) or not urls or len(urls) > 8 or not all(isinstance(u, str) for u in urls):
+        raise ValueError("Runtime artifact has no bounded download sources")
+    for url in urls:
+        safe_url(url)
+    return dict(value)
+
+
+def resolve_artifact(repo: str, ref: str) -> dict[str, Any]:
+    if not re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", repo):
+        raise ValueError("Update repo must use GitHub owner/name syntax")
+    if ref == "stable":
+        url = f"https://github.com/{repo}/releases/latest/download/loopx-runtime.json"
+    elif ref == "main":
+        url = f"https://github.com/{repo}/releases/download/runtime-main/loopx-runtime.json"
+    elif SHA.fullmatch(ref):
+        url = f"https://github.com/{repo}/releases/download/runtime-main/loopx-runtime-{ref}.json"
+    elif re.fullmatch(r"v[0-9][A-Za-z0-9.+-]*", ref):
+        url = f"https://github.com/{repo}/releases/download/{ref}/loopx-runtime.json"
+    else:
+        raise ValueError("This ref has no runtime feed; explicitly use --source to build it")
+    artifact = validate_manifest(read_json(url), revision=ref if SHA.fullmatch(ref) else None)
+    if ref in {"main", "stable"} and artifact.get("channel") != ref:
+        raise ValueError("Runtime artifact belongs to another channel")
+    return artifact
+
+
+def target_direction(repo: str, current: str | None, target: str) -> str:
+    if current == target:
+        return "identical"
+    if not SHA.fullmatch(current or ""):
+        return "unknown"
+    try:
+        value = read_json(f"https://api.github.com/repos/{repo}/compare/{current}...{target}")
+        return str(value.get("status")) if value.get("status") in {"ahead", "behind", "diverged", "identical"} else "unknown"
+    except ValueError:
+        return "unknown"
+
+
+def cache_root() -> Path:
+    return Path(os.environ.get("LOOPX_UPDATE_CACHE") or (
+        Path(os.environ.get("XDG_CACHE_HOME") or Path.home() / ".cache") / "loopx/updates"
+    )).expanduser()
+
+
+def atomic_json(path: Path, value: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+    descriptor, raw = tempfile.mkstemp(dir=path.parent, prefix=".update-")
+    try:
+        with os.fdopen(descriptor, "w", encoding="utf-8") as output:
+            json.dump(value, output, ensure_ascii=False)
+            output.flush()
+            os.fsync(output.fileno())
+        os.replace(raw, path)
+    finally:
+        Path(raw).unlink(missing_ok=True)
+
+
+@contextmanager
+def artifact_lock(path: Path):
+    # Kernel locks are released on process death. A killed downloader leaves
+    # its partial bytes/validator, never an unbreakable sentinel lock.
+    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+    with path.open("a+b") as handle:
+        if os.name == "nt":
+            import msvcrt
+            handle.write(b"\0")
+            handle.flush()
+            handle.seek(0)
+            msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
+        else:
+            import fcntl
+            fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        try:
+            yield
+        finally:
+            if os.name == "nt":
+                handle.seek(0)
+                msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
+            else:
+                fcntl.flock(handle, fcntl.LOCK_UN)
+
+
+def file_hash(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for block in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(block)
+    return digest.hexdigest()
+
+
+def download_artifact(manifest: dict[str, Any], *, progress=lambda _event: None,
+                      timeout: float = 600, cache: Path | None = None) -> Path:
+    manifest = validate_manifest(manifest)
+    directory = cache or cache_root()
+    directory.mkdir(parents=True, exist_ok=True, mode=0o700)
+    key = f"{manifest['source_revision']}-{manifest['sha256']}"
+    final = directory / (key + ".tar.gz")
+    partial = directory / (key + ".partial")
+    metadata = directory / (key + ".json")
+    deadline = time.monotonic() + timeout
+    expected = manifest["size"]
+    with artifact_lock(directory / (key + ".lock")):
+        if final.is_file() and final.stat().st_size == expected and file_hash(final) == manifest["sha256"]:
+            progress({"stage": "download", "received": expected, "total": expected, "cached": True})
+            return final
+        final.unlink(missing_ok=True)
+        urls = list(manifest["urls"])
+        # Mirrors are operator-configured or publisher-listed, never selected
+        # by country/proxy guessing. All serve the same hash-pinned bytes.
+        filename = Path(urlsplit(urls[0]).path).name
+        for base in os.environ.get("LOOPX_DOWNLOAD_MIRRORS", "").split(","):
+            if base.strip():
+                urls.append(safe_url(base.strip().rstrip("/") + "/" + filename))
+        errors: list[str] = []
+        for attempt in range(3 * len(urls)):
+            if time.monotonic() >= deadline:
+                break
+            url = urls[attempt % len(urls)]
+            offset, etag = 0, None
+            try:
+                saved = json.loads(metadata.read_text())
+                candidate = saved.get("etag")
+                if (saved.get("url_hash") == hashlib.sha256(url.encode()).hexdigest()
+                    and isinstance(candidate, str) and candidate.startswith('"') and not candidate.startswith('W/')
+                    and partial.is_file() and partial.stat().st_size < expected):
+                    offset, etag = partial.stat().st_size, candidate
+            except (OSError, ValueError):
+                pass
+            headers = {"User-Agent": "LoopX-update", "Accept-Encoding": "identity"}
+            if offset and etag:
+                headers.update({"Range": f"bytes={offset}-", "If-Range": etag})
+            progress({"stage": "download", "received": offset, "total": expected, "attempt": attempt + 1})
+            try:
+                with build_opener(UpdateRedirectHandler()).open(Request(safe_url(url), headers=headers), timeout=min(10, max(.1, deadline - time.monotonic()))) as response:
+                    status = response.status
+                    if status == 206:
+                        content_range = response.headers.get("Content-Range", "")
+                        if not offset or content_range != f"bytes {offset}-{expected - 1}/{expected}" or response.headers.get("ETag") != etag:
+                            partial.unlink(missing_ok=True)
+                            metadata.unlink(missing_ok=True)
+                            raise ValueError("invalid_range")
+                    elif status == 200:
+                        offset = 0  # Range ignored or validator changed: replace, never append.
+                    else:
+                        raise ValueError("unexpected_status")
+                    length = response.headers.get("Content-Length")
+                    if length is not None and int(length) != expected - offset:
+                        raise ValueError("invalid_length")
+                    atomic_json(metadata, {"etag": response.headers.get("ETag"),
+                        "url_hash": hashlib.sha256(url.encode()).hexdigest()})
+                    received = offset
+                    window_start, window_bytes, emitted = time.monotonic(), received, 0.0
+                    with partial.open("ab" if offset else "wb") as output:
+                        while True:
+                            if time.monotonic() >= deadline:
+                                raise TimeoutError("download_deadline")
+                            block = response.read1(64 * 1024)
+                            if not block:
+                                break
+                            received += len(block)
+                            if received > expected:
+                                raise ValueError("invalid_length")
+                            output.write(block)
+                            now = time.monotonic()
+                            if now - emitted >= 1:
+                                progress({"stage": "download", "received": received, "total": expected})
+                                emitted = now
+                            # A sustained <16 KiB/s window can switch to another
+                            # trusted source. A lone slow packet cannot trigger it.
+                            if now - window_start >= 30:
+                                if len(urls) > 1 and (received - window_bytes) / (now - window_start) < 16 * 1024:
+                                    raise TimeoutError("slow_source")
+                                window_start, window_bytes = now, received
+                        output.flush()
+                        os.fsync(output.fileno())
+                if partial.stat().st_size < expected:
+                    raise OSError("incomplete_transfer")
+                if partial.stat().st_size != expected or file_hash(partial) != manifest["sha256"]:
+                    partial.unlink(missing_ok=True)
+                    metadata.unlink(missing_ok=True)
+                    raise ValueError("integrity_mismatch")
+                os.replace(partial, final)
+                progress({"stage": "download", "received": expected, "total": expected, "cached": False})
+                return final
+            except (OSError, ValueError, HTTPException) as exc:
+                code = str(exc) if isinstance(exc, ValueError) else (f"http_{exc.code}" if isinstance(exc, HTTPError) else "transport_interrupted")
+                errors.append(code if code in {"invalid_range", "invalid_length", "unexpected_status", "integrity_mismatch"} or code.startswith("http_") else "transport_interrupted")
+                # Reject stale/complete partials after 416; retries retain only
+                # well-formed, validator-bound incomplete transfers.
+                if isinstance(exc, HTTPError) and exc.code == 416:
+                    partial.unlink(missing_ok=True)
+                    metadata.unlink(missing_ok=True)
+        raise ValueError("Runtime download failed (" + ", ".join(errors[-3:]) + "); cached partial retained when safe")
+
+
+def extract_artifact(archive: Path, destination: Path) -> Path:
+    with tarfile.open(archive, "r:gz") as source:
+        members = source.getmembers()
+        if sum(member.size for member in members) > MAX_EXPANDED or len(members) > 100_000:
+            raise ValueError("Runtime archive exceeds extraction limits")
+        for member in members:
+            path = PurePosixPath(member.name)
+            if path.is_absolute() or ".." in path.parts or "\\" in member.name or ":" in member.name or not (member.isfile() or member.isdir()):
+                raise ValueError("Runtime archive contains an unsafe entry")
+        # No links, traversal, devices or platform-specific path escapes.
+        options = {"filter": "data"} if hasattr(tarfile, "data_filter") else {}
+        source.extractall(destination, members=members, **options)
+    if not (destination / "scripts/install-local.sh").is_file():
+        raise ValueError("Runtime archive has no snapshot installer")
+    return destination
+
+
+def run_installer(command: list[str], *, env: dict[str, str], timeout: float,
+                  progress=lambda _event: None) -> subprocess.CompletedProcess[str]:
+    """Stream only explicit installer events; retain bounded private diagnostics."""
+    import threading
+    tails = {"stdout": "", "stderr": ""}
+    process = subprocess.Popen(command, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+        text=True, encoding="utf-8", errors="replace", start_new_session=os.name != "nt")
+    def consume(name, stream):
+        for line in stream:
+            if name == "stderr" and line.startswith("LOOPX_UPDATE_EVENT "):
+                try:
+                    event = json.loads(line[len("LOOPX_UPDATE_EVENT "):])
+                    if isinstance(event, dict) and event.get("stage") in {"download", "frontend", "copy", "validate", "migrate", "activate", "skills"}:
+                        safe = {"stage": event["stage"]}
+                        for field in ("received", "total"):
+                            if type(event.get(field)) is int and event[field] >= 0:
+                                safe[field] = event[field]
+                        progress(safe)
+                except ValueError:
+                    pass
+            else:
+                tails[name] = (tails[name] + line)[-16_384:]
+    threads = [threading.Thread(target=consume, args=(name, stream), daemon=True)
+        for name, stream in (("stdout", process.stdout), ("stderr", process.stderr))]
+    for thread in threads:
+        thread.start()
+    try:
+        code = process.wait(timeout=timeout)
+    except subprocess.TimeoutExpired:
+        import signal
+        if os.name != "nt":
+            os.killpg(process.pid, signal.SIGTERM)
+        else:
+            process.terminate()
+        try:
+            process.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            if os.name != "nt":
+                os.killpg(process.pid, signal.SIGKILL)
+            else:
+                process.kill()
+            process.wait()
+        code = 124
+    for thread in threads:
+        thread.join(timeout=2)
+    return subprocess.CompletedProcess(command, code, tails["stdout"], tails["stderr"])
+
+
+def install_artifact(manifest: dict[str, Any], *, env: dict[str, str], timeout_seconds: int,
+                     progress=lambda _event: None):
+    started = time.monotonic()
+    archive = download_artifact(manifest, progress=progress, timeout=timeout_seconds)
+    progress({"stage": "verify"})
+    with tempfile.TemporaryDirectory(prefix="loopx-runtime-") as raw:
+        root = extract_artifact(archive, Path(raw))
+        identity = json.loads((root / "runtime-artifact.json").read_text())
+        if identity.get("source_revision") != manifest["source_revision"] or identity.get("package_version") != manifest["package_version"]:
+            raise ValueError("Runtime identity disagrees with its release metadata")
+        selected = dict(env)
+        selected.update({"LOOPX_RESOLVED_SOURCE_GIT_COMMIT": manifest["source_revision"],
+            "LOOPX_ARCHIVE_SHA256": manifest["sha256"], "LOOPX_ARCHIVE_URL": manifest["urls"][0],
+            "LOOPX_PREBUILT_RUNTIME": "1", "LOOPX_INSTALL_CANARY": "0", "LOOPX_PROMOTE_DEFAULT": "1",
+            "LOOPX_PROMOTION_MODE": "trusted_github_archive"})
+        result = run_installer(["bash", str(root / "scripts/install-local.sh")], env=selected,
+            timeout=max(.1, timeout_seconds - (time.monotonic() - started)), progress=progress)
+    return result, {"stage": "runtime_artifact", "source_revision": manifest["source_revision"],
+        "sha256": manifest["sha256"], "bytes": manifest["size"], "elapsed_seconds": round(time.monotonic() - started, 3)}
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--repo")
+    parser.add_argument("--ref")
+    args = parser.parse_args()
+    def report(event):
+        print("LOOPX_UPDATE_EVENT " + json.dumps(event), file=sys.stderr, flush=True)
+    try:
+        report({"stage": "resolve"})
+        installed = {}
+        entry = Path(os.environ.get("LOOPX_BIN_DIR") or Path.home() / ".local/bin") / "loopx"
+        if entry.is_symlink():
+            try:
+                installed = json.loads((entry.resolve().parents[1] / "release.json").read_text()).get("source", {})
+            except (OSError, ValueError):
+                pass
+        if (entry.exists() or entry.is_symlink()) and not installed:
+            raise ValueError("An existing CLI has another or unknown owner; use its `loopx upgrade` or original package manager")
+        selected_repo = args.repo or (os.environ.get("LOOPX_REPO") if os.environ.get("LOOPX_UPDATE_REPO_EXPLICIT") == "1" else None) or installed.get("repo") or "loopx-project/loopx"
+        selected_ref = args.ref or (os.environ.get("LOOPX_REF") if os.environ.get("LOOPX_UPDATE_REF_EXPLICIT") == "1" else None) or installed.get("ref") or "stable"
+        manifest = resolve_artifact(selected_repo, selected_ref)
+        if installed and target_direction(selected_repo, installed.get("git_commit"), manifest["source_revision"]) not in {"ahead", "identical"} and os.environ.get("LOOPX_ALLOW_DOWNGRADE") != "1":
+            raise ValueError("Older, diverged or unverified target; inspect it before explicitly setting LOOPX_ALLOW_DOWNGRADE=1")
+        env = dict(os.environ, LOOPX_REPO=selected_repo, LOOPX_REF=selected_ref, LOOPX_UPDATE_CHANNEL=selected_ref, LOOPX_UPDATE_EVENTS="1")
+        result, _ = install_artifact(manifest, env=env,
+            timeout_seconds=int(os.environ.get("LOOPX_INSTALLER_TIMEOUT_SECONDS", "600")), progress=report)
+        print(result.stdout, end="")
+        print(result.stderr, end="", file=sys.stderr)
+        return result.returncode
+    except (OSError, ValueError) as exc:
+        print(f"LoopX runtime update failed: {exc}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/loopx/cli_commands/update_operation.py b/loopx/cli_commands/update_operation.py
new file mode 100644
index 0000000000..3c7f10d0af
--- /dev/null
+++ b/loopx/cli_commands/update_operation.py
@@ -0,0 +1,183 @@
+"""Private observation/worker IO for the existing update command.
+
+An observer leaving does not cancel an installer. A kernel lock, held by the
+worker (not the observer), prevents a reconnect from launching another update.
+No Goal/Agent state is copied here. Installation decisions remain in self_update.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import sys
+import time
+from uuid import uuid4
+
+from ..file_lock import exclusive_file_lock, process_is_alive
+from .update_artifact import atomic_json
+
+
+def operation_root() -> Path:
+    root = Path(os.environ.get("LOOPX_UPDATE_STATE_DIR") or Path.home() / ".local/state/loopx/updates")
+    root.mkdir(parents=True, exist_ok=True, mode=0o700)
+    root.chmod(0o700)
+    return root
+
+
+def installation_key() -> str:
+    # Two registries updating the same executable must still share one lock.
+    from ..self_update import _user_loopx_bin
+    package_root = Path(__file__).resolve().parents[2]
+    if (package_root / "release.json").is_file():
+        entry = Path(os.environ.get("LOOPX_BIN_DIR") or _user_loopx_bin().parent) / "loopx"
+        identity = str(entry.absolute())
+    else:
+        identity = str(Path(sys.executable).absolute()) + ":" + str(package_root)
+    return hashlib.sha256(identity.encode()).hexdigest()[:24]
+
+
+def read_operation(operation_id: str | None = None) -> dict:
+    root = operation_root()
+    if operation_id is None:
+        try:
+            operation_id = json.loads((root / f"{installation_key()}.active.json").read_text())["operation_id"]
+        except (OSError, ValueError, KeyError):
+            return {"ok": True, "status": "none", "operation_id": None}
+    if not re.fullmatch(r"[a-f0-9]{32}", operation_id):
+        raise ValueError("Invalid update operation id")
+    folder = root / operation_id
+    try:
+        state = json.loads((folder / "status.json").read_text())
+    except (OSError, ValueError):
+        state = {"operation_id": operation_id, "status": "unavailable", "ok": False}
+    result_file = folder / "result.json"
+    if result_file.is_file():
+        try:
+            result = json.loads(result_file.read_text())
+            state.update(status="succeeded" if result.get("ok") else "failed", ok=bool(result.get("ok")), result=result)
+        except (OSError, ValueError):
+            state.update(status="interrupted", ok=False)
+    elif state.get("status") in {"running", "starting"}:
+        # PID alone is not sufficient proof of work after a reboot/PID reuse.
+        try:
+            worker = json.loads((folder / "worker.json").read_text())
+        except (OSError, ValueError):
+            worker = {}
+        observed = process_is_alive(state.get("pid") or worker.get("pid")) and time.time() - state.get("updated_at", 0) < 30
+        state.update(ok=observed, status="running" if observed else "interrupted")
+    return state
+
+
+def _argv(args: argparse.Namespace, registry_path: Path, registry_was_supplied: bool) -> list[str]:
+    argv = ["--format", "json"]
+    if registry_was_supplied:
+        argv += ["--registry", str(registry_path)]
+    if args.runtime_root:
+        argv += ["--runtime-root", str(args.runtime_root)]
+    argv += [args.command, "apply", "--timeout-seconds", str(args.timeout_seconds)]
+    for name in ("repo", "ref", "archive_url"):
+        if getattr(args, name, None):
+            argv += ["--" + name.replace("_", "-"), str(getattr(args, name))]
+    for name in ("source", "allow_downgrade"):
+        if getattr(args, name, False):
+            argv.append("--" + name.replace("_", "-"))
+    return argv
+
+
+def start_or_observe(args: argparse.Namespace, registry_path: Path, *, registry_was_supplied: bool = False) -> dict:
+    root = operation_root()
+    key = installation_key()
+    with exclusive_file_lock(root / (key + ".launch"), timeout_seconds=5, operation="update_launch"):
+        current = read_operation()
+        if current.get("status") == "running":
+            return dict(current, reattached=True)
+        try:
+            with exclusive_file_lock(root / (key + ".worker"), timeout_seconds=0, operation="update_probe"):
+                pass
+        except TimeoutError:
+            # A stale observation is not permission to replay an installation.
+            return dict(current, ok=True, status="running", reattached=True,
+                observation_unavailable=True)
+        operation_id = uuid4().hex
+        folder = root / operation_id
+        folder.mkdir(mode=0o700)
+        atomic_json(folder / "request.json", {"argv": _argv(args, registry_path, registry_was_supplied), "installation_key": key})
+        atomic_json(folder / "status.json", {"schema_version": "loopx_update_progress_v1",
+            "operation_id": operation_id, "status": "starting", "stage": "resolve", "updated_at": time.time()})
+        # Separate process/session and files survive an Agent, terminal or HTTP
+        # observer disconnect. No pipe backpressure can block the installation.
+        with (folder / "worker.log").open("wb") as log:
+            os.chmod(folder / "worker.log", 0o600)
+            options = {"start_new_session": True} if os.name != "nt" else {
+                "creationflags": subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.DETACHED_PROCESS}
+            child = subprocess.Popen([sys.executable, "-m", __name__, "--worker", operation_id],
+                stdin=subprocess.DEVNULL, stdout=log, stderr=log, **options)
+        atomic_json(root / f"{key}.active.json", {"operation_id": operation_id, "pid": child.pid})
+        atomic_json(folder / "worker.json", {"pid": child.pid})
+        return {"ok": True, "status": "running", "stage": "resolve", "operation_id": operation_id,
+            "pid": child.pid, "reattached": False}
+
+
+def observe(operation_id: str, *, json_output: bool = False) -> dict:
+    previous = None
+    while True:
+        state = read_operation(operation_id)
+        marker = (state.get("updated_at"), state.get("status"))
+        if marker != previous:
+            try:
+                if json_output:
+                    print(json.dumps({k: v for k, v in state.items() if k != "result"}), file=sys.stderr, flush=True)
+                else:
+                    received, total = state.get("received"), state.get("total")
+                    size = f" · {received}/{total if total is not None else '?'} bytes" if received is not None else ""
+                    waiting = " · no new progress for 10s" if state.get("stalled") else ""
+                    print(f"LoopX [{operation_id[:8]}]: {state.get('stage', state['status'])}{size}{waiting}", file=sys.stderr, flush=True)
+            except (OSError, BrokenPipeError):
+                return {"ok": True, "operation_id": operation_id, "status": "running", "observer_disconnected": True}
+            previous = marker
+        if state.get("status") not in {"running", "starting"}:
+            result = state.get("result") or {"ok": False, "error": "Update worker interrupted; inspect this operation before retrying"}
+            return dict(result, operation_id=operation_id, operation_status=state["status"])
+        time.sleep(.25)
+
+
+def worker(operation_id: str) -> int:
+    folder = operation_root() / operation_id
+    request = json.loads((folder / "request.json").read_text())
+    # The lifetime lock covers planning, downloading, install and readback.
+    with exclusive_file_lock(operation_root() / (request["installation_key"] + ".worker"),
+                             timeout_seconds=0, operation="update_worker"):
+        os.environ["LOOPX_UPDATE_OPERATION_ID"] = operation_id
+        os.environ["LOOPX_UPDATE_STATUS_PATH"] = str(folder / "status.json")
+        os.environ["LOOPX_UPDATE_WORKER"] = "1"
+        from ..cli import main
+        from contextlib import redirect_stdout
+        import io
+        output = io.StringIO()
+        try:
+            with redirect_stdout(output):
+                code = main(request["argv"])
+            payload = json.loads(output.getvalue())
+        except BaseException:
+            payload = {"ok": False, "changes_applied": False,
+                "error": "Update worker failed; inspect private worker.log and installed doctor before retrying"}
+            code = 1
+        try:
+            progress = json.loads((folder / "status.json").read_text())
+            progress.update(status="succeeded" if payload.get("ok") else "failed", updated_at=time.time())
+            atomic_json(folder / "status.json", progress)
+        except (OSError, ValueError):
+            pass
+        atomic_json(folder / "result.json", payload)
+        return code or 0
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--worker", required=True)
+    raise SystemExit(worker(parser.parse_args().worker))
diff --git a/loopx/cli_commands/update_source.py b/loopx/cli_commands/update_source.py
new file mode 100644
index 0000000000..4112c81bb7
--- /dev/null
+++ b/loopx/cli_commands/update_source.py
@@ -0,0 +1,76 @@
+"""Installation-provider target resolution before the existing activation path.
+
+Kept in the Python bootstrap/OS provider: upgrading or repairing an installation
+must not first require a working Node/control-plane runtime. This extends the
+existing self_update owner and does not add Goal lifecycle vocabulary.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from .update_artifact import resolve_artifact, target_direction
+from ..activation_qualification import runtime_activation_qualification
+
+
+def prepare_target(payload: dict[str, Any], *, freshness: dict, resolve_remote: bool,
+                   allow_downgrade: bool) -> dict[str, Any]:
+    if payload["install_lifecycle"]["owner"] != "loopx_release_snapshot" or not payload["install_lifecycle"]["loopx_apply_supported"]:
+        return payload
+    source = payload["source"]
+    target = None
+    reason = None
+    if source["distribution"] == "runtime":
+        if resolve_remote:
+            try:
+                artifact = resolve_artifact(source["repo"], source["ref"])
+                source["artifact"] = artifact
+                target = artifact["source_revision"]
+                version = artifact["package_version"]
+                payload["source_version_check"].update(attempted=True, status="available",
+                    version=version, version_tag="v" + version,
+                    matches_current=version == freshness.get("current_version"),
+                    reason="Version and revision come from the same qualified runtime metadata")
+            except ValueError as exc:
+                reason = str(exc)
+        else:
+            source["target_status"] = "not_resolved"
+    else:
+        check = payload.get("source_commit_check", {})
+        target = check.get("target_commit")
+        if not target and not source.get("explicit_archive") and resolve_remote:
+            reason = "Source target could not be locked; retry online or select a full commit SHA"
+    source["resolved_revision"] = target
+    current = freshness.get("manifest_source_git_commit")
+    direction = target_direction(source["repo"], current, target) if target and resolve_remote and (
+        source["distribution"] == "runtime" or payload["execute_requested"]) else (
+        "identical" if current and current == target else "unknown")
+    source["direction"] = direction
+    source["allow_downgrade"] = allow_downgrade
+    if target:
+        source["target_status"] = "resolved"
+        payload["source_commit_check"].update(status="available", target_commit=target,
+            matches_current=current == target,
+            reason="Target revision resolved before installation")
+        payload["runtime_activation_qualification"] = runtime_activation_qualification(
+            install_freshness=freshness, source=source, resolved_source_commit=target,
+            selected_source_unresolved=False)
+    # An explicit arbitrary archive remains a developer compatibility path.
+    # Never infer direction from SHA ordering, release age or UI timestamps.
+    if target and direction not in {"ahead", "identical"} and not allow_downgrade and (
+        source["distribution"] == "runtime" or payload["execute_requested"]):
+        reason = "Target is older, diverged or its lineage is unverified; inspect it and explicitly use --allow-downgrade if intended"
+    if resolve_remote and reason:
+        source["target_status"] = "blocked"
+        payload["plan"].update(apply_supported=False, blocked_reason=reason, mutates_release_install=False)
+        payload["recommended_action"] = reason
+        payload["error"] = reason
+        payload["ok"] = not payload["execute_requested"]
+        payload["next_action"] = {"kind": "review_update_target", "mutating": False,
+            "requires_explicit_approval": False, "command": None, "reason": reason}
+    elif target and source["distribution"] == "runtime":
+        recommendation = "Target revision is already installed; apply revalidates it" if current == target else "Prebuilt target resolved; review the installation and activation result"
+        payload["recommended_action"] = recommendation
+        payload["next_action"]["reason"] = recommendation
+    payload["plan"]["distribution"] = source["distribution"]
+    return payload
diff --git a/loopx/release_manifest.py b/loopx/release_manifest.py
index bc912dd104..b212a39579 100644
--- a/loopx/release_manifest.py
+++ b/loopx/release_manifest.py
@@ -155,7 +155,7 @@ def build_release_manifest(
     archive_url = source_env.get("LOOPX_ARCHIVE_URL") or source_metadata.get("archive_url")
     archive_sha256 = source_env.get("LOOPX_ARCHIVE_SHA256") or source_metadata.get("archive_sha256")
     repo = source_env.get("LOOPX_REPO") or source_metadata.get("repo")
-    ref = source_env.get("LOOPX_REF") or source_metadata.get("ref")
+    ref = source_env.get("LOOPX_UPDATE_CHANNEL") or source_env.get("LOOPX_REF") or source_metadata.get("ref")
     source_kind = "github_archive" if archive_url else (source_metadata.get("kind") or "local_checkout")
     promotion_mode = source_env.get("LOOPX_PROMOTION_MODE") or source_metadata.get("promotion_mode")
     skills_root = release_root / "skills"
diff --git a/loopx/self_update.py b/loopx/self_update.py
index cc4d9db19e..c340cf5272 100644
--- a/loopx/self_update.py
+++ b/loopx/self_update.py
@@ -94,10 +94,14 @@ def _source_config(
     repo: str | None,
     ref: str | None,
     archive_url: str | None,
+    installed: dict[str, Any] | None = None,
+    build_from_source: bool = False,
 ) -> dict[str, Any]:
-    selected_repo = repo or os.environ.get("LOOPX_REPO") or DEFAULT_UPDATE_REPO
+    installed = installed or {}
+    selected_repo = repo or os.environ.get("LOOPX_REPO") or installed.get("manifest_source_repo") or DEFAULT_UPDATE_REPO
     ref_override = ref or os.environ.get("LOOPX_REF")
-    selected_ref = ref_override or DEFAULT_UPDATE_REF
+    inherited_ref = installed.get("manifest_source_ref")
+    selected_ref = ref_override or inherited_ref or DEFAULT_UPDATE_REF
     archive_url_override = archive_url or os.environ.get("LOOPX_ARCHIVE_URL")
     selected_archive_url = archive_url_override
     if not selected_archive_url:
@@ -114,9 +118,11 @@ def _source_config(
         "channel": channel,
         "repo": selected_repo,
         "ref": selected_ref,
-        "ref_source": "default_stable" if not ref_override else "override",
+        "ref_source": "override" if ref_override else "installed_manifest" if inherited_ref else "default_stable",
         "archive_url": selected_archive_url,
         "installer_url": NO_CLONE_INSTALL_URL,
+        "distribution": "source" if build_from_source or archive_url_override else "runtime",
+        "explicit_archive": bool(archive_url_override),
     }
 
 
@@ -153,7 +159,10 @@ def _installer_env_for_source(
             )
         env["LOOPX_PYTHON"] = persisted_python
     env["LOOPX_REPO"] = str(source.get("repo") or DEFAULT_UPDATE_REPO)
-    env["LOOPX_REF"] = str(source.get("ref") or DEFAULT_UPDATE_REF)
+    env["LOOPX_REF"] = str(source.get("resolved_revision") or source.get("ref") or DEFAULT_UPDATE_REF)
+    env["LOOPX_UPDATE_CHANNEL"] = str(source.get("ref") or DEFAULT_UPDATE_REF)
+    env["LOOPX_BUILD_FROM_SOURCE"] = "1" if source.get("distribution", "source") == "source" else "0"
+    env["LOOPX_UPDATE_EVENTS"] = "1"
     if source.get("channel") == "github_archive_url_override" and source.get(
         "archive_url"
     ):
@@ -472,6 +481,10 @@ def _update_action_command(action: UpdateAction, source: dict[str, Any]) -> str:
         parts.extend(["--ref", str(source.get("ref") or DEFAULT_UPDATE_REF)])
         if source.get("channel") == "github_archive_url_override":
             parts.extend(["--archive-url", str(source.get("archive_url") or "")])
+    if source.get("distribution") == "source" and not source.get("explicit_archive"):
+        parts.append("--source")
+    if source.get("allow_downgrade"):
+        parts.append("--allow-downgrade")
     return shlex.join(parts)
 
 
@@ -640,6 +653,9 @@ def build_update_plan(
     doctor_payload: dict[str, Any] | None = None,
     registry_path: Path | None = None,
     runtime_root: str | None = None,
+    build_from_source: bool = False,
+    allow_downgrade: bool = False,
+    resolve_remote: bool | None = None,
 ) -> dict[str, Any]:
     requested_action = resolve_update_action(
         action,
@@ -667,8 +683,12 @@ def build_update_plan(
         if isinstance(release_manifest.get("manifest"), dict)
         else {}
     )
-    source = _source_config(repo=repo, ref=ref, archive_url=archive_url)
     lifecycle = _install_lifecycle(doctor)
+    source = _source_config(repo=repo, ref=ref, archive_url=archive_url,
+        installed=install_freshness if lifecycle["owner"] == "loopx_release_snapshot" else None,
+        build_from_source=build_from_source)
+    source["allow_downgrade"] = allow_downgrade
+    remote = doctor_payload is None if resolve_remote is None else resolve_remote
     archive_install_command = _command_for_source(source)
     dry_run = not execute
     path = doctor.get("path") if isinstance(doctor.get("path"), dict) else {}
@@ -676,7 +696,7 @@ def build_update_plan(
     current_version = install_freshness.get("current_version")
     source_version_check = (
         _source_version_check(source)
-        if check_only and lifecycle["owner"] == "loopx_release_snapshot"
+        if check_only and lifecycle["owner"] == "loopx_release_snapshot" and source["distribution"] == "source"
         else {
             "schema_version": SOURCE_VERSION_CHECK_SCHEMA_VERSION,
             "attempted": False,
@@ -707,9 +727,9 @@ def build_update_plan(
         _source_commit_check(
             source,
             installed_commit=install_freshness.get("manifest_source_git_commit"),
-            allow_remote=doctor_payload is None,
+            allow_remote=remote,
         )
-        if not execute and lifecycle["owner"] == "loopx_release_snapshot"
+        if lifecycle["owner"] == "loopx_release_snapshot" and source["distribution"] == "source"
         else {"schema_version": SOURCE_COMMIT_CHECK_SCHEMA_VERSION,
               "attempted": False, "status": "not_requested",
               "installed_commit": install_freshness.get("manifest_source_git_commit"),
@@ -943,6 +963,13 @@ def build_update_plan(
         ),
         "recommended_action": recommended_action,
     }
+    from .cli_commands.update_source import prepare_target
+    payload = prepare_target(payload, freshness=install_freshness,
+        resolve_remote=remote, allow_downgrade=allow_downgrade)
+    if lifecycle["owner"] == "python_package_manager" and (ref or build_from_source or archive_url):
+        reason = "The Python package manager owns this installation; choose main/source in a separate snapshot installation"
+        payload.update(ok=not execute, error=reason, recommended_action=reason)
+        payload["plan"].update(apply_supported=False, blocked_reason=reason, mutates_release_install=False)
     return (_windows_console_update_refusal(payload) if execute else None) or payload
 
 
@@ -1152,6 +1179,8 @@ def execute_update_plan(
     registry_path: Path | None = None, runtime_root: str | None = None,
     progress: Callable[[str], None] = lambda _stage: None,
 ) -> dict[str, Any]:
+    if not payload.get("plan", {}).get("apply_supported", True):
+        return dict(payload, ok=False, changes_applied=False)
     route_arguments = []
     if registry_path is not None:
         route_arguments += ["--registry", str(registry_path.resolve())]
@@ -1211,12 +1240,16 @@ def execute_update_plan(
         env["LOOPX_REGISTRY"] = str(registry_path.resolve())
     if runtime_root is not None:
         env["LOOPX_RUNTIME_ROOT"] = runtime_root
-    install_result, download_observation = run_archive_installer(
-        installer_url,
-        env=env,
-        timeout_seconds=timeout_seconds,
-        progress=progress,
-    )
+    if source.get("distribution") == "runtime":
+        from .cli_commands.update_artifact import install_artifact
+        if not source.get("artifact"):
+            return dict(payload, ok=False, changes_applied=False,
+                error="No resolved runtime artifact; rebuild the update plan online before applying")
+        install_result, download_observation = install_artifact(source["artifact"], env=env,
+            timeout_seconds=timeout_seconds, progress=progress)
+    else:
+        install_result, download_observation = run_archive_installer(
+            installer_url, env=env, timeout_seconds=timeout_seconds, progress=progress)
     # Match install-local.sh's LOOPX_BIN_DIR override, including its empty-value
     # default. Read back the installation that this update actually replaced.
     loopx_bin = Path(env.get("LOOPX_BIN_DIR") or _user_loopx_bin().parent) / "loopx"
@@ -1238,8 +1271,21 @@ def execute_update_plan(
         "doctor_stderr_tail": doctor_result.stderr[-2000:],
         "installed_runtime_identity": _installed_readback(doctor_result) if doctor_result.returncode == 0 else {},
     }
+    try:
+        installed_freshness = json.loads(doctor_result.stdout).get("install_freshness", {})
+    except (ValueError, AttributeError):
+        installed_freshness = {}
+    if not isinstance(installed_freshness, dict):
+        installed_freshness = {}
+    # A host-material step can fail after a healthy candidate became active.
+    # Do not report "nothing changed" merely from the installer's exit code.
+    activation_observed = bool(doctor_result.returncode == 0 and source.get("resolved_revision")
+        and installed_freshness.get("manifest_source_git_commit") == source["resolved_revision"]
+        and installed_freshness.get("release_id")
+        and installed_freshness["release_id"] != payload.get("current", {}).get("release_id"))
+    execution["activation_observed"] = activation_observed
     extension_doctor_result = None
-    if install_result.returncode == 0 and doctor_result.returncode == 0:
+    if (install_result.returncode == 0 or activation_observed) and doctor_result.returncode == 0:
         progress("Validating enabled extensions")
         extension_doctor_result = subprocess.run(
             [
@@ -1266,15 +1312,15 @@ def execute_update_plan(
         )
     else:
         execution["extension_doctor_status"] = "skipped_release_update_failed"
-    runtime_ready = install_result.returncode == 0 and doctor_result.returncode == 0
+    runtime_ready = (install_result.returncode == 0 or activation_observed) and doctor_result.returncode == 0
     updated = dict(payload)
     updated["execution"] = execution
     updated["ok"] = (
-        runtime_ready
+        install_result.returncode == 0 and runtime_ready
         and extension_doctor_result is not None
         and extension_doctor_result.returncode == 0
     )
-    updated["changes_applied"] = install_result.returncode == 0
+    updated["changes_applied"] = install_result.returncode == 0 or activation_observed
     execution.update(
         restart_services_for_runtime_activation(
             changes_applied=updated["changes_applied"],
diff --git a/loopx/self_update_download.py b/loopx/self_update_download.py
index beacd2f185..a5651f0b92 100644
--- a/loopx/self_update_download.py
+++ b/loopx/self_update_download.py
@@ -76,14 +76,13 @@ def run_archive_installer(
                     max(1, int(remaining))
                 )
                 try:
-                    return subprocess.run(
-                        ["bash", str(script)],
-                        check=False,
-                        text=True, encoding="utf-8", errors="replace",
-                        capture_output=True,
-                        env=installer_env,
-                        timeout=remaining,
-                    ), observation
+                    if installer_env.get("LOOPX_UPDATE_EVENTS") == "1":
+                        from .cli_commands.update_artifact import run_installer
+                        return run_installer(["bash", str(script)], env=installer_env,
+                            timeout=remaining, progress=progress), observation
+                    return subprocess.run(["bash", str(script)], check=False, text=True,
+                        encoding="utf-8", errors="replace", capture_output=True,
+                        env=installer_env, timeout=remaining), observation
                 except subprocess.TimeoutExpired:
                     return subprocess.CompletedProcess(
                         [],
diff --git a/scripts/install-from-github.sh b/scripts/install-from-github.sh
index 1b595fd1d6..03fa981b2c 100755
--- a/scripts/install-from-github.sh
+++ b/scripts/install-from-github.sh
@@ -1,7 +1,9 @@
 #!/usr/bin/env bash
 set -euo pipefail
 
+repo_explicit="${LOOPX_REPO:+1}"
 repo="${LOOPX_REPO:-loopx-project/loopx}"
+ref_explicit="${LOOPX_REF:+1}"
 ref="${LOOPX_REF:-stable}"
 archive_url_override="${LOOPX_ARCHIVE_URL:-}"
 archive_url="$archive_url_override"
@@ -38,6 +40,19 @@ cleanup() {
 }
 trap cleanup EXIT
 
+# Default snapshots use a verified, prebuilt runtime. Developer source builds
+# remain explicit; a missing release never silently starts npm or a whole-repo
+# download. The bootstrap is published beside this script over the same TLS
+# origin. Its stdlib transport preserves the operator's proxy environment.
+if [[ "${LOOPX_BUILD_FROM_SOURCE:-0}" != "1" && -z "$archive_url_override" ]]; then
+  export LOOPX_UPDATE_REF_EXPLICIT="${ref_explicit:-0}"
+  export LOOPX_UPDATE_REPO_EXPLICIT="${repo_explicit:-0}"
+  curl -fsSL --connect-timeout 10 --max-time 45 --retry 2 --retry-max-time 60 \
+    https://loopx-project.github.io/loopx/update-artifact.py -o "$tmp_dir/update-artifact.py"
+  "$python_bin" "$tmp_dir/update-artifact.py"
+  exit $?
+fi
+
 if [[ -z "$archive_url" ]]; then
   if [[ ! "$repo" =~ ^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$ ]]; then
     echo "loopx installer error: LOOPX_REPO must use GitHub owner/name syntax" >&2
@@ -125,6 +140,29 @@ archive_attempts_completed=0
 archive_downloaded=0
 last_curl_code=28
 last_http_status=0
+download_source_archive() {
+  local curl_pid curl_result
+  local -a arguments=(--silent --show-error --fail --location
+    --connect-timeout 10 --max-time "$attempt_timeout"
+    --continue-at - --write-out '%{http_code}' "$archive_url" -o "$archive_path")
+  if [[ "${LOOPX_UPDATE_EVENTS:-0}" != "1" ]]; then
+    curl "${arguments[@]}"
+    return $?
+  fi
+  curl "${arguments[@]}" >"$tmp_dir/http-status" &
+  curl_pid=$!
+  while kill -0 "$curl_pid" 2>/dev/null; do
+    if [[ -f "$archive_path" ]]; then
+      local received
+      received="$(wc -c <"$archive_path" | tr -d ' ')"
+      printf 'LOOPX_UPDATE_EVENT {"stage":"download","received":%s,"total":null}\n' "$received" >&2
+    fi
+    sleep 2
+  done
+  if wait "$curl_pid"; then curl_result=0; else curl_result=$?; fi
+  cat "$tmp_dir/http-status"
+  return "$curl_result"
+}
 while [[ "$archive_attempt" -le "$archive_max_attempts" ]]; do
   remaining=$((archive_deadline - SECONDS))
   if [[ "$remaining" -le 0 ]]; then
@@ -145,10 +183,7 @@ while [[ "$archive_attempt" -le "$archive_max_attempts" ]]; do
   if [[ -s "$archive_path" ]]; then
     resuming_archive=1
   fi
-  if http_status="$(curl --silent --show-error --fail --location \
-    --connect-timeout 10 --max-time "$attempt_timeout" \
-    --continue-at - --write-out '%{http_code}' \
-    "$archive_url" -o "$archive_path")"; then
+  if http_status="$(download_source_archive)"; then
     # curl can treat 416 as an already-complete resumed transfer. The local
     # partial file is not proof of that; recover by downloading it afresh.
     if [[ "$resuming_archive" -eq 1 && "$http_status" == 416 ]]; then
diff --git a/scripts/install-local.sh b/scripts/install-local.sh
index 623bd86358..9681fbf640 100755
--- a/scripts/install-local.sh
+++ b/scripts/install-local.sh
@@ -76,6 +76,10 @@ release_dir=""
 release_tmp=""
 install_lock=""
 install_lock_owned=0
+previous_loopx=""
+previous_rrule=""
+activation_started=0
+activation_verified=0
 skill_install_lock=""
 skill_install_lock_owned=0
 legacy_line=""
@@ -109,6 +113,18 @@ PY
 }
 
 cleanup_install_lock() {
+  local result=$?
+  if [[ "$result" != "0" && "$activation_started" == "1" && "$activation_verified" != "1" ]]; then
+    # Restore executable links owned by this activation only. Data migration
+    # receipts/backups are independent and are deliberately never rolled back.
+    if [[ -n "$previous_loopx" && "$(readlink "$bin_dir/loopx" 2>/dev/null || true)" == "$release_dir/scripts/loopx" ]]; then
+      install_symlink "$previous_loopx" "$bin_dir/loopx" || true
+      echo "loopx installer: restored the previous executable after activation failure; migration backups retained" >&2
+    fi
+    if [[ -n "$previous_rrule" && "$(readlink "$bin_dir/loopx-apply-rrule" 2>/dev/null || true)" == "$release_dir/scripts/loopx-apply-rrule" ]]; then
+      install_symlink "$previous_rrule" "$bin_dir/loopx-apply-rrule" || true
+    fi
+  fi
   if [[ -n "$release_tmp" && -d "$release_tmp" ]]; then
     rm -rf "$release_tmp"
   fi
@@ -120,6 +136,12 @@ cleanup_install_lock() {
   fi
 }
 
+update_stage() {
+  if [[ "${LOOPX_UPDATE_EVENTS:-0}" == "1" ]]; then
+    printf 'LOOPX_UPDATE_EVENT {"stage":"%s"}\n' "$1" >&2
+  fi
+}
+
 trap cleanup_install_lock EXIT
 
 run_under_install_guard() {
@@ -769,6 +791,11 @@ if [[ "$promote_default" == "1" ]]; then
   mkdir -p "$releases_dir"
   run_under_install_guard "$@"
 fi
+update_stage frontend
+if [[ "${LOOPX_PREBUILT_RUNTIME:-0}" == "1" ]]; then
+  # A release may not silently turn a missing/corrupt bundle into a local build.
+  "${LOOPX_PYTHON:-python3}" "$repo_root/scripts/chat_bundle.py" verify --source
+fi
 chat_bundle_args=(ensure)
 if [[ -L "$bin_dir/loopx" ]]; then
   previous_chat_assets="$("${LOOPX_PYTHON:-python3}" - "$bin_dir/loopx" <<'PYTHON'
@@ -819,6 +846,7 @@ if [[ -z "$legacy_line" ]]; then
   legacy_line="- legacy command disabled: not present"
 fi
 reserve_release_id
+update_stage copy
 copy_path "$repo_root/loopx" "$release_tmp/loopx"
 copy_path "$repo_root/scripts" "$release_tmp/scripts"
 copy_path "$repo_root/skills" "$release_tmp/skills"
@@ -829,6 +857,8 @@ copy_apps "$repo_root/apps" "$release_tmp/apps"
 copy_path "$repo_root/.github" "$release_tmp/.github"
 copy_path "$repo_root/README.md" "$release_tmp/README.md"
 copy_path "$repo_root/LICENSE" "$release_tmp/LICENSE"
+copy_path "$repo_root/NOTICE" "$release_tmp/NOTICE"
+copy_path "$repo_root/LICENSE-MIT" "$release_tmp/LICENSE-MIT"
 copy_path "$repo_root/pyproject.toml" "$release_tmp/pyproject.toml"
 copy_path "$repo_root/setup.py" "$release_tmp/setup.py"
 copy_path "$repo_root/MANIFEST.in" "$release_tmp/MANIFEST.in"
@@ -849,6 +879,7 @@ PYTHONPATH="$release_tmp" "${LOOPX_PYTHON:-python3}" \
 chmod +x "$release_tmp/scripts/loopx" "$release_tmp/scripts/loopx-apply-rrule"
 mv "$release_tmp" "$release_dir"
 release_tmp=""
+update_stage validate
 if ! validate_release_candidate "$release_dir"; then
   rm -rf "$release_dir"
   exit 1
@@ -859,10 +890,15 @@ if ! preflight_workflow_skills "$release_dir/skills" "$release_dir" "$bin_dir/lo
 fi
 # Data upgrade is a separate, resumable operation. Never delete its backups or
 # roll migrated stores back merely because a later launcher/skill step fails.
+update_stage migrate
 if ! "$release_dir/scripts/loopx" --format json authority-archive upgrade --all-known --execute; then
   echo "loopx installer error: authority format upgrade failed; backups are retained. Retry with this candidate before activation." >&2
   exit 1
 fi
+update_stage activate
+previous_loopx="$(readlink "$bin_dir/loopx" 2>/dev/null || true)"
+previous_rrule="$(readlink "$bin_dir/loopx-apply-rrule" 2>/dev/null || true)"
+activation_started=1
 install_symlink "$release_dir/scripts/loopx" "$bin_dir/loopx"
 install_symlink "$release_dir/scripts/loopx-apply-rrule" "$bin_dir/loopx-apply-rrule"
 verify_default_promotion
@@ -923,6 +959,7 @@ fi
 
 export PATH="$bin_dir:$PATH"
 "$bin_dir/loopx" doctor --installation-only >/dev/null
+activation_verified=1
 if [[ "${LOOPX_INSTALL_REVALIDATE_EXTENSIONS:-1}" != "0" ]] && ! "$bin_dir/loopx" --format json extension doctor --all-enabled --execute >/dev/null; then
   echo "loopx installer warning: one or more enabled extensions failed post-install doctor revalidation" >&2
   echo "Run 'loopx extension doctor --all-enabled --execute' after repairing the provider." >&2
@@ -931,6 +968,7 @@ if [[ "$install_canary" != "0" ]]; then
   "$bin_dir/loopx-canary" doctor --installation-only >/dev/null
 fi
 
+update_stage skills
 install_workflow_skills "$release_dir/skills" "$release_dir" "$bin_dir/loopx"
 
 slash_line="- slash commands: skipped"
diff --git a/scripts/qualify_runtime_artifact.py b/scripts/qualify_runtime_artifact.py
new file mode 100644
index 0000000000..45b920038d
--- /dev/null
+++ b/scripts/qualify_runtime_artifact.py
@@ -0,0 +1,133 @@
+#!/usr/bin/env python3
+"""Install and upgrade a runtime archive in a disposable, isolated environment."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+import subprocess
+import shutil
+import sys
+import tempfile
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from loopx.cli_commands.update_artifact import extract_artifact, file_hash, validate_manifest
+
+
+def qualify(directory: Path) -> dict:
+    manifest = validate_manifest(json.loads((directory / "loopx-runtime.json").read_text()))
+    archive = directory / f"loopx-runtime-{manifest['source_revision']}.tar.gz"
+    assert archive.stat().st_size == manifest["size"] and file_hash(archive) == manifest["sha256"]
+    with tempfile.TemporaryDirectory(prefix="loopx-runtime-qualification-") as raw:
+        scratch = Path(raw)
+        source = scratch / "source"
+        source.mkdir()
+        extract_artifact(archive, source)
+        env = dict(os.environ)
+        # No developer's runtime, home, skills or host config may participate.
+        for key in tuple(env):
+            if key.startswith("LOOPX_") or key in {"CODEX_HOME", "PYTHONPATH", "GIT_DIR", "GIT_WORK_TREE"}:
+                env.pop(key)
+        env.update(HOME=str(scratch / "home"), CODEX_HOME=str(scratch / "home/codex"),
+            XDG_CONFIG_HOME=str(scratch / "home/config"), XDG_DATA_HOME=str(scratch / "home/data"),
+            XDG_CACHE_HOME=str(scratch / "home/cache"), XDG_STATE_HOME=str(scratch / "home/state"),
+            LOOPX_BIN_DIR=str(scratch / "bin"), LOOPX_RELEASES_DIR=str(scratch / "releases"),
+            LOOPX_MAN_ROOT=str(scratch / "man"), LOOPX_SKILLS_DIR=str(scratch / "skills"),
+            LOOPX_REGISTRY=str(scratch / "registry.json"), LOOPX_RUNTIME_ROOT=str(scratch / "runtime"),
+            LOOPX_INSTALL_CANARY="0", LOOPX_PROMOTE_DEFAULT="1", LOOPX_PREBUILT_RUNTIME="1",
+            LOOPX_INSTALL_SLASH_COMMANDS="0", LOOPX_INSTALL_REVALIDATE_EXTENSIONS="0",
+            LOOPX_RESOLVED_SOURCE_GIT_COMMIT=manifest["source_revision"], LOOPX_REF=manifest["channel"],
+            LOOPX_PYTHON=sys.executable, LOOPX_USAGE_PING="0", LOOPX_UPDATE_EVENTS="1")
+        (scratch / "home").mkdir()
+        tools = scratch / "tools"
+        tools.mkdir()
+        for name in ("npm", "npx"):
+            command = tools / name
+            command.write_text("#!/bin/sh\necho 'Frontend build tools must not run during runtime installation' >&2\nexit 91\n")
+            command.chmod(0o755)
+        env["PATH"] = str(tools) + os.pathsep + env.get("PATH", "")
+        results = []
+        for _ in range(2):
+            result = subprocess.run(["bash", str(source / "scripts/install-local.sh")], env=env,
+                text=True, capture_output=True, timeout=180)
+            if result.returncode:
+                raise RuntimeError("Isolated runtime install failed:\n" + result.stdout[-4000:] + result.stderr[-4000:])
+            results.append(result.returncode)
+        cli = scratch / "bin/loopx"
+        def read(*args):
+            result = subprocess.run([str(cli), "--format", "json", *args], env=env,
+                text=True, capture_output=True, timeout=60, check=True)
+            return json.loads(result.stdout)
+        doctor = read("doctor", "--installation-only")
+        skills = read("workflow-skills", "--skills-dir", str(scratch / "skills"))
+        if not doctor.get("ok") or not skills.get("ok"):
+            raise RuntimeError("Installed runtime/skills readback failed")
+        release = cli.resolve().parents[1]
+        for name in ("LICENSE", "NOTICE", "LICENSE-MIT"):
+            assert (release / name).read_bytes() == (source / name).read_bytes()
+        source_record = json.loads((release / "release.json").read_text())["source"]
+        assert source_record["git_commit"] == manifest["source_revision"]
+        assert source_record["ref"] == manifest["channel"]
+        # Every preserved skill reference and TS input is supplied by the same
+        # committed archive; validate the exact Chat bundle through its owner.
+        subprocess.run([sys.executable, str(release / "scripts/chat_bundle.py"), "verify", "--source"],
+            env=env, check=True, stdout=subprocess.DEVNULL)
+        bundle = json.loads((release / "loopx/web/chat/bundle-manifest.json").read_text())
+        assert bundle["source_revision"] == manifest["source_revision"]
+        # Exercise the installed delivery owner, not a development file server.
+        subprocess.run([sys.executable, "-c", """
+import hashlib
+import json
+import threading
+from urllib.request import urlopen
+from loopx.chat_server import ChatHTTPServer, ChatRequestHandler, default_chat_assets_dir
+assets = default_chat_assets_dir()
+manifest = json.loads((assets / 'bundle-manifest.json').read_text())
+server = ChatHTTPServer(('127.0.0.1', 0), ChatRequestHandler)
+server.assets_dir, server.verbose = assets, False
+worker = threading.Thread(target=server.serve_forever, daemon=True)
+worker.start()
+try:
+    for name, digest in manifest['files'].items():
+        with urlopen(f'http://127.0.0.1:{server.server_port}/chat/{name}', timeout=10) as response:
+            assert response.status == 200
+            assert hashlib.sha256(response.read()).hexdigest() == digest, name
+finally:
+    server.shutdown()
+    server.server_close()
+    worker.join()
+"""], env={**env, "PYTHONPATH": str(release)}, cwd=release,
+            check=True, timeout=60, stdout=subprocess.DEVNULL)
+        # Inject a post-switch executable failure at the manpage chmod step.
+        # Only a disposable candidate is touched, never a real installation.
+        real_chmod = shutil.which("chmod")
+        failing_chmod = tools / "chmod"
+        failing_chmod.write_text(f"#!{sys.executable}\n" +
+            "import os,subprocess,sys\nfrom pathlib import Path\n" +
+            f"subprocess.run([{real_chmod!r}, *sys.argv[1:]], check=True)\n" +
+            "if sys.argv[1] == '0644':\n" +
+            "    entry = Path(os.environ['LOOPX_BIN_DIR']) / 'loopx'\n" +
+            "    entry.resolve().chmod(0o644)\n")
+        failing_chmod.chmod(0o755)
+        failed = subprocess.run(["bash", str(source / "scripts/install-local.sh")], env=env,
+            text=True, capture_output=True, timeout=180)
+        if failed.returncode == 0 or cli.resolve().parents[1] != release:
+            raise RuntimeError("Activation failure did not preserve the previous executable: " + failed.stderr[-3000:])
+        assert read("doctor", "--installation-only")["ok"]
+        subprocess.run([sys.executable, "-c",
+            "from loopx.control_plane.effect_runtime import restart_effect_runtime; "
+            "restart_effect_runtime(wait_for_exit=True)"],
+            env={**env, "PYTHONPATH": str(release)}, cwd=release,
+            check=True, timeout=15, stdout=subprocess.DEVNULL)
+        return {"ok": True, "source_revision": manifest["source_revision"], "bytes": manifest["size"],
+            "clean_install": results[0] == 0, "upgrade": results[1] == 0,
+            "doctor": True, "skills": True, "frontend_bundle_contract": True, "frontend_http": True,
+            "no_frontend_build_tools": True, "license_notices": True, "activation_failure_restored": True}
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("directory", type=Path)
+    print(json.dumps(qualify(parser.parse_args().directory)))
diff --git a/scripts/runtime_artifact.py b/scripts/runtime_artifact.py
new file mode 100644
index 0000000000..5b4cac2997
--- /dev/null
+++ b/scripts/runtime_artifact.py
@@ -0,0 +1,128 @@
+#!/usr/bin/env python3
+"""Build audited runtime + optional resources from one committed source revision.
+
+Keep runtime code, all skills/references, text documentation, developer smokes,
+and the exact CI-bound frontend bundle contract. Frontend development sources
+and named presentation media are optional; docs/examples text stays available.
+The archive is installed by the existing snapshot installer, without npm.
+"""
+
+from __future__ import annotations
+
+import argparse
+import gzip
+import hashlib
+import io
+import json
+from pathlib import Path
+import subprocess
+import sys
+import tarfile
+import tomllib
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+from loopx.presentation.chat_bundle import source_digest, validate_bundle
+from loopx.cli_commands.update_artifact import RUNTIME_ARTIFACT_SCHEMA_VERSION, validate_manifest
+
+PATHS = ("loopx", "scripts", "skills", "docs", "man", "examples", "apps/presentation",
+         ".github", "README.md", "LICENSE", "NOTICE", "LICENSE-MIT", "pyproject.toml", "setup.py", "MANIFEST.in")
+# Static product illustrations/browser captures are not used by CLI execution,
+# skills, installer validation, TS runtime or Chat's source/bundle validator.
+# Text/HTML/SVG and any exact frontend input remain in the runtime below.
+OPTIONAL_MEDIA_ROOTS = (
+    "docs/assets/", "docs/development/assets/",
+    "docs/architecture/rfcs/", "docs/reference/assets/", "docs/reference/media/",
+    "examples/personal-workspace-browser/evidence/",
+    "examples/personal-workspace-browser/assets/",
+    "examples/personal-workspace-browser/screenshots/",
+    "examples/collaboration-delivery/screenshots/",
+)
+MEDIA_SUFFIXES = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".mp4", ".mov", ".webm"}
+
+
+def build(root: Path, output: Path, *, repository: str, channel: str) -> dict:
+    revision = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=root, text=True).strip()
+    manifest = validate_bundle(root / "loopx/web/chat", source_root=root)
+    if manifest["source_revision"] != revision:
+        raise ValueError("Frontend must belong to the exact committed source revision")
+    for name, digest in manifest["source_files"].items():
+        committed = subprocess.check_output(["git", "show", f"{revision}:{name}"], cwd=root)
+        if source_digest(Path(name), committed) != digest:
+            raise ValueError(f"Frontend input differs from committed source: {name}")
+    output.mkdir(parents=True, exist_ok=True)
+    filename = f"loopx-runtime-{revision}.tar.gz"
+    optional_name = f"loopx-resources-{revision}.tar.gz"
+    inventory: dict[str, list] = {"runtime": [], "optional_resources": []}
+    version = None
+    # Stream Git, never enumerate the working directory (private files stay out).
+    git = subprocess.Popen(["git", "archive", revision, *PATHS], cwd=root, stdout=subprocess.PIPE)
+    with (
+        (output / filename).open("wb") as runtime_file,
+        (output / optional_name).open("wb") as resources_file,
+        gzip.GzipFile(fileobj=runtime_file, mode="wb", mtime=0, filename="") as runtime_gzip,
+        gzip.GzipFile(fileobj=resources_file, mode="wb", mtime=0, filename="") as resources_gzip,
+        tarfile.open(fileobj=runtime_gzip, mode="w|", format=tarfile.PAX_FORMAT) as runtime,
+        tarfile.open(fileobj=resources_gzip, mode="w|", format=tarfile.PAX_FORMAT) as resources,
+        tarfile.open(fileobj=git.stdout, mode="r|") as source,
+    ):
+        for member in source:
+            if not member.isfile():
+                if not member.isdir():
+                    raise ValueError(f"Non-file runtime entry: {member.name}")
+                continue
+            if member.name.startswith("loopx/web/chat/"):
+                continue
+            content = source.extractfile(member).read()
+            if member.name == "pyproject.toml":
+                version = tomllib.loads(content.decode())["project"]["version"]
+            optional = member.name.startswith("apps/presentation/") or (member.name.startswith(OPTIONAL_MEDIA_ROOTS)
+                and Path(member.name).suffix.lower() in MEDIA_SUFFIXES
+                and member.name not in manifest["source_files"])
+            sink = resources if optional else runtime
+            group = "optional_resources" if optional else "runtime"
+            member.uid = member.gid = member.mtime = 0
+            member.uname = member.gname = ""
+            member.pax_headers = {}
+            sink.addfile(member, io.BytesIO(content))
+            inventory[group].append({"path": member.name, "bytes": len(content),
+                "sha256": hashlib.sha256(content).hexdigest()})
+        for name in sorted({*manifest["files"], "bundle-manifest.json"}):
+            content = (root / "loopx/web/chat" / name).read_bytes()
+            member = tarfile.TarInfo("loopx/web/chat/" + name)
+            member.size, member.mode = len(content), 0o644
+            runtime.addfile(member, io.BytesIO(content))
+            inventory["runtime"].append({"path": member.name, "bytes": len(content),
+                "sha256": hashlib.sha256(content).hexdigest()})
+        identity = json.dumps({"schema_version": RUNTIME_ARTIFACT_SCHEMA_VERSION, "source_revision": revision,
+            "package_version": version}).encode()
+        member = tarfile.TarInfo("runtime-artifact.json")
+        member.size, member.mode = len(identity), 0o644
+        runtime.addfile(member, io.BytesIO(identity))
+        inventory["runtime"].append({"path": member.name, "bytes": len(identity),
+            "sha256": hashlib.sha256(identity).hexdigest()})
+    if git.wait() != 0 or not version:
+        raise ValueError("Git runtime archive failed")
+    tag = "runtime-main" if channel == "main" else "v" + version
+    archive = output / filename
+    result = validate_manifest({"schema_version": RUNTIME_ARTIFACT_SCHEMA_VERSION, "source_revision": revision,
+        "package_version": version, "channel": channel, "size": archive.stat().st_size,
+        "sha256": hashlib.sha256(archive.read_bytes()).hexdigest(),
+        "urls": [f"https://github.com/{repository}/releases/download/{tag}/{filename}"],
+        "resources": {"filename": optional_name, "size": (output / optional_name).stat().st_size,
+            "sha256": hashlib.sha256((output / optional_name).read_bytes()).hexdigest()}})
+    (output / "loopx-runtime.json").write_text(json.dumps(result, indent=2) + "\n")
+    (output / "runtime-inventory.json").write_text(json.dumps(inventory, indent=2) + "\n")
+    print(json.dumps({"revision": revision, "runtime_bytes": result["size"],
+        "optional_bytes": result["resources"]["size"],
+        "runtime_files": len(inventory["runtime"]), "optional_files": len(inventory["optional_resources"])}))
+    return result
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--output", type=Path, default=ROOT / "dist/runtime")
+    parser.add_argument("--repository", default="loopx-project/loopx")
+    parser.add_argument("--channel", choices=("main", "stable"), required=True)
+    args = parser.parse_args()
+    build(ROOT, args.output, repository=args.repository, channel=args.channel)
diff --git a/tests/test_archive_installer_commit_response.py b/tests/test_archive_installer_commit_response.py
index 481d41fbff..f82a640bb5 100644
--- a/tests/test_archive_installer_commit_response.py
+++ b/tests/test_archive_installer_commit_response.py
@@ -94,6 +94,7 @@ def test_invalid_commit_override_precedes_temp_directory_failure(tmp_path):
     source = Path(__file__).resolve().parents[1]
     env = {k: v for k, v in os.environ.items() if not k.startswith("LOOPX_")}
     env.update(
+        LOOPX_BUILD_FROM_SOURCE="1",
         TMPDIR=str(tmp_path / "missing"),
         LOOPX_PYTHON=sys.executable,
         LOOPX_RESOLVED_SOURCE_GIT_COMMIT="not-a-full-sha",
@@ -173,6 +174,7 @@ def test_commit_response_uses_file_transport_and_cleans_up(tmp_path, valid, ref_
     receipt = tmp_path / "receipt"
     env = {k: v for k, v in os.environ.items() if not k.startswith("LOOPX_")}
     env.update(
+        LOOPX_BUILD_FROM_SOURCE="1",
         PATH=f"{binary}{os.pathsep}{env['PATH']}",
         TMPDIR=str(scratch),
         LOOPX_PYTHON=sys.executable,
@@ -247,6 +249,7 @@ def test_archive_download_recovers_after_attempt_timeout(
     receipt = tmp_path / "receipt"
     env = {k: v for k, v in os.environ.items() if not k.startswith("LOOPX_")}
     env.update(
+        LOOPX_BUILD_FROM_SOURCE="1",
         TMPDIR=str(scratch),
         LOOPX_PYTHON=sys.executable,
         LOOPX_ARCHIVE_URL=f"http://127.0.0.1:{server.server_port}/archive.tar.gz",
@@ -302,6 +305,7 @@ def test_archive_timeout_failure_never_extracts_partial_file(tmp_path, resume_st
     scratch.mkdir()
     env = {k: v for k, v in os.environ.items() if not k.startswith("LOOPX_")}
     env.update(
+        LOOPX_BUILD_FROM_SOURCE="1",
         PATH=f"{binary}{os.pathsep}{env['PATH']}",
         TMPDIR=str(scratch),
         LOOPX_PYTHON=sys.executable,
@@ -352,6 +356,7 @@ def test_complete_corrupt_archive_is_rejected_before_extraction(tmp_path):
     marker = tmp_path / "extracted"
     env = {k: v for k, v in os.environ.items() if not k.startswith("LOOPX_")}
     env.update(
+        LOOPX_BUILD_FROM_SOURCE="1",
         PATH=f"{binary}{os.pathsep}{env['PATH']}", TMPDIR=str(scratch),
         LOOPX_PYTHON=sys.executable, LOOPX_ARCHIVE_URL=archive.as_uri(),
         TEST_TAR_MARKER=str(marker),
diff --git a/tests/test_self_update_runtime_activation.py b/tests/test_self_update_runtime_activation.py
index 00c6c04363..7661721ffc 100644
--- a/tests/test_self_update_runtime_activation.py
+++ b/tests/test_self_update_runtime_activation.py
@@ -88,7 +88,7 @@ def doctor_payload(
 
 def build_check(doctor: dict[str, object], *, ref: str = "main") -> dict[str, object]:
     with mock.patch("loopx.self_update.urlopen", return_value=FakeVersionResponse()):
-        return build_update_plan(
+        return build_update_plan(build_from_source=True,
             repo="example/loopx",
             ref=ref,
             check_only=True,
@@ -110,7 +110,7 @@ def response(request: object, **_kwargs: object) -> FakeVersionResponse:
     with mock.patch("loopx.self_update.collect_doctor", return_value=doctor), mock.patch(
         "loopx.self_update.urlopen", side_effect=response
     ):
-        return build_update_plan(
+        return build_update_plan(build_from_source=True,
             repo="example/loopx", ref="main", check_only=True
         )
 
@@ -132,7 +132,7 @@ def test_update_plan_resolves_mutable_ref_commit_before_claiming_no_update(
         ),
     )
 
-    payload = build_update_plan(repo="example/loopx", ref="main", action="plan")
+    payload = build_update_plan(build_from_source=True, repo="example/loopx", ref="main", action="plan")
     commit_check = payload["source_commit_check"]
     assert commit_check["status"] == "available"
     assert commit_check["matches_current"] is False
@@ -156,7 +156,7 @@ def test_update_plan_accepts_exact_commit_under_pinned_install_ref(
         lambda *_args, **_kwargs: FakeVersionResponse(INSTALLED_COMMIT.encode()),
     )
 
-    payload = build_update_plan(repo="example/loopx", ref="main", action="plan")
+    payload = build_update_plan(build_from_source=True, repo="example/loopx", ref="main", action="plan")
     assert payload["source_commit_check"]["matches_current"] is True
     assert payload["runtime_activation_qualification"]["decision"] == "runtime_active"
     assert "no update needed" in payload["recommended_action"]
@@ -174,7 +174,7 @@ def test_update_plan_offline_never_calls_unqualified_source_current(
     monkeypatch.setattr("loopx.self_update.collect_doctor", lambda: doctor)
     monkeypatch.setattr("loopx.self_update.urlopen", lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("offline")))
 
-    payload = build_update_plan(repo="example/loopx", ref="main", action="plan")
+    payload = build_update_plan(build_from_source=True, repo="example/loopx", ref="main", action="plan")
     assert payload["source_commit_check"]["status"] == "unavailable"
     assert payload["runtime_activation_qualification"]["decision"] == "activation_qualification_required"
     assert "no update needed" not in payload["recommended_action"]
@@ -233,7 +233,7 @@ def collect_bounded_doctor() -> dict[str, object]:
     monkeypatch.setattr("loopx.self_update.urlopen", lambda *_args, **_kwargs: FakeVersionResponse())
 
     started = time.monotonic()
-    payload = build_update_plan(
+    payload = build_update_plan(build_from_source=True,
         action=action,
         repo="example/loopx",
         ref="main",
@@ -641,7 +641,7 @@ def test_pipx_distribution_apply_preserves_the_pipx_environment() -> None:
 def test_archive_apply_restarts_managed_services_when_only_extensions_are_blocked(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    payload = build_update_plan(action="apply", doctor_payload=doctor_payload())
+    payload = build_update_plan(build_from_source=True, action="apply", doctor_payload=doctor_payload())
 
     assert payload["install_lifecycle"]["execution_driver"] == "archive_snapshot"
 
@@ -681,7 +681,7 @@ def test_archive_apply_restarts_managed_services_when_only_extensions_are_blocke
 def test_archive_apply_skips_managed_service_restart_when_the_runtime_is_unhealthy(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    payload = build_update_plan(action="apply", doctor_payload=doctor_payload())
+    payload = build_update_plan(build_from_source=True, action="apply", doctor_payload=doctor_payload())
 
     failed_install = subprocess.CompletedProcess([], 1, "", "installer failed")
     monkeypatch.setenv("LOOPX_PYTHON", sys.executable)
@@ -713,7 +713,7 @@ def test_live_checkout_apply_never_mutates_git_or_switches_install_channels() ->
         "/workspace/loopx/scripts/install-local.sh\nloopx doctor"
     )
 
-    payload = build_update_plan(action="apply", doctor_payload=doctor)
+    payload = build_update_plan(build_from_source=True, action="apply", doctor_payload=doctor)
 
     assert payload["ok"] is False
     assert payload["install_lifecycle"]["owner"] == "source_checkout"
@@ -938,7 +938,7 @@ def test_failed_authority_upgrade_does_not_activate_services_or_continue_host_up
 ])
 def test_update_preserves_selected_route_through_install_and_readback(tmp_path, monkeypatch, driver):
     registry, runtime = tmp_path / "project/.loopx/registry.json", tmp_path / "selected runtime"
-    payload = build_update_plan(action="apply", doctor_payload=doctor_payload())
+    payload = build_update_plan(build_from_source=True, action="apply", doctor_payload=doctor_payload())
     payload["install_lifecycle"].update(execution_driver=driver)
     calls, installer_env = [], {}
     monkeypatch.setenv("LOOPX_PYTHON", sys.executable)
@@ -1018,7 +1018,7 @@ def test_archive_readback_and_activation_follow_installer_bin_owner(
 @pytest.mark.parametrize("driver", ["python_pip", "python_pipx"])
 def test_python_distribution_activation_ignores_archive_bin_override(tmp_path, monkeypatch, driver):
     monkeypatch.setenv("LOOPX_BIN_DIR", str(tmp_path / "custom/bin"))
-    payload = build_update_plan(action="apply", doctor_payload=doctor_payload())
+    payload = build_update_plan(build_from_source=True, action="apply", doctor_payload=doctor_payload())
     payload["install_lifecycle"].update(execution_driver=driver)
     passed = subprocess.CompletedProcess([], 0, '{"ok":true}', '')
     with (
diff --git a/tests/test_self_upgrade.py b/tests/test_self_upgrade.py
index 8e7ea35d0b..6711a792d5 100644
--- a/tests/test_self_upgrade.py
+++ b/tests/test_self_upgrade.py
@@ -139,6 +139,7 @@ def test_multiple_apps_block_the_whole_plan_before_cli_mutation(monkeypatch, tmp
 def test_bare_upgrade_applies_while_bare_update_stays_read_only(
     monkeypatch, tmp_path, command, expected
 ):
+    monkeypatch.setenv("LOOPX_UPDATE_WORKER", "1")
     parser = argparse.ArgumentParser()
     register_update_command(parser.add_subparsers(dest="command"), lambda _parser: None)
     args = parser.parse_args([command])
@@ -309,6 +310,7 @@ def test_cli_only_missing_identity_never_certifies_success():
 
 
 def test_cli_only_discovery_failure_exit_render_and_recovery(monkeypatch, tmp_path):
+    monkeypatch.setenv("LOOPX_UPDATE_WORKER", "1")
     home = tmp_path / "codex"
     database = home / "sqlite/codex-dev.db"
     database.parent.mkdir(parents=True)
@@ -374,7 +376,7 @@ def test_failed_cli_does_not_restart_the_app(monkeypatch):
     assert result["desktop_upgrade"]["status"] == "skipped_runtime_update_failed"
 
 
-def test_progress_stays_on_stderr_and_json_mode_is_quiet(capsys):
+def test_human_progress_stays_on_stderr_and_explicit_disable_is_quiet(capsys):
     progress = UpdateProgress(True)
     progress.report("Installing CLI and building workspace resources")
     progress.stop()
diff --git a/tests/test_update_artifact.py b/tests/test_update_artifact.py
new file mode 100644
index 0000000000..a1dd0fd3bf
--- /dev/null
+++ b/tests/test_update_artifact.py
@@ -0,0 +1,216 @@
+"""Qualification over real HTTP/files, including interrupted and hostile transfers."""
+
+from contextlib import contextmanager
+import hashlib
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import io
+import json
+import os
+import tarfile
+import threading
+
+import pytest
+
+from loopx.cli_commands import update_artifact as update
+
+
+@contextmanager
+def server(body: bytes):
+    state = {"requests": [], "interrupt": False, "offline": False, "ignore_range": False,
+             "etag": '"one"', "no_length": False, "wrong_range": False}
+    class Handler(BaseHTTPRequestHandler):
+        def do_GET(self):
+            if state.get("redirect"):
+                self.send_response(302)
+                self.send_header("Location", state["redirect"])
+                self.end_headers()
+                return
+            offset = int(self.headers.get("Range", "bytes=0-")[6:-1])
+            state["requests"].append((self.path, self.headers.get("Range"), self.headers.get("If-Range")))
+            if state["offline"] or self.path == "/bad":
+                self.send_response(503)
+                self.send_header("Content-Length", "0")
+                self.end_headers()
+                return
+            if state["ignore_range"]:
+                offset = 0
+            self.send_response(206 if offset else 200)
+            self.send_header("ETag", state["etag"])
+            if not state["no_length"]:
+                self.send_header("Content-Length", str(len(body) - offset))
+            self.send_header("Connection", "close")
+            if offset:
+                self.send_header("Content-Range", f"bytes {offset + int(state['wrong_range'])}-{len(body) - 1}/{len(body)}")
+            self.end_headers()
+            if state["interrupt"]:
+                state["offline"] = True
+                self.wfile.write(body[:len(body) // 2])
+                self.wfile.flush()
+                self.close_connection = True
+            else:
+                self.wfile.write(body[offset:])
+        def log_message(self, *_args):
+            pass
+    http = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=http.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield f"http://127.0.0.1:{http.server_port}/runtime.tar.gz", state
+    finally:
+        http.shutdown()
+        http.server_close()
+        thread.join()
+
+
+def manifest(body: bytes, url: str):
+    return {"schema_version": update.RUNTIME_ARTIFACT_SCHEMA_VERSION, "source_revision": "a" * 40,
+        "package_version": "1.2.3", "channel": "main", "size": len(body), "sha256": hashlib.sha256(body).hexdigest(), "urls": [url]}
+
+
+def test_redirects_cannot_weaken_source_tls():
+    from urllib.request import Request
+    handler = update.UpdateRedirectHandler()
+    with pytest.raises(ValueError, match="preserve TLS"):
+        handler.redirect_request(Request("https://publisher.invalid/runtime"), None, 302, "", {}, "http://127.0.0.1/runtime")
+    with server(b"{}") as (url, state):
+        state["redirect"] = "http://publisher.invalid/runtime"
+        with pytest.raises(ValueError, match="metadata unavailable") as caught:
+            update.read_json(url)
+        assert "require HTTPS" in str(caught.value.__cause__)
+
+
+def test_pinned_runtime_metadata_is_revision_addressed_within_the_channel_release(monkeypatch):
+    body = b"runtime"
+    descriptor = manifest(body, "https://publisher.invalid/runtime.tar.gz")
+    requests = []
+    def metadata(url):
+        requests.append(url)
+        return descriptor
+    monkeypatch.setattr(update, "read_json", metadata)
+    assert update.resolve_artifact("example/loopx", "a" * 40)["source_revision"] == "a" * 40
+    assert requests == ["https://github.com/example/loopx/releases/download/runtime-main/loopx-runtime-" + "a" * 40 + ".json"]
+
+
+@pytest.mark.parametrize("ignore_range,new_etag,wrong_range", [(False, False, False), (True, False, False), (False, True, False), (False, False, True)])
+def test_cross_invocation_resume_validates_origin_range_and_etag(tmp_path, ignore_range, new_etag, wrong_range):
+    body = os.urandom(192 * 1024)
+    with server(body) as (url, state):
+        descriptor = manifest(body, url)
+        state["interrupt"] = True
+        with pytest.raises(ValueError, match="download failed"):
+            update.download_artifact(descriptor, cache=tmp_path)
+        partial = next(tmp_path.glob("*.partial"))
+        assert partial.stat().st_size == len(body) // 2
+        state.update(offline=False, interrupt=False, ignore_range=ignore_range, wrong_range=wrong_range)
+        if new_etag:
+            state["etag"] = '"two"'
+        boundary = len(state["requests"])
+        result = update.download_artifact(descriptor, cache=tmp_path)
+        assert result.read_bytes() == body
+        assert state["requests"][boundary][1:] == (f"bytes={len(body) // 2}-", '"one"')
+        if new_etag or wrong_range:
+            assert state["requests"][boundary + 1][1] is None
+        requests = len(state["requests"])
+        state["offline"] = True
+        assert update.download_artifact(descriptor, cache=tmp_path) == result
+        assert len(state["requests"]) == requests  # Qualified cache is network-free.
+        result.write_bytes(b"corrupt")
+        with pytest.raises(ValueError):
+            update.download_artifact(descriptor, cache=tmp_path)
+        assert not result.exists()  # A damaged cache is never returned.
+
+
+def test_unknown_http_length_has_real_byte_progress_and_expected_manifest_size(tmp_path):
+    body = os.urandom(128 * 1024)
+    with server(body) as (url, state):
+        state["no_length"] = True
+        events = []
+        result = update.download_artifact(manifest(body, url), cache=tmp_path, progress=events.append)
+        assert result.read_bytes() == body
+        assert events[-1]["received"] == events[-1]["total"] == len(body)
+
+
+def test_mirror_serves_only_the_identical_hash_and_bad_bytes_never_install(tmp_path):
+    body = os.urandom(8192)
+    with server(body) as (url, state):
+        descriptor = manifest(body, url)
+        descriptor["urls"].insert(0, url.replace("/runtime.tar.gz", "/bad"))
+        result = update.download_artifact(descriptor, cache=tmp_path)
+        assert result.read_bytes() == body
+        assert [entry[0] for entry in state["requests"]] == ["/bad", "/runtime.tar.gz"]
+        descriptor["sha256"] = "0" * 64
+        with pytest.raises(ValueError, match="integrity_mismatch"):
+            update.download_artifact(descriptor, cache=tmp_path)
+        assert len(list(tmp_path.glob("*.tar.gz"))) == 1
+
+
+@pytest.mark.parametrize("url", ["http://example.com/update", "https://user:secret@example.com/update", "file:///tmp/archive"])
+def test_update_source_policy_has_no_tls_or_credential_fallback(url):
+    with pytest.raises(ValueError):
+        update.validate_manifest(manifest(b"bytes", url))
+
+
+@pytest.mark.parametrize("name,link", [("../escape", False), ("/escape", False), ("C:/escape", False), ("safe/link", True)])
+def test_unsafe_archive_is_rejected_before_extraction(tmp_path, name, link):
+    archive = tmp_path / "bad.tar.gz"
+    with tarfile.open(archive, "w:gz") as output:
+        member = tarfile.TarInfo(name)
+        member.size = 1
+        if link:
+            member.type, member.linkname = tarfile.SYMTYPE, "../../escape"
+        output.addfile(member, None if link else io.BytesIO(b"x"))
+    destination = tmp_path / "extracted"
+    destination.mkdir()
+    with pytest.raises(ValueError, match="unsafe entry"):
+        update.extract_artifact(archive, destination)
+    assert not list(destination.iterdir())
+
+
+def test_resolution_pins_exact_artifact_and_never_falls_back_to_codeload(monkeypatch):
+    descriptor = manifest(b"bytes", "https://example.com/runtime.tar.gz")
+    calls = []
+    monkeypatch.setattr(update, "read_json", lambda url: calls.append(url) or descriptor)
+    assert update.resolve_artifact("loopx-project/loopx", "main")["source_revision"] == "a" * 40
+    assert "/runtime-main/" in calls[0]
+    with pytest.raises(ValueError, match="another revision"):
+        update.resolve_artifact("loopx-project/loopx", "b" * 40)
+    with pytest.raises(ValueError, match="explicitly use --source"):
+        update.resolve_artifact("loopx-project/loopx", "developer-branch")
+
+
+def test_real_installer_streams_only_explicit_events_and_bounds_diagnostics(tmp_path):
+    import sys
+    events = []
+    script = tmp_path / "installer.py"
+    script.write_text("import sys\nprint('private output'*3000)\nprint('LOOPX_UPDATE_EVENT {\"stage\":\"activate\"}',file=sys.stderr)\n")
+    result = update.run_installer([sys.executable, str(script)], env=dict(os.environ), timeout=5, progress=events.append)
+    assert result.returncode == 0
+    assert events == [{"stage": "activate"}]
+    assert len(result.stdout) <= 16384
+
+
+def test_bootstrap_inherits_snapshot_choice_without_taking_over_a_package_owner(tmp_path, monkeypatch):
+    import subprocess
+    import sys
+    binary = tmp_path / "bin"
+    scripts = tmp_path / "release/scripts"
+    scripts.mkdir(parents=True)
+    binary.mkdir()
+    (scripts / "loopx").touch()
+    entry = binary / "loopx"
+    entry.symlink_to(scripts / "loopx")
+    (scripts.parent / "release.json").write_text(json.dumps({"source": {
+        "repo": "example/fork", "ref": "main", "git_commit": "a" * 40}}))
+    monkeypatch.setenv("LOOPX_BIN_DIR", str(binary))
+    monkeypatch.setenv("LOOPX_UPDATE_REF_EXPLICIT", "0")
+    monkeypatch.setenv("LOOPX_UPDATE_REPO_EXPLICIT", "0")
+    monkeypatch.setattr(sys, "argv", ["update-artifact.py"])
+    selected = []
+    monkeypatch.setattr(update, "resolve_artifact", lambda repo, ref: selected.append((repo, ref)) or manifest(b"x", "https://example.com/runtime"))
+    monkeypatch.setattr(update, "install_artifact", lambda *_a, **_kw: (subprocess.CompletedProcess([], 0, "", ""), {}))
+    assert update.main() == 0
+    assert selected == [("example/fork", "main")]
+    entry.unlink()
+    entry.touch()  # pip/another installer owns a regular console entry.
+    assert update.main() == 1
+    assert len(selected) == 1
diff --git a/tests/test_update_operation.py b/tests/test_update_operation.py
new file mode 100644
index 0000000000..a9d66a62af
--- /dev/null
+++ b/tests/test_update_operation.py
@@ -0,0 +1,168 @@
+"""Update intent, progress and detached observation without live user state."""
+
+import argparse
+import json
+import os
+import subprocess
+import sys
+import time
+import zipfile
+import zipimport
+
+import pytest
+
+from loopx.cli_commands.support_control_update_progress import UpdateProgress
+from loopx.cli_commands import update_operation as operations
+from loopx.self_update import build_update_plan, _source_config, _installer_env_for_source
+from test_self_update_runtime_activation import doctor_payload
+
+
+def test_installed_channel_and_pinned_ref_survive_both_update_entrypoints(monkeypatch):
+    for name in ("LOOPX_REF", "LOOPX_REPO", "LOOPX_ARCHIVE_URL"):
+        monkeypatch.delenv(name, raising=False)
+    for ref in ("main", "stable", "f" * 40):
+        plan = build_update_plan(doctor_payload=doctor_payload(source_ref=ref))
+        assert plan["source"]["ref"] == ref
+        assert plan["source"]["ref_source"] == "installed_manifest"
+        assert plan["source"]["distribution"] == "runtime"
+    assert _source_config(repo=None, ref=None, archive_url=None)["ref"] == "stable"
+
+
+def test_runtime_check_uses_one_qualified_version_and_revision(monkeypatch):
+    from loopx.cli_commands import update_source
+    def no_branch_probe(*_args):
+        raise AssertionError("The mutable branch version must not describe a qualified artifact")
+    monkeypatch.setattr("loopx.self_update._source_version_check", no_branch_probe)
+    monkeypatch.setattr(update_source, "resolve_artifact", lambda *_: {
+        "source_revision": "2" * 40, "package_version": "1.2.3"})
+    monkeypatch.setattr(update_source, "target_direction", lambda *_: "ahead")
+    plan = build_update_plan(action="check", doctor_payload=doctor_payload(), resolve_remote=True)
+    assert plan["source"]["resolved_revision"] == "2" * 40
+    assert plan["source_version_check"]["version"] == "1.2.3"
+
+
+@pytest.mark.parametrize("direction,accepted", [("ahead", True), ("identical", True), ("behind", False), ("diverged", False), ("unknown", False)])
+def test_target_is_resolved_before_any_mutation_and_downgrade_needs_intent(monkeypatch, direction, accepted):
+    from loopx.cli_commands import update_source
+    artifact = {"source_revision": "2" * 40, "package_version": "1.0.0"}
+    monkeypatch.setattr(update_source, "resolve_artifact", lambda *_: artifact)
+    monkeypatch.setattr(update_source, "target_direction", lambda *_: direction)
+    plan = build_update_plan(action="apply", doctor_payload=doctor_payload(), resolve_remote=True)
+    assert plan["source"]["resolved_revision"] == artifact["source_revision"]
+    assert plan["plan"]["apply_supported"] is accepted
+    override = build_update_plan(action="apply", doctor_payload=doctor_payload(), resolve_remote=True, allow_downgrade=True)
+    assert override["plan"]["apply_supported"] is True
+
+
+def test_installer_uses_frozen_revision_but_persists_channel_and_proxy(monkeypatch):
+    monkeypatch.setenv("HTTPS_PROXY", "http://operator-proxy.invalid:8080")
+    source = _source_config(repo=None, ref="main", archive_url=None, build_from_source=True)
+    source["resolved_revision"] = "b" * 40
+    env = _installer_env_for_source(source)
+    assert env["LOOPX_REF"] == "b" * 40
+    assert env["LOOPX_UPDATE_CHANNEL"] == "main"
+    assert env["HTTPS_PROXY"] == "http://operator-proxy.invalid:8080"
+    assert env["LOOPX_BUILD_FROM_SOURCE"] == "1"
+
+
+def test_json_stdout_is_final_only_and_heartbeat_does_not_invent_progress(tmp_path, capsys):
+    progress = UpdateProgress(json_output=True, status_path=tmp_path / "status.json")
+    try:
+        progress.report({"stage": "download", "received": 1024, "total": None})
+        progress.changed -= 11
+        progress._write()
+        captured = capsys.readouterr()
+        assert captured.out == ""
+        events = [json.loads(line) for line in captured.err.splitlines()]
+        assert events[-1]["total"] is None and events[-1]["stalled"] is True
+        assert events[0]["operation_id"] == events[-1]["operation_id"]
+        assert events[-1]["last_progress_at"] < events[-1]["updated_at"] - 10
+        assert json.loads((tmp_path / "status.json").read_text())["received"] == 1024
+    finally:
+        progress.stop()
+
+
+def test_active_worker_is_reattached_and_status_never_starts_an_install(tmp_path, monkeypatch):
+    monkeypatch.setenv("LOOPX_UPDATE_STATE_DIR", str(tmp_path / "state"))
+    monkeypatch.setenv("LOOPX_BIN_DIR", str(tmp_path / "bin"))
+    root = operations.operation_root()
+    operation_id = "c" * 32
+    folder = root / operation_id
+    folder.mkdir()
+    operations.atomic_json(root / f"{operations.installation_key()}.active.json", {"operation_id": operation_id})
+    operations.atomic_json(folder / "status.json", {"operation_id": operation_id,
+        "status": "running", "pid": os.getpid(), "updated_at": time.time()})
+    def no_spawn(*_args, **_kwargs):
+        raise AssertionError("A second installation must not start")
+    monkeypatch.setattr(subprocess, "Popen", no_spawn)
+    result = operations.start_or_observe(argparse.Namespace(), tmp_path / "registry.json")
+    assert result["operation_id"] == operation_id and result["reattached"] is True
+    assert operations.read_operation()["status"] == "running"
+    operations.atomic_json(folder / "result.json", {"ok": False, "error": "activation failed"})
+    assert operations.read_operation()["status"] == "failed"
+
+
+def test_real_detached_worker_finishes_after_its_observer_exits(tmp_path):
+    env = dict(os.environ, LOOPX_UPDATE_STATE_DIR=str(tmp_path / "state"),
+        LOOPX_BIN_DIR=str(tmp_path / "bin"), LOOPX_USAGE_PING="0")
+    # An unsupported source checkout is a real negative update journey: the
+    # owning CLI resolves it and records failure, without running an installer.
+    command = [sys.executable, "-m", "loopx.cli", "--registry", str(tmp_path / "registry.json"),
+        "--format", "json", "update", "apply", "--background", "--source", "--ref", "main"]
+    started = subprocess.run(command, env=env, text=True, capture_output=True, timeout=30, check=True)
+    operation_id = json.loads(started.stdout)["operation_id"]
+    assert len(operation_id) == 32
+    for _ in range(80):
+        result = subprocess.run([sys.executable, "-m", "loopx.cli", "--format", "json",
+            "update", "status", "--operation-id", operation_id], env=env, text=True, capture_output=True, timeout=15)
+        payload = json.loads(result.stdout)
+        if payload["status"] not in {"starting", "running"}:
+            break
+        time.sleep(.1)
+    assert payload["status"] == "failed", payload
+    assert payload["result"]["changes_applied"] is False
+    assert payload["result"]["plan"]["apply_supported"] is False
+
+
+@pytest.mark.parametrize("switched", [True, False])
+def test_material_failure_reports_an_observed_switch_but_never_invents_one(monkeypatch, switched):
+    from loopx.self_update import execute_update_plan
+    plan = build_update_plan(action="apply", doctor_payload=doctor_payload(), build_from_source=True)
+    plan["source"]["resolved_revision"] = "1" * 40
+    readback = {"install_freshness": {"manifest_source_git_commit": "1" * 40,
+        "release_id": "new-release" if switched else "fixture"}}
+    monkeypatch.setenv("LOOPX_PYTHON", sys.executable)
+    monkeypatch.setattr("loopx.self_update.run_archive_installer", lambda *_a, **_kw: (
+        subprocess.CompletedProcess([], 1, "", "host material failed"), {}))
+    monkeypatch.setattr("loopx.self_update.subprocess.run", lambda *_a, **_kw:
+        subprocess.CompletedProcess([], 0, json.dumps(readback), ""))
+    monkeypatch.setattr("loopx.runtime_activation.restart_managed_loopx_services", lambda: [])
+    result = execute_update_plan(plan)
+    assert result["ok"] is False
+    assert result["changes_applied"] is switched
+    assert result["execution"]["activation_observed"] is switched
+
+
+def test_windows_console_guard_runs_in_original_process_before_detaching(tmp_path, monkeypatch):
+    from loopx.cli_commands import support_control_update as command
+    archive = tmp_path / "loopx.exe"
+    with zipfile.ZipFile(archive, "w") as output:
+        output.writestr("__main__.py", "")
+    monkeypatch.setattr(sys, "platform", "win32")
+    monkeypatch.setattr(sys.modules["__main__"], "__loader__", zipimport.zipimporter(str(archive)))
+    monkeypatch.delenv("LOOPX_UPDATE_WORKER", raising=False)
+    monkeypatch.setattr(command, "build_update_plan", lambda **_kw: {
+        "ok": False, "plan": {"apply_supported": False}, "error": "original_console_refusal"})
+    def no_detach(*_args, **_kwargs):
+        raise AssertionError("Detaching would hide the locked console from its existing guard")
+    monkeypatch.setattr(operations, "start_or_observe", no_detach)
+    parser = argparse.ArgumentParser()
+    command.register_update_command(parser.add_subparsers(dest="command"), lambda _parser: None)
+    args = parser.parse_args(["update", "apply", "--background"])
+    args.runtime_root = None
+    values = []
+    code = command.handle_update_command(args, registry_path=tmp_path / "registry.json",
+        registry_was_supplied=True, print_payload=lambda value, *_: values.append(value),
+        output_format=lambda _: "json")
+    assert code == 1
+    assert values[0]["error"] == "original_console_refusal"