diff --git a/docs/guides/personal-workspace-user-guide.md b/docs/guides/personal-workspace-user-guide.md
index e30f731715..d82d2ec9d5 100644
--- a/docs/guides/personal-workspace-user-guide.md
+++ b/docs/guides/personal-workspace-user-guide.md
@@ -280,6 +280,25 @@ loopx goal-lifecycle --goal-id <goal-id> --operation resume --actor-kind owner -
 
 执行时，LoopX 会写入权威 source registry、同步全局 registry，并验证两端 readback；任一端未验证成功时不会宣称操作完成。
 
+暂停/恢复只由持久化的 `activation.state` 控制。Registry 的 `status` 继续表达注册、
+归档等状态，不再接受旧的 `paused` / `stopped` 值；前端响应中的 `activation_state`
+是派生展示值，不能写回 registry。
+
+使用 `loopx upgrade` 或 snapshot 安装器时，会在启用候选运行时前，通过已有的
+`authority-archive upgrade --all-known --execute` 流程备份并迁移这些旧元数据。
+直接通过 pip 安装、源码运行或恢复旧备份时，可先只读检查：
+
+```bash
+loopx authority-archive upgrade --all-known --require-current
+# 需要迁移时，先备份原字节，再执行迁移
+loopx authority-archive upgrade --all-known --execute
+```
+
+原 registry 保存在相邻 `backups/registry-preimages/` 下，迁移结果返回具体备份路径。
+迁移保留暂停意图；字段冲突时保持停止，不会自动 Resume，也不改 Todo、执行回执或额度。
+之后仍可从「已停止」区或上面的同一 CLI 恢复。迁移失败时升级不会宣称完成，备份保留，
+原命令可重试；不要用备份覆盖迁移之后产生的新工作。
+
 切换到 SSH 状态来源后，只有来源与本机 OpenSSH 配置中的精确 Host alias 绑定时，
 侧边栏才显示停止/恢复按钮。操作通过 SSH 在目标主机执行同一个
 `goal-lifecycle` typed contract，并验证远端投影；不会回退修改本机同名 Goal。
diff --git a/loopx/capabilities/periodic_report/machine_defaults.py b/loopx/capabilities/periodic_report/machine_defaults.py
index 9cf721319b..bc67b44d8c 100644
--- a/loopx/capabilities/periodic_report/machine_defaults.py
+++ b/loopx/capabilities/periodic_report/machine_defaults.py
@@ -399,7 +399,7 @@ def resolve_active_goal_periodic_report_subscription(
     if (
         not isinstance(goal, Mapping)
         or goal_is_stopped(goal)
-        or goal.get("status") in {"stopped", "paused", "archived"}
+        or goal.get("status") == "archived"
     ):
         return None
     subscription = resolve_goal_periodic_report_subscription(goal, machine_defaults)
diff --git a/loopx/cli_commands/authority_archive.py b/loopx/cli_commands/authority_archive.py
index 070d5500cf..8c43c3009d 100644
--- a/loopx/cli_commands/authority_archive.py
+++ b/loopx/cli_commands/authority_archive.py
@@ -9,7 +9,11 @@
 
 from ..control_plane.effect_runtime import EffectRuntimeResponseAmbiguous, effect_runtime_result
 from ..paths import global_registry_path, resolve_runtime_root, select_default_runtime_root
-from ..control_plane.projects.registry_codec import load_registry
+from ..control_plane.projects.registry_codec import (
+    load_registry,
+    read_activation_upgrade_registry,
+)
+from ..control_plane.goals.activation_service import upgrade_goal_activation_registries
 
 
 def register_authority_archive_command(
@@ -76,6 +80,9 @@ def handle_authority_archive_command(
         if args.authority_archive_action == "inspect":
             request["source"] = str(args.source.expanduser().resolve())
         elif args.authority_archive_action == "upgrade":
+            registries = authority_upgrade_registries(
+                registry_path, runtime_root_arg, all_known=args.all_known)
+            activation_plan = upgrade_goal_activation_registries(registries, execute=False)
             request.update(runtime_roots=authority_upgrade_roots(
                 registry_path, runtime_root_arg, all_known=args.all_known), execute=args.execute)
         elif args.authority_archive_action in {"plan-migration", "migrate"}:
@@ -107,6 +114,16 @@ def handle_authority_archive_command(
         result = effect_runtime_result(
             "coordination.authority_archive.manage", request, timeout=300.0, retry_safe=False
         )
+        if args.authority_archive_action == "upgrade" and result.get("status") != "failed":
+            result["goal_activation_upgrade"] = (
+                upgrade_goal_activation_registries(registries, execute=True)
+                if args.execute else activation_plan
+            )
+            if args.require_current and any(
+                row["status"] == "planned"
+                for row in activation_plan["results"]
+            ):
+                result.update(status="failed", reason="Goal activation metadata upgrade required before activating this runtime.")
     except (OSError, RuntimeError, ValueError) as error:
         uncertain = args.authority_archive_action == "migrate" and args.execute
         result = {"status": "failed", "reason": str(error),
@@ -137,7 +154,7 @@ def handle_authority_archive_command(
 def authority_upgrade_roots(registry_path: Path, runtime_root_arg: str | None,
                             *, all_known: bool) -> list[str]:
     """Bounded discovery from existing registries; never scan arbitrary home paths."""
-    registry = load_registry(registry_path) if registry_path.exists() else {}
+    registry = read_activation_upgrade_registry(registry_path)
     override = runtime_root_arg or os.environ.get("LOOPX_RUNTIME_ROOT")
     selected = resolve_runtime_root(registry, override, registry_path=registry_path).resolve()
     roots = {selected}
@@ -145,7 +162,7 @@ def authority_upgrade_roots(registry_path: Path, runtime_root_arg: str | None,
         common = Path(override).expanduser().resolve() if override else select_default_runtime_root().resolve()
         roots.add(common)
         global_path = global_registry_path(common)
-        global_registry = load_registry(global_path) if global_path.exists() else {}
+        global_registry = read_activation_upgrade_registry(global_path)
         for goal in global_registry.get("goals", []):
             source = goal.get("source_registry") if isinstance(goal, dict) else None
             if not source:
@@ -155,5 +172,23 @@ def authority_upgrade_roots(registry_path: Path, runtime_root_arg: str | None,
                 raise ValueError("Registered source_registry must be absolute for automatic upgrade")
             # A disconnected checkout does not hide the common runtime store.
             if path.exists():
-                roots.add(resolve_runtime_root(load_registry(path), registry_path=path).resolve())
+                roots.add(resolve_runtime_root(read_activation_upgrade_registry(path), registry_path=path).resolve())
     return sorted(str(root) for root in roots)
+
+
+def authority_upgrade_registries(registry_path: Path, runtime_root_arg: str | None,
+                                *, all_known: bool) -> list[Path]:
+    """Use the same registered routes as store discovery, including the selection."""
+    roots = authority_upgrade_roots(registry_path, runtime_root_arg, all_known=all_known)
+    paths = {registry_path.expanduser().resolve()}
+    for root in roots:
+        path = global_registry_path(Path(root))
+        paths.add(path)
+        for goal in read_activation_upgrade_registry(path).get("goals", []):
+            source = goal.get("source_registry") if isinstance(goal, dict) else None
+            if source:
+                source_path = Path(str(source)).expanduser()
+                if not source_path.is_absolute():
+                    raise ValueError("Registered source_registry must be absolute for automatic upgrade")
+                paths.add(source_path.resolve())
+    return sorted(path for path in paths if path.exists())
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index 98a611c07f..a2fb062a58 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -345,6 +345,7 @@ export function createEffectRuntimeHandlers(
     ["goal.vision_wait.project", lazyHandler(() => import("./goals/vision_wait_coverage.ts"), ({projectGoalVisionWaitState}) => projectGoalVisionWaitState)],
     ["goal.shared_goal_alignment.project", lazyHandler(() => import("./goals/shared_goal_alignment.ts"), ({projectSharedGoalAlignment}) => projectSharedGoalAlignment)],
     ["goal.operator_actions.project", lazyHandler(() => import("./goals/operator_actions.ts"), ({projectGoalOperatorActions}) => projectGoalOperatorActions)],
+    ["goal.activation.migrate", lazyHandler(() => import("./goals/activation_migration.ts"), ({migrateGoalActivationMetadata}) => migrateGoalActivationMetadata)],
     ["goal.amendment.execute", lazyHandler(() => import("./goals/goal_amendment_authority.ts"), ({goalAmendmentLocal}) => goalAmendmentLocal)],
     ["goal.direction.execute", lazyHandler(() => import("./goals/direction_authority.ts"), ({directionLocal}) => directionLocal)],
     ["goal.amendment_proposal.admit", lazyHandler(() => import("./goals/goal_amendment_proposal.ts"), ({admitGoalAmendmentProposal}) => admitGoalAmendmentProposal)],
diff --git a/loopx/control_plane/goals/acceptance.py b/loopx/control_plane/goals/acceptance.py
index 8f639ad4d0..27efca2718 100644
--- a/loopx/control_plane/goals/acceptance.py
+++ b/loopx/control_plane/goals/acceptance.py
@@ -161,7 +161,7 @@ def refine_goal_outcome(
     registry_path, goal, route = _routing(registry_path, goal_id, runtime_root, agent_id)
     if "owner_agent_id" in refinement and refinement["owner_agent_id"] not in registered_agent_ids_for_goal(goal):
         raise ValueError("Outcome owner must be a registered Agent")
-    # The registry's legacy omitted-status form denotes an active Goal too.
+    # Registration eligibility is independent of the typed activation gate.
     if goal.get("status", "active") not in {"active", "ready"} or goal_is_stopped(goal):
         raise ValueError("Outcome refinement requires an active Goal")
     return _result("goal.acceptance.refine", {**route, "actor_agent_id": agent_id,
diff --git a/loopx/control_plane/goals/activation.py b/loopx/control_plane/goals/activation.py
index 4d11940220..26cf1947dd 100644
--- a/loopx/control_plane/goals/activation.py
+++ b/loopx/control_plane/goals/activation.py
@@ -24,14 +24,29 @@ def normalize_goal_activation_state(value: Any) -> GoalActivationState:
         raise ValueError("goal activation state must be active or stopped") from exc
 
 
+def require_current_registry_goal_activation(goal: Mapping[str, Any]) -> None:
+    """Reject retired storage fields, without interpreting them as live authority."""
+    if "activation_state" in goal or goal.get("status") in ("paused", "stopped"):
+        raise ValueError(
+            "Goal activation metadata requires upgrade; run "
+            "loopx authority-archive upgrade --all-known --execute "
+            "to back up and migrate the registry before activation"
+        )
+
+
 def goal_activation_state(goal: Mapping[str, Any] | None) -> GoalActivationState:
-    """Return the typed Goal activation state; older registries default active."""
+    """Read canonical activation or its derived read-model projection.
+
+    Registry IO rejects stored activation_state aliases and retired pause values;
+    only the explicit upgrade decoder may consume them. Omitted activation is active.
+    """
 
     if not isinstance(goal, Mapping):
         return GoalActivationState.ACTIVE
     projected = goal.get("activation_state")
     if projected is not None:
         return normalize_goal_activation_state(projected)
+    require_current_registry_goal_activation(goal)
     activation = goal.get("activation")
     if activation is None:
         return GoalActivationState.ACTIVE
diff --git a/loopx/control_plane/goals/activation_migration.ts b/loopx/control_plane/goals/activation_migration.ts
new file mode 100644
index 0000000000..ac424ac5cf
--- /dev/null
+++ b/loopx/control_plane/goals/activation_migration.ts
@@ -0,0 +1,75 @@
+/** Upgrade-only decoder for retired registry pause values and stored readback aliases.
+ * Normal readers and writers require current metadata; this never grants a Resume. */
+import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
+import type {JsonObject} from "../effect_program.ts";
+import {requireJsonObject, requireNonEmptyString} from "../runtime_decode.ts";
+import {SOURCE_SESSION_PROFILE_ID} from "./source_session_lifetime.ts";
+
+const SCHEMA = "loopx_goal_activation_v1";
+
+function state(value: unknown): "active" | "stopped" {
+  if (value === "active" || value === "stopped") return value;
+  throw new EffectRuntimeRequestError("Goal activation state must be active or stopped");
+}
+
+export function migrateGoalActivationMetadata(params: JsonObject): JsonObject {
+  if (!Array.isArray(params.registries)) throw new EffectRuntimeRequestError("registries must be an array");
+  const registries = params.registries.map(raw => {
+    const row = requireJsonObject(raw, "registry snapshot");
+    const path = requireNonEmptyString(row.path, "registry path");
+    const payload = requireJsonObject(row.payload, "registry payload");
+    if (payload.goals !== undefined && !Array.isArray(payload.goals)) throw new EffectRuntimeRequestError("registry goals must be an array");
+    return {path, payload, goals: (payload.goals ?? []) as unknown[]};
+  });
+  if (new Set(registries.map(row => row.path)).size !== registries.length) throw new EffectRuntimeRequestError("duplicate registry path");
+  const nodes = registries.flatMap(registry => registry.goals.map((raw, index) => {
+    const goal = requireJsonObject(raw, "registry Goal");
+    const activation = goal.activation == null ? null : requireJsonObject(goal.activation, "Goal activation");
+    if (activation?.schema_version && activation.schema_version !== SCHEMA) {
+      throw new EffectRuntimeRequestError("unsupported Goal activation schema");
+    }
+    const current = activation === null ? "active" : state(activation.state);
+    const hasAlias = Object.hasOwn(goal, "activation_state");
+    const alias = goal.activation_state == null ? null : state(goal.activation_state);
+    const legacyPause = goal.status === "paused" || goal.status === "stopped";
+    return {registry, index, goal, activation, current, legacyPause,
+      legacy: legacyPause || hasAlias,
+      stopped: legacyPause || alias === "stopped" || current === "stopped"};
+  }));
+  // Only an exact source route and the same Goal lifetime join a projection to
+  // its source. A stale projection cannot stop a recreated Goal of the same name.
+  const components = new Map<typeof nodes[number], Set<typeof nodes[number]>>();
+  for (const node of nodes) components.set(node, new Set([node]));
+  for (const node of nodes) {
+    if (!node.goal.source_registry) continue;
+    const matches = nodes.filter(candidate => candidate.registry.path === node.goal.source_registry
+      && candidate.goal.id === node.goal.id
+      && (candidate.goal.goal_instance_id ?? null) === (node.goal.goal_instance_id ?? null));
+    if (matches.length > 1) throw new EffectRuntimeRequestError("ambiguous source Goal during activation upgrade");
+    const source = matches[0];
+    if (!source) continue;
+    const merged = new Set([...components.get(node)!, ...components.get(source)!]);
+    for (const member of merged) components.set(member, merged);
+  }
+  const replacements = new Map<typeof nodes[number], JsonObject>();
+  for (const node of nodes) {
+    const component = [...components.get(node)!];
+    if (!component.some(member => member.legacy)) continue;
+    // Conflicting historical representations never silently resume execution.
+    // Resume remains an explicit owner operation after the backup and upgrade.
+    const nextState = component.some(member => member.stopped) ? "stopped" : "active";
+    if (nextState === "stopped" && node.registry.payload.profile_id === SOURCE_SESSION_PROFILE_ID
+      && (node.legacyPause || node.current !== "stopped")) {
+      throw new EffectRuntimeRequestError("Lifecycle-only source_session_v1 cannot migrate an execution pause before its lifecycle owner supports Resume");
+    }
+    if (!node.legacy && node.current === nextState) continue;
+    replacements.set(node, {...node.activation, schema_version: SCHEMA, state: nextState});
+  }
+  return {registries: registries.map(registry => ({path: registry.path,
+    changes: nodes.filter(node => node.registry === registry && replacements.has(node)).map(node => ({
+      index: node.index, goal_id: requireNonEmptyString(node.goal.id, "Goal id"),
+      activation: replacements.get(node)!,
+      ...(node.legacyPause ? {status: "active"} : {}),
+    })),
+  }))};
+}
diff --git a/loopx/control_plane/goals/activation_service.py b/loopx/control_plane/goals/activation_service.py
index edaec5db1e..f302017007 100644
--- a/loopx/control_plane/goals/activation_service.py
+++ b/loopx/control_plane/goals/activation_service.py
@@ -1,13 +1,19 @@
 from __future__ import annotations
 
-from contextlib import nullcontext
+from contextlib import ExitStack, nullcontext
 from dataclasses import dataclass
 from enum import Enum
 import hashlib
 from pathlib import Path
 from typing import Any
 
-from ..projects.registry_codec import project_registry_transaction
+from ..projects.registry_codec import (
+    activation_upgrade_registry_transaction,
+    project_registry_transaction,
+    read_activation_upgrade_registry,
+)
+from ..effect_runtime import effect_runtime_result
+from ...paths import GLOBAL_REGISTRY_FILENAME
 from ...configuration_transaction import configuration_payload_revision
 from ...file_lock import exclusive_cross_runtime_file_lock
 from ...global_registry import sync_project_registry_to_global
@@ -458,7 +464,6 @@ def set_goal_activation_state(
                 )
                 return payload
             if changed:
-                locked_goal.pop("activation_state", None)
                 locked_goal["activation"] = proposed_activation
                 transaction.commit(source_payload)
                 payload["written"] = True
@@ -520,3 +525,82 @@ def render_goal_activation_markdown(payload: dict[str, Any]) -> str:
     if payload.get("error"):
         lines.extend(["", f"Error: {payload.get('error')}"])
     return "\n".join(lines).rstrip() + "\n"
+
+
+# Upgrade-only IO reuses the lifecycle service and codec transactions. The
+# typed migration decoder is not another normal activation decision owner.
+def upgrade_goal_activation_registries(
+    paths: list[Path], *, execute: bool,
+) -> dict[str, Any]:
+    initial_snapshots = {path.resolve(): read_activation_upgrade_registry(path)
+                         for path in paths if path.exists()}
+    paths = sorted(initial_snapshots, key=lambda path: (
+        path.name == GLOBAL_REGISTRY_FILENAME
+        or (initial_snapshots[path].get("registry_role") or initial_snapshots[path].get("role")) == "global-local",
+        str(path),
+    ))
+    with ExitStack() as locks:
+        transactions = {}
+        payloads = {}
+        snapshots = []
+        for path in paths:
+            initial = initial_snapshots[path]
+            if execute:
+                transaction = locks.enter_context(activation_upgrade_registry_transaction(path))
+                transactions[str(path)] = transaction
+                initial = transaction.payload_copy()
+            payloads[str(path)] = initial
+            # Send only migration facts, never complete Goal configuration or
+            # business state. Normalized source paths are transport identity.
+            goals = []
+            for goal in initial.get("goals", []):
+                if not isinstance(goal, dict):
+                    raise ValueError("Registry Goal must be an object")
+                facts = {key: goal[key] for key in (
+                    "id", "goal_instance_id", "status", "activation_state",
+                ) if key in goal}
+                if "activation" in goal:
+                    activation = goal["activation"]
+                    facts["activation"] = ({key: activation[key] for key in ("schema_version", "state")
+                                             if key in activation}
+                                            if isinstance(activation, dict) else activation)
+                if goal.get("source_registry"):
+                    source = Path(goal["source_registry"]).expanduser().resolve()
+                    if source.exists() and source not in initial_snapshots:
+                        raise ValueError("Registered source route changed during activation upgrade; rediscover and retry")
+                    facts["source_registry"] = str(source)
+                goals.append(facts)
+            snapshots.append({"path": str(path), "payload": {"goals": goals,
+                              "profile_id": initial.get("profile_id")}})
+        plan = effect_runtime_result("goal.activation.migrate", {"registries": snapshots})
+        rows = plan["registries"]
+        if {row["path"] for row in rows} != set(payloads) or len(rows) != len(payloads):
+            raise ValueError("Goal activation migration returned different registry routes")
+        results = []
+        # Verify every required backup before committing any registry. A later
+        # commit failure fails the upgrade; completed stopped metadata is safe
+        # to retain, and a retry converges without replaying business effects.
+        for row in rows:
+            changed = bool(row["changes"])
+            payload = payloads[row["path"]]
+            for change in row["changes"]:
+                goal = payload["goals"][change["index"]]
+                if goal.get("id") != change["goal_id"]:
+                    raise ValueError("Goal activation upgrade changed the Goal identity")
+                prior_activation = goal.get("activation")
+                goal["activation"] = {**(prior_activation if isinstance(prior_activation, dict) else {}),
+                                      **change["activation"]}
+                goal.pop("activation_state", None)
+                if "status" in change:
+                    goal["status"] = change["status"]
+            result = {"registry": row["path"], "changed_goal_ids": [change["goal_id"] for change in row["changes"]],
+                      "status": "planned" if changed else "current"}
+            if changed and execute:
+                result["backup"] = str(transactions[row["path"]].backup_preimage())
+            results.append(result)
+        if execute:
+            for row, result in zip(rows, results, strict=True):
+                if result["status"] == "planned":
+                    transactions[row["path"]].commit(payloads[row["path"]])
+                    result["status"] = "upgraded"
+        return {"status": "upgraded" if execute else "planned", "results": results}
diff --git a/loopx/control_plane/projects/registry_codec.py b/loopx/control_plane/projects/registry_codec.py
index f05952425a..f4a86be3af 100644
--- a/loopx/control_plane/projects/registry_codec.py
+++ b/loopx/control_plane/projects/registry_codec.py
@@ -13,6 +13,9 @@
 from typing import Any, TypeVar
 
 from ..content_digest import ENVELOPED_SHA256_PATTERN
+from ..goals.activation import require_current_registry_goal_activation
+from ..runtime.document_io import fsync_state_directory
+from ..runtime.private_permissions import protect_private_path, require_private_path
 from ...file_lock import exclusive_cross_runtime_file_lock
 from ...paths import GLOBAL_REGISTRY_FILENAME
 
@@ -183,10 +186,22 @@ def _read_document(path: Path) -> _ProjectRegistryDocument:
     return _decode_document(path.read_bytes())
 
 
+def _require_current_goal_activation(payload: dict[str, Any]) -> None:
+    goals = payload.get("goals")
+    for goal in goals if isinstance(goals, list) else []:
+        if isinstance(goal, dict):
+            try:
+                require_current_registry_goal_activation(goal)
+            except ValueError as error:
+                raise ProjectRegistryProtocolError(str(error)) from error
+
+
 def load_project_registry(path: Path) -> dict[str, Any]:
     """Load either supported project-registry wire format."""
 
-    return _read_document(path.expanduser()).payload
+    payload = _read_document(path.expanduser()).payload
+    _require_current_goal_activation(payload)
+    return payload
 
 
 def decode_project_registry(raw_bytes: bytes) -> dict[str, Any]:
@@ -202,12 +217,14 @@ def decode_registry_snapshot(path: Path, raw_bytes: bytes) -> dict[str, Any]:
         payload = json.loads(raw_bytes)
         if not isinstance(payload, dict):
             raise ProjectRegistryError("global registry root must be a JSON object")
+        _require_current_goal_activation(payload)
         return payload
     payload = decode_project_registry(raw_bytes)
     require_runtime_compatible_project_registry(
         payload,
         operation="generic registry read",
     )
+    _require_current_goal_activation(payload)
     return payload
 
 
@@ -369,6 +386,7 @@ def commit(self, payload: dict[str, Any]) -> bool:
             raise RuntimeError("project registry transaction is already finished")
         if not isinstance(payload, dict):
             raise TypeError("project registry payload must be a JSON object")
+        _require_current_goal_activation(payload)
         wrote = not self._existed or payload != self._document.payload
         if not wrote:
             self._finished = True
@@ -404,6 +422,32 @@ def commit(self, payload: dict[str, Any]) -> bool:
         self._finished = True
         return True
 
+    def backup_preimage(self) -> Path:
+        """Verify a private original-byte backup before a caller's migration."""
+        if not self._existed or self._finished:
+            raise RuntimeError("registry backup requires an open existing transaction")
+        raw = self._document.raw_bytes
+        if self._path.read_bytes() != raw:
+            raise ProjectRegistryMutationError("Registry changed before preimage backup")
+        directory = self._path.parent / "backups" / "registry-preimages"
+        for parent in (directory.parent, directory):
+            if parent.is_symlink():
+                raise OSError("Registry backup path is redirected")
+        directory.mkdir(parents=True, exist_ok=True, mode=0o700)
+        protect_private_path(directory, directory=True)
+        target = directory / f"{self._path.name}.{hashlib.sha256(raw).hexdigest()}.bak"
+        if not target.exists():
+            with target.open("xb") as stream:
+                protect_private_path(target)
+                stream.write(raw)
+                stream.flush()
+                os.fsync(stream.fileno())
+            fsync_state_directory(directory)
+        require_private_path(target)
+        if target.read_bytes() != raw:
+            raise ProjectRegistryMutationError("Registry backup does not match original bytes")
+        return target
+
     def remove(self) -> bool:
         if self._finished:
             raise RuntimeError("project registry transaction is already finished")
@@ -446,6 +490,7 @@ def _registry_transaction(
     create_document: Callable[[], _ProjectRegistryDocument] | None,
     require_writer: Callable[[_ProjectRegistryDocument], None],
     agent_id: str | None = None,
+    allow_retired_activation: bool = False,
 ) -> Iterator[ProjectRegistryTransaction]:
     expanded = path.expanduser()
     with exclusive_cross_runtime_file_lock(
@@ -465,6 +510,8 @@ def _registry_transaction(
             document = create_document()
             mode = None
         require_writer(document)
+        if not allow_retired_activation:
+            _require_current_goal_activation(document.payload)
         yield ProjectRegistryTransaction(
             expanded,
             document=document,
@@ -473,6 +520,40 @@ def _registry_transaction(
         )
 
 
+@contextmanager
+def activation_upgrade_registry_transaction(path: Path) -> Iterator[ProjectRegistryTransaction]:
+    """The upgrade-only writer can read retired metadata, then commit current data.
+
+    Reuse each format's supported writer and exact-byte compensation; never
+    downgrade an envelope or bypass an unknown minimum writer protocol.
+    """
+    with _registry_transaction(path, operation="upgrade_goal_activation_metadata",
+                               create_document=None, require_writer=_require_activation_upgrade_writer,
+                               allow_retired_activation=True) as transaction:
+        yield transaction
+
+
+def _require_activation_upgrade_writer(document: _ProjectRegistryDocument) -> None:
+    if (document.format is _ProjectRegistryFormat.STRICT_ENVELOPE_V2
+            or document.payload.get("profile_id") == SOURCE_SESSION_PROFILE_ID):
+        _require_source_session_writer(document)
+    else:
+        _require_supported_writer(document)
+
+
+def read_activation_upgrade_registry(path: Path) -> dict[str, Any]:
+    """Upgrade-only read: retain wire/writer validation but accept retired fields."""
+    if not path.exists():
+        return {}
+    document = _read_document(path)
+    if path.name == GLOBAL_REGISTRY_FILENAME and document.format is not _ProjectRegistryFormat.LEGACY_OBJECT:
+        raise ProjectRegistryError("global registry root must be a JSON object")
+    _require_activation_upgrade_writer(document)
+    if not isinstance(document.payload.get("goals", []), list):
+        raise ProjectRegistryError("registry goals must be an array")
+    return document.payload
+
+
 def _created_document(
     create: Callable[[], dict[str, Any]],
     *,
diff --git a/loopx/semantics/project_registry_io.py b/loopx/semantics/project_registry_io.py
index fa8a2635e0..773f264e2a 100644
--- a/loopx/semantics/project_registry_io.py
+++ b/loopx/semantics/project_registry_io.py
@@ -33,6 +33,7 @@
         "decode_registry_snapshot",
         "load_project_registry",
         "load_registry",
+        "read_activation_upgrade_registry",
     }
 )
 APPROVED_WRITE_APIS = frozenset({"mutate_project_registry"})
@@ -40,6 +41,7 @@
     {
         "project_registry_transaction",
         "source_session_registry_transaction",
+        "activation_upgrade_registry_transaction",
     }
 )
 DIRECT_READ_APIS = frozenset(
diff --git a/tests/architecture/test_project_registry_io_census.py b/tests/architecture/test_project_registry_io_census.py
index 978b08690f..47437b4b12 100644
--- a/tests/architecture/test_project_registry_io_census.py
+++ b/tests/architecture/test_project_registry_io_census.py
@@ -54,6 +54,9 @@ def test_python_scan_separates_codec_calls_from_direct_json_io() -> None:
             "import json\n"
             "def update(registry_path, payload):\n"
             "    current = load_registry(registry_path)\n"
+            "    old = read_activation_upgrade_registry(registry_path)\n"
+            "    with activation_upgrade_registry_transaction(registry_path):\n"
+            "        pass\n"
             "    with source_session_registry_transaction(registry_path, "
             "operation='test'):\n"
             "        pass\n"
@@ -68,6 +71,8 @@ def test_python_scan_separates_codec_calls_from_direct_json_io() -> None:
         for row in observations
     ] == [
         ("codec_read", "load_registry"),
+        ("codec_read", "read_activation_upgrade_registry"),
+        ("codec_transaction", "activation_upgrade_registry_transaction"),
         ("codec_transaction", "source_session_registry_transaction"),
         ("direct_json_read", "json.loads"),
         ("direct_json_write", "atomic_write_json"),
diff --git a/tests/control_plane/test_goal_acceptance_cli.py b/tests/control_plane/test_goal_acceptance_cli.py
index 9041657584..d64096c6ec 100644
--- a/tests/control_plane/test_goal_acceptance_cli.py
+++ b/tests/control_plane/test_goal_acceptance_cli.py
@@ -4,6 +4,7 @@
 
 import hashlib
 import json
+from pathlib import Path
 import sys
 
 import pytest
@@ -114,7 +115,7 @@ def test_outcome_refinement_retains_legacy_active_goal_admission(acceptance_goal
     code, result = cli(*args)
     assert code == 0 and result["status"] == "applied", result
     before = cli("inspect")[1]
-    for status in ("paused", "achieved", "abandoned"):
+    for status in ("achieved", "abandoned"):
         registry["goals"][0]["status"] = status
         registry_path.write_text(json.dumps(registry), encoding="utf-8")
         code, denied = cli(*args)
@@ -144,6 +145,42 @@ def test_outcome_refinement_retains_legacy_active_goal_admission(acceptance_goal
         assert cli("inspect")[1] == before, "Stop rejects both saved operation replay and new refinements"
 
 
+def test_upgrade_retains_pause_then_owner_resume_enables_refinement(acceptance_goal):
+    project, _, cli, run = acceptance_goal
+    revision = cli("inspect")[1]["provider_revision"]
+    registry_path = project.parent / "registry.json"
+    registry = json.loads(registry_path.read_bytes())
+    registry["goals"][0].update(status="paused", activation_state="active")
+    registry_path.write_text(json.dumps(registry))
+    original, state_before = registry_path.read_bytes(), (project / "state.md").read_bytes()
+    code, preview = run("authority-archive", "upgrade")
+    assert code == 0, preview
+    assert registry_path.read_bytes() == original
+    code, upgraded = run("authority-archive", "upgrade", "--execute")
+    assert code == 0, upgraded
+    migrated = json.loads(registry_path.read_bytes())["goals"][0]
+    assert "activation_state" not in migrated and migrated.get("activation", {}).get("state") == "stopped", migrated
+    rows = upgraded["goal_activation_upgrade"]["results"]
+    assert any(Path(row["backup"]).read_bytes() == original for row in rows if row.get("backup"))
+    assert migrated["status"] == "active", "registration no longer owns pause"
+    document = project / "resumed-outcome.json"
+    document.write_text(json.dumps({"outcome_id": "export", "owner_agent_id": "agent-a",
+        "objective": "Produce the export", "requirements": [{"id": "content", "description": "Retain content"}],
+        "source": {"ref": "message:owner-resume", "summary": "The owner resumed the export"}}))
+    assert cli("inspect")[1]["provider_revision"] == revision, "metadata upgrade must preserve the canonical authority revision"
+    args = ("refine", "--agent-id", "agent-a", "--document", str(document),
+        "--expected-provider-revision", revision, "--operation-id", "resumed-outcome", "--execute")
+    assert cli(*args)[0] == 1, "upgrading metadata never resumes work"
+    code, resumed = run("goal-lifecycle", "--goal-id", "goal-acceptance", "--operation", "resume",
+        "--actor-kind", "owner", "--execute")
+    assert code == 0 and resumed["readback"]["verified"], resumed
+    assert cli(*args)[0] == 0
+    assert (project / "state.md").read_bytes() == state_before
+    assert run("goal-lifecycle", "--goal-id", "goal-acceptance", "--operation", "stop",
+        "--actor-kind", "owner", "--execute")[0] == 0
+    assert cli(*args)[0] == 1, "typed stop also guards operation replay"
+
+
 def test_outcome_requirement_identity_survives_configure_verify_and_readback(acceptance_goal):
     project, checks_file, cli, _ = acceptance_goal
     requirement_id = "export:content@v1"
diff --git a/tests/control_plane/test_goal_activation.py b/tests/control_plane/test_goal_activation.py
index 21f17619aa..19946a0083 100644
--- a/tests/control_plane/test_goal_activation.py
+++ b/tests/control_plane/test_goal_activation.py
@@ -544,19 +544,23 @@ def test_idempotent_execute_reconciles_drifted_global_projection(
     assert goal_activation_state(_goal(global_registry)) is GoalActivationState.STOPPED
 
 
-def test_stop_migrates_legacy_projected_activation_state(
+def test_upgrade_retires_stored_activation_alias_before_stop(
     connected_registries: tuple[Path, Path],
 ) -> None:
     source_registry, global_registry = connected_registries
     source_payload = load_registry(source_registry)
     registry_goals(source_payload)[0]["activation_state"] = "active"
     _write_json(source_registry, source_payload)
-    sync_project_registry_to_global(
-        registry_path=source_registry,
-        runtime_root_override=None,
-        goal_id="goal-one",
-        dry_run=False,
-    )
+    from loopx.control_plane.goals.activation_service import upgrade_goal_activation_registries
+
+    before = source_registry.read_bytes()
+    # The existing fail-safe Stop can still stop the shared projection while
+    # its source awaits upgrade. It must not silently migrate that source.
+    fallback = set_goal_activation_state(registry_path=global_registry, goal_id="goal-one",
+                                         state="stopped", actor_kind="owner", execute=True)
+    assert fallback["ok"] and fallback["readback"]["verified"]
+    assert source_registry.read_bytes() == before
+    upgrade_goal_activation_registries([source_registry, global_registry], execute=True)
 
     stopped = set_goal_activation_state(
         registry_path=global_registry,
@@ -605,6 +609,65 @@ def test_owner_confirmed_typed_action_stops_goal(
     assert goal_activation_state(_goal(global_registry)) is GoalActivationState.STOPPED
 
 
+def test_http_resume_after_upgrade_rejects_old_confirmation_and_replays_once(
+    connected_registries: tuple[Path, Path], tmp_path: Path,
+) -> None:
+    import http.client
+    from loopx.chat_server import ChatHTTPServer, ChatRequestHandler
+    from loopx.control_plane.goals.activation_service import upgrade_goal_activation_registries
+
+    source, shared = connected_registries
+    server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler)
+    server.registry_path = shared
+    server.runtime_root = shared.parent
+    server.runtime_root_override = str(shared.parent)
+    server.action_store = ChatActionStore(tmp_path / "http-actions")
+    server.action_service = ChatActionService(store=server.action_store, registry_path=shared)
+    server.verbose = False
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+
+    def post(path, body):
+        connection = http.client.HTTPConnection("127.0.0.1", server.server_port, timeout=10)
+        try:
+            connection.request("POST", path, json.dumps(body), {"Content-Type": "application/json",
+                "Origin": f"http://127.0.0.1:{server.server_port}"})
+            response = connection.getresponse()
+            return response.status, json.loads(response.read())
+        finally:
+            connection.close()
+
+    def preview(key):
+        code, result = post("/api/actions/preview", {"action_kind": "goal.lifecycle",
+            "summary": "Resume the Goal", "normalized_parameters": {"goal_id": "goal-one", "operation": "resume"},
+            "context": {"kind": "goal_directory"}, "idempotency_key": key})
+        assert code == 201, result
+        return result["proposal"]["proposal_id"]
+
+    try:
+        old = preview("before-upgrade")
+        registry = load_registry(source)
+        registry["goals"][0]["status"] = "paused"
+        _write_json(source, registry)
+        upgrade_goal_activation_registries([source, shared], execute=True)
+        before = source.read_bytes(), shared.read_bytes()
+        code, refused = post(f"/api/actions/{old}/apply", {})
+        assert code == 409 and refused["error_code"] == "action_stale", refused
+        assert (source.read_bytes(), shared.read_bytes()) == before
+        fresh = preview("after-upgrade")
+        code, applied = post(f"/api/actions/{fresh}/apply", {})
+        assert code == 200 and applied["proposal"]["status"] == "applied", applied
+        assert goal_activation_state(_goal(source)) is goal_activation_state(_goal(shared)) is GoalActivationState.ACTIVE
+        after = source.read_bytes(), shared.read_bytes()
+        code, replay = post(f"/api/actions/{fresh}/apply", {})
+        assert code == 200 and replay["proposal"]["receipt"] == applied["proposal"]["receipt"], replay
+        assert (source.read_bytes(), shared.read_bytes()) == after
+    finally:
+        server.shutdown()
+        thread.join(timeout=5)
+        server.server_close()
+
+
 @pytest.mark.parametrize(
     ("operation", "initial_state"),
     [
diff --git a/tests/control_plane/test_goal_activation_migration.py b/tests/control_plane/test_goal_activation_migration.py
new file mode 100644
index 0000000000..14f73be16b
--- /dev/null
+++ b/tests/control_plane/test_goal_activation_migration.py
@@ -0,0 +1,210 @@
+"""Retire old pause metadata without resuming work or losing registry content."""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from loopx.control_plane.goals.activation_service import upgrade_goal_activation_registries
+from loopx.control_plane.projects.registry_codec import (
+    SOURCE_SESSION_WRITER_PROTOCOL,
+    STRICT_SCHEMA_VERSION,
+    CURRENT_WRITER_PROTOCOL,
+    SOURCE_SESSION_SCHEMA_VERSION,
+    decode_project_registry,
+    load_registry,
+    load_project_registry,
+    project_registry_transaction,
+    source_session_registry_transaction,
+)
+
+
+@pytest.mark.parametrize("wire", ["object", "v1", "v2"])
+def test_upgrade_is_backup_first_lossless_and_idempotent(tmp_path, wire):
+    path = tmp_path / "registry.json"
+    goal = {"id": "example", "status": "active", "goal_instance_id": "original",
+            "metadata": {"nested": [None, False, "原文"], "opaque": 12345678901234567890},
+            "activation": {"schema_version": "loopx_goal_activation_v1", "state": "active",
+                           "updated_at": "2026-01-01T00:00:00Z", "reason": "Original reason",
+                           "metadata": {"large": 12345678901234567890, "fraction": 0.123456789}}}
+    payload = {"goals": [goal, {"id": "unrelated", "status": "archived"}],
+               "projects": [{"project_id": "project"}], "opaque": [1, {"keep": True}]}
+    factory = project_registry_transaction
+    if wire == "v2":
+        payload["profile_id"] = "source_session_v1"
+        factory = source_session_registry_transaction
+    with factory(path, operation="fixture", create=lambda: payload) as transaction:
+        transaction.commit(payload)
+    # A historical file is authored outside the current writer on purpose.
+    document = json.loads(path.read_bytes())
+    old = document if wire == "object" else payload
+    if wire != "v2":
+        old["goals"][0]["status"] = "paused"
+    old["goals"][0]["activation_state"] = "active"
+    if wire != "object":
+        from loopx.configuration_transaction import configuration_payload_revision
+        protocol = SOURCE_SESSION_WRITER_PROTOCOL if wire == "v2" else CURRENT_WRITER_PROTOCOL
+        schema = SOURCE_SESSION_SCHEMA_VERSION if wire == "v2" else STRICT_SCHEMA_VERSION
+        document = [{"schema_version": schema, "minimum_writer_protocol": protocol,
+                     "payload_sha256": configuration_payload_revision(old)}, old]
+    path.write_text(json.dumps(document, ensure_ascii=False, indent=3) + "\n")
+    original = path.read_bytes()
+    with pytest.raises(ValueError, match="activation metadata requires upgrade"):
+        load_project_registry(path)
+    with pytest.raises(ValueError, match="activation metadata requires upgrade"):
+        with factory(path, operation="ordinary writer"):
+            pytest.fail("ordinary writer must reject before business effects")
+    preview = upgrade_goal_activation_registries([path], execute=False)
+    assert preview["results"][0]["status"] == "planned"
+    assert path.read_bytes() == original and not (tmp_path / "backups").exists()
+    result = upgrade_goal_activation_registries([path], execute=True)
+    assert result["results"][0]["status"] == "upgraded"
+    assert Path(result["results"][0]["backup"]).read_bytes() == original
+    actual = load_project_registry(path)
+    expected = decode_project_registry(original)
+    expected["goals"][0]["status"] = "active"
+    expected["goals"][0].pop("activation_state")
+    expected["goals"][0]["activation"]["state"] = "active" if wire == "v2" else "stopped"
+    assert actual == expected
+    if wire != "object":
+        assert json.loads(path.read_bytes())[0]["minimum_writer_protocol"] == protocol
+    migrated = path.read_bytes()
+    assert upgrade_goal_activation_registries([path], execute=True)["results"][0]["status"] == "current"
+    assert path.read_bytes() == migrated
+
+
+@pytest.mark.parametrize("old_side", ["source", "shared"])
+def test_migration_preserves_pause_across_exact_source_projection(tmp_path, old_side):
+    source, shared = tmp_path / "registry.json", tmp_path / "registry.global.json"
+    for path in (source, shared):
+        goal = {"id": "same", "goal_instance_id": "lifetime", "status": "active"}
+        if path == shared:
+            goal["source_registry"] = str(source)
+        if path == (source if old_side == "source" else shared):
+            goal["status"] = "paused"
+        path.write_text(json.dumps({"goals": [goal]}))
+    result = upgrade_goal_activation_registries([shared, source], execute=True)
+    assert all(row["status"] == "upgraded" for row in result["results"])
+    for path in (source, shared):
+        goal = load_registry(path)["goals"][0]
+        assert goal["activation"]["state"] == "stopped" and goal["status"] == "active"
+
+
+def test_old_projection_cannot_stop_a_recreated_goal(tmp_path):
+    source, shared = tmp_path / "registry.json", tmp_path / "registry.global.json"
+    source.write_text(json.dumps({"goals": [{"id": "same", "goal_instance_id": "new", "status": "active"}]}))
+    shared.write_text(json.dumps({"goals": [{"id": "same", "goal_instance_id": "old", "status": "paused",
+                                          "source_registry": str(source)}]}))
+    before = source.read_bytes()
+    upgrade_goal_activation_registries([source, shared], execute=True)
+    assert source.read_bytes() == before
+    assert load_registry(shared)["goals"][0]["activation"]["state"] == "stopped"
+
+
+def test_changed_source_route_requires_rediscovery_before_any_write(tmp_path):
+    selected, changed, shared = tmp_path / "old.json", tmp_path / "new.json", tmp_path / "registry.global.json"
+    for path in (selected, changed):
+        path.write_text(json.dumps({"goals": [{"id": "example", "status": "paused"}]}))
+    shared.write_text(json.dumps({"goals": [{"id": "example", "status": "paused", "source_registry": str(changed)}]}))
+    before = {path: path.read_bytes() for path in (selected, changed, shared)}
+    with pytest.raises(ValueError, match="rediscover and retry"):
+        upgrade_goal_activation_registries([selected, shared], execute=True)
+    assert {path: path.read_bytes() for path in before} == before and not (tmp_path / "backups").exists()
+
+
+def test_lifecycle_only_profile_never_gains_execution_admission_from_migration(tmp_path):
+    path = tmp_path / "registry.json"
+    payload = {"profile_id": "source_session_v1", "goals": [{"id": "example", "status": "active"}]}
+    with source_session_registry_transaction(path, operation="fixture", create=lambda: payload) as transaction:
+        transaction.commit(payload)
+    from loopx.configuration_transaction import configuration_payload_revision
+    document = json.loads(path.read_bytes())
+    document[1]["goals"][0]["status"] = "paused"
+    document[0]["payload_sha256"] = configuration_payload_revision(document[1])
+    path.write_text(json.dumps(document))
+    before = path.read_bytes()
+    with pytest.raises(RuntimeError, match="lifecycle owner supports Resume"):
+        upgrade_goal_activation_registries([path], execute=False)
+    assert path.read_bytes() == before and not (tmp_path / "backups").exists()
+
+
+@pytest.mark.parametrize("alias", [None, "active", "stopped"])
+def test_alias_retirement_preserves_current_stop(tmp_path, alias):
+    path = tmp_path / "registry.json"
+    path.write_text(json.dumps({"goals": [{"id": "example", "activation_state": alias,
+        "activation": {"schema_version": "loopx_goal_activation_v1", "state": "stopped"}}]}))
+    upgrade_goal_activation_registries([path], execute=True)
+    goal = load_registry(path)["goals"][0]
+    assert "activation_state" not in goal and goal["activation"]["state"] == "stopped"
+
+
+@pytest.mark.parametrize("stored_alias", [False, True])
+def test_null_activation_retains_the_existing_active_default(tmp_path, stored_alias):
+    path = tmp_path / "registry.json"
+    goal = {"id": "example", "status": "active", "activation": None}
+    if stored_alias:
+        goal["activation_state"] = None
+    path.write_text(json.dumps({"goals": [goal]}))
+    before = path.read_bytes()
+    upgraded = upgrade_goal_activation_registries([path], execute=True)
+    actual = load_registry(path)["goals"][0]
+    if stored_alias:
+        assert "activation_state" not in actual and actual["activation"]["state"] == "active"
+    else:
+        assert upgraded["results"][0]["status"] == "current" and path.read_bytes() == before
+
+
+def test_failed_shared_commit_can_resume_without_losing_the_source_pause(tmp_path, monkeypatch):
+    from loopx.control_plane.projects.registry_codec import ProjectRegistryTransaction
+    source, shared = tmp_path / "registry.json", tmp_path / "registry.global.json"
+    source.write_text(json.dumps({"goals": [{"id": "example", "status": "active"}]}))
+    shared.write_text(json.dumps({"goals": [{"id": "example", "status": "paused", "source_registry": str(source)}]}))
+    commit = ProjectRegistryTransaction.commit
+
+    def fail_shared(self, payload):
+        if self.source_read_witness()["path"] == str(shared):
+            raise OSError("Synthetic shared commit failure")
+        return commit(self, payload)
+
+    with monkeypatch.context() as patch:
+        patch.setattr(ProjectRegistryTransaction, "commit", fail_shared)
+        with pytest.raises(OSError, match="Synthetic shared commit failure"):
+            upgrade_goal_activation_registries([source, shared], execute=True)
+    first_source = source.read_bytes()
+    assert load_registry(source)["goals"][0]["activation"]["state"] == "stopped"
+    upgrade_goal_activation_registries([source, shared], execute=True)
+    assert source.read_bytes() == first_source
+    assert load_registry(shared)["goals"][0]["activation"]["state"] == "stopped"
+
+
+def test_backup_failure_prevents_every_registry_commit(tmp_path, monkeypatch):
+    from loopx.control_plane.projects.registry_codec import ProjectRegistryTransaction
+    paths = [tmp_path / "a.json", tmp_path / "b.json"]
+    for path in paths:
+        path.write_text(json.dumps({"goals": [{"id": path.stem, "status": "paused"}]}))
+    before = {path: path.read_bytes() for path in paths}
+    backup = ProjectRegistryTransaction.backup_preimage
+
+    def fail_second(self):
+        if self.source_read_witness()["path"] == str(paths[1]):
+            raise OSError("Synthetic backup failure")
+        return backup(self)
+
+    monkeypatch.setattr(ProjectRegistryTransaction, "backup_preimage", fail_second)
+    with pytest.raises(OSError, match="Synthetic backup failure"):
+        upgrade_goal_activation_registries(paths, execute=True)
+    assert {path: path.read_bytes() for path in paths} == before
+
+
+@pytest.mark.parametrize("field,value", [("status", "paused"), ("status", "stopped"), ("activation_state", "active")])
+def test_normal_writer_cannot_reintroduce_retired_pause_metadata(tmp_path, field, value):
+    path = tmp_path / "registry.json"
+    path.write_text(json.dumps({"goals": [{"id": "example", "status": "active"}]}))
+    before = path.read_bytes()
+    with project_registry_transaction(path, operation="fixture") as transaction:
+        payload = transaction.payload_copy()
+        payload["goals"][0][field] = value
+        with pytest.raises(ValueError, match="activation metadata requires upgrade"):
+            transaction.commit(payload)
+    assert path.read_bytes() == before