diff --git a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md
index 59effe5c8b..19e666af30 100644
--- a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md
+++ b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md
@@ -967,9 +967,20 @@ Goal path resolver; real File/SQLite import and original-receipt recovery pass
 with the four old normal producer modules physically absent in a disposable
 package. This qualifies that command's loading boundary, not every other CLI
 caller or removal of those modules. Keep T4 retirement on actual callers: the
-retained Python prose-write guard still serves live callers and its obligation
-must survive adapter removal. This partial stage does not retire the supported
-old writer or qualify a released default.
+retained prose-write guard now belongs to the existing source/fence boundary;
+the source partition projector has one shared owner, with compatible old
+imports. Configuration and direct guard checks run without both capture modules.
+Pure runtime-root routing and its actual canonical Todo, acceptance, Chat and
+CLI callers now use the existing paths owner, with an identical compatibility
+reexport for capture callers. Source and fresh-wheel File/SQLite App HTTP import,
+process restart, original-operation recovery, later canonical writes and prose
+checks pass the same oracle with both capture modules present and physically
+absent. Source/maintenance identity, Todo/handoff invariance, JSON types,
+project-relative routing and failed-write behavior remain unchanged. Historical
+outbox disposition and independent data/history recovery retain their separate
+qualifications below; authorized reactivation and remaining App writer callers
+stay open. This stage does not retire the supported old writer or qualify a
+released default.
 The shared TS source check admits runtime directory aliases by physical identity,
 preserving the original snapshot and foreign-source rejection. File/SQLite HTTP
 joins alias-bound import, externally configured registry backup, later writes and original-operation recovery; a
diff --git a/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md b/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md
index 42909dcdb1..e4190adb85 100644
--- a/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md
+++ b/docs/architecture/rfcs/loopx-overall-roadmap-v0.zh-CN.md
@@ -628,6 +628,17 @@ owned Host 进程、未晋升源的原生租约和 File/SQLite 验证：进程
 完整状态重新激活继续开放；协调源备份验证不结算完整恢复。详见
 [冷源导入与支持边界](../../reference/local-authority-provider-selection.md)。
 
+仍有调用方的 prose 写入保护现归属于既有 source/fence 边界，source partition
+projector 共用同一 owner，并保留旧导入兼容。配置和直接 guard 可在两个 capture
+模块物理缺席时运行。纯 runtime-root 解析及其真实 canonical Todo、terminal
+lifecycle、acceptance、Chat 和 CLI 调用方现复用既有 paths owner；旧 capture adapter
+兼容导出同一个函数。源码与 fresh wheel 的 File/SQLite App HTTP 使用同一 oracle，
+在两个 capture 模块均存在/均物理缺席时验证导入、进程重启、原操作恢复、后续
+canonical 写入及 prose 保护。保留相对项目路由、显式 override 优先、source/maintenance
+身份、Todo/handoff 不变、JSON 类型与失败不写；历史 outbox 处置和独立数据／历史
+恢复保留下述独立资格，授权重新激活及其余 App writer 调用方继续开放，不据此
+删除受支持旧 writer 或结算发布默认资格。
+
 冷旧源保留检查点：本项目全量备份现盘点注册的自定义状态与来源 registry 路由。
 真实 CLI 冷导入→新增 canonical 写入→全量备份→独立静态解包现已串联验证
 File/SQLite 完整 head、全部 committed journal、未引用的归档要求、存储配置和原始
diff --git a/docs/reference/local-authority-provider-selection.md b/docs/reference/local-authority-provider-selection.md
index 8c2a6d5faa..b717fa57ff 100644
--- a/docs/reference/local-authority-provider-selection.md
+++ b/docs/reference/local-authority-provider-selection.md
@@ -406,11 +406,21 @@ the existing Goal path resolver; its File/SQLite import and original-receipt
 recovery run with `todos.py`, `bootstrap.py`, `runtime_shadow_writer_adapter.py`
 and `local_authority_shadow_outbox.py` physically absent in a disposable package.
 This proves that command's independence, not that other commands or supported
-writers can lose those files. Retained prose-write guards have live callers and
-must move to their owning boundary before their adapter is removed.
-The packaged App uses the real File/SQLite backend; App loading with those old
-modules absent remains a separate acceptance. This stage does not qualify
-supported old-writer retirement,
+writers can lose those files. The retained prose-write guard now belongs to the
+existing source/fence boundary and shares the source partition projector with
+capture. Its old import remains compatible. Configuration and the guard load
+with both capture modules absent. Pure runtime-root routing now belongs to
+`paths.effective_runtime_root`; canonical Todo, terminal lifecycle, acceptance,
+Chat and CLI callers import that owner directly. The old adapter reexports the
+same function for supported capture callers. Source and fresh-wheel File/SQLite
+HTTP import, process restart, original-operation recovery, later writes and
+canonical prose checks use the same oracle with both capture modules present
+and physically absent. Project-relative routes, explicit override precedence,
+source identity, maintenance, Todo/handoff, JSON type and failed-write protections
+remain intact. Historical outbox disposition and independent data/history
+recovery retain their separate qualifications; authorized reactivation and
+remaining App writer callers still need their original acceptance. This stage
+does not qualify supported old-writer retirement,
 release default or historical support cutoff.
 Source checks accept registered runtime directory aliases only when both paths
 resolve to the same physical directory. Retargeting an alias after preview still
diff --git a/loopx/chat_todo_actions.py b/loopx/chat_todo_actions.py
index c61b3bbf7e..32731aeb64 100644
--- a/loopx/chat_todo_actions.py
+++ b/loopx/chat_todo_actions.py
@@ -8,7 +8,7 @@
 from .control_plane.coordination.local_authority import (
     LocalCoordinationAuthorityUnavailable, read_canonical_todos_if_promoted,
 )
-from .control_plane.coordination.local_authority_shadow_adapter import effective_runtime_root
+from .paths import effective_runtime_root
 from .control_plane.todos.provider_projection import projection_delivery_requires_ack
 
 
diff --git a/loopx/cli.py b/loopx/cli.py
index 92f1197ff4..1d3f4d4436 100644
--- a/loopx/cli.py
+++ b/loopx/cli.py
@@ -73,7 +73,7 @@
     build_periodic_report_post_writeback_projection,
     periodic_report_post_writeback_hooks_for_goal,
 )
-from .control_plane.coordination.local_authority_shadow_adapter import (
+from .paths import (
     effective_runtime_root,
 )
 from .capabilities.semantic_preference.cli import (
diff --git a/loopx/cli_commands/authority_shadow.py b/loopx/cli_commands/authority_shadow.py
index e62a3b4f67..e5756547d2 100644
--- a/loopx/cli_commands/authority_shadow.py
+++ b/loopx/cli_commands/authority_shadow.py
@@ -7,10 +7,10 @@
 from ..control_plane.coordination.local_authority_shadow_adapter import (
     CLI_DRAIN_LOCK_TIMEOUT_SECONDS,
     drain_local_authority_shadow_outbox,
-    effective_runtime_root,
     local_authority_shadow_status,
 )
 from ..control_plane.coordination.local_authority_shadow_outbox import OutboxError
+from ..paths import effective_runtime_root
 from ..file_lock import LockAcquireTimeoutError
 
 
diff --git a/loopx/cli_commands/todo_event.py b/loopx/cli_commands/todo_event.py
index 4869e0bf87..4114c61d9b 100644
--- a/loopx/cli_commands/todo_event.py
+++ b/loopx/cli_commands/todo_event.py
@@ -14,7 +14,7 @@
 from ..file_lock import lock_timeout_error_fields
 from ..control_plane.coordination.legacy_writer_fence import LegacyCoordinationWriterFenced
 from ..control_plane.coordination.shadow_management import ShadowManagementError
-from ..control_plane.coordination.runtime_shadow_writer_adapter import ActiveStateAuthorityMutationError
+from ..control_plane.coordination.legacy_writer_fence import ActiveStateAuthorityMutationError
 from ..control_plane.coordination.local_authority import LocalCoordinationAuthorityUnavailable
 from .todo_argument_validation import TodoClaimArgumentError
 
diff --git a/loopx/cli_runtime.py b/loopx/cli_runtime.py
index 9db8f2ae37..a2d0c06ff8 100644
--- a/loopx/cli_runtime.py
+++ b/loopx/cli_runtime.py
@@ -340,7 +340,7 @@ def _dispatch_common_command(
 		return 0
 	if args.command == "delegation":
 		from .cli_commands.delegation import handle_delegation
-		from .control_plane.coordination.local_authority_shadow_adapter import (
+		from .paths import (
 			effective_runtime_root,
 		)
 
@@ -411,7 +411,7 @@ def _dispatch_common_command(
 		)
 		from .cli_commands.todo import handle_todo_command
 		from .cli_rollout import append_cli_rollout_event
-		from .control_plane.coordination.local_authority_shadow_adapter import (
+		from .paths import (
 			effective_runtime_root,
 		)
 
diff --git a/loopx/control_plane/coordination/legacy_writer_fence.py b/loopx/control_plane/coordination/legacy_writer_fence.py
index 0954876f69..5161a6883a 100644
--- a/loopx/control_plane/coordination/legacy_writer_fence.py
+++ b/loopx/control_plane/coordination/legacy_writer_fence.py
@@ -19,7 +19,7 @@
 from ...file_lock import exclusive_cross_runtime_file_lock
 from ..projects.registry_codec import load_registry
 from ...paths import resolve_runtime_root
-from ...registry import registry_goals, resolve_state_file
+from ...registry import find_registry_goal, registry_goals, resolve_state_file
 from ..goals.active_state_metadata import parse_state_frontmatter
 from .shadow_management import (
     ShadowManagementError, read_shadow_bootstrap_source_path, read_shadow_management_state, require_shadow_primary_write_allowed,
@@ -192,6 +192,58 @@ def require_registry_source_write_allowed(
         require_legacy_coordination_write_allowed(runtime_root=registered_root, goal_id=goal_id)
 
 
+class ActiveStateAuthorityMutationError(ValueError):
+    """A prose-only writer attempted to change canonical coordination state."""
+
+    code = "active_state_authority_mutation_forbidden"
+    payload = {"primary_writeback_preserved": True}
+
+
+def require_prose_state_write_allowed(
+    *, registry_path: Path, runtime_root: Path, goal_id: str, state_path: Path,
+    original_text: str, planned_text: str,
+) -> None:
+    """Under S, enforce source maintenance and the owned prose-only invariant."""
+
+    require_registry_source_write_allowed(
+        registry_path=registry_path, runtime_root=runtime_root, goal_id=goal_id,
+        state_file=state_path, canonical_mutation=False,
+    )
+
+    from ...rollout_event_log import load_rollout_events, rollout_event_log_path
+    from ..todos.todo_index import MAX_TODO_INDEX_ROLLOUT_EVENTS_PER_GOAL
+    from .runtime_shadow import todo_partition_projector
+    from .local_authority_shadow_projection import canonical_bytes, partition_comparison_view
+
+    try:
+        goal = find_registry_goal(load_registry(registry_path), goal_id)
+        events = load_rollout_events(
+            rollout_event_log_path(runtime_root, goal_id),
+            limit=MAX_TODO_INDEX_ROLLOUT_EVENTS_PER_GOAL,
+        )
+        projector = todo_partition_projector(
+            goal, state_path=state_path, rollout_events=events,
+        )
+        # Todo projections include the read-time resume-condition evaluation
+        # clock. Compare the same semantic view used by shadow continuity
+        # digests so an owned prose update is not rejected merely because the
+        # two projections were evaluated milliseconds apart.
+        original_projection = partition_comparison_view(projector(original_text))
+        planned_projection = partition_comparison_view(projector(planned_text))
+        # Python structural equality aliases JSON booleans with integers.
+        # Use the existing exact-byte identity contract before authorizing IO.
+        if canonical_bytes(original_projection) != canonical_bytes(planned_projection):
+            raise ActiveStateAuthorityMutationError(
+                "prose update would change canonical Todo or handoff state"
+            )
+    except ActiveStateAuthorityMutationError:
+        raise
+    except Exception as error:
+        raise ActiveStateAuthorityMutationError(
+            "prose update cannot prove that canonical coordination state is unchanged"
+        ) from error
+
+
 def require_legacy_state_replacement_allowed(
     *, runtime_root: Path, goal_id: str, goal: dict[str, Any] | None,
 ) -> None:
diff --git a/loopx/control_plane/coordination/local_authority_shadow_adapter.py b/loopx/control_plane/coordination/local_authority_shadow_adapter.py
index 19da5ff9f6..29b02bd55f 100644
--- a/loopx/control_plane/coordination/local_authority_shadow_adapter.py
+++ b/loopx/control_plane/coordination/local_authority_shadow_adapter.py
@@ -7,13 +7,12 @@
 from __future__ import annotations
 
 import sys
-from collections.abc import Mapping
 from dataclasses import asdict, dataclass, field, fields
 from pathlib import Path
 from typing import Any
 
 from ..projects.registry_codec import load_registry
-from ...paths import resolve_runtime_root
+from ...paths import effective_runtime_root as effective_runtime_root, resolve_runtime_root
 from ...registry import find_registry_goal
 from ..effect_runtime import effect_runtime_result
 from . import local_authority_shadow_outbox as outbox
@@ -25,9 +24,11 @@
 )
 from .local_authority_shadow_projection import (
     head_digest,
-    todo_partition_projection,
 )
-from .runtime_shadow import resolve_coordination_runtime_shadow_config, capture_todo_archive_dependencies
+from .runtime_shadow import (
+    resolve_coordination_runtime_shadow_config,
+    todo_partition_projector as todo_partition_projector,
+)
 from .shadow_management import read_shadow_capture_binding, shadow_management_state_path
 from .shadow_goal_scope import shadow_goal_scope
 
@@ -35,24 +36,6 @@
 from .runtime_shadow import local_authority_shadow_summary
 
 
-def effective_runtime_root(
-    registry_path: Path,
-    runtime_root_override: str | Path | None,
-) -> Path:
-    """Resolve the one runtime root every writer hook of a CLI call must share.
-
-    ``--runtime-root`` wins when given; otherwise the registry's
-    ``common_runtime_root`` applies, and a relative value resolves against the
-    registry's project root rather than the caller's working directory. Todo,
-    follow-up, handoff-mode, and task-lease hooks all consume this value so one
-    goal never splits into two candidate lineages.
-    """
-
-    registry = load_registry(registry_path)
-    override = str(runtime_root_override) if runtime_root_override is not None else None
-    return resolve_runtime_root(registry, override, registry_path=registry_path)
-
-
 # ---------------------------------------------------------------------------
 # Transaction-bound outbox drain (Stage 2C second half plumbing).
 #
@@ -129,34 +112,6 @@ def to_payload(self) -> dict[str, Any]:
         return payload
 
 
-def todo_partition_projector(
-    goal: Mapping[str, Any] | None,
-    *,
-    state_path: Path,
-    rollout_events: list[dict[str, Any]] | None = None,
-) -> outbox.TodoPartitionProjector:
-    """Production projector: parse active-state text into the todos partition."""
-
-    from ...control_plane.todos.handoff_mode import goal_handoff_mode
-    from ..todos.goal_todo_projection import project_goal_todo_items
-
-    goal_record = dict(goal) if isinstance(goal, Mapping) else None
-    events = list(rollout_events or [])
-
-    def project(state_text: str) -> dict[str, Any]:
-        return todo_partition_projection(
-            handoff_mode=goal_handoff_mode(state_text),
-            todos=capture_todo_archive_dependencies(project_goal_todo_items(
-                goal_record,
-                state_text=state_text,
-                state_path=state_path,
-                rollout_events=events,
-            ), state_text),
-        )
-
-    return project
-
-
 def read_local_authority_shadow(
     *,
     runtime_root: Path,
diff --git a/loopx/control_plane/coordination/runtime_shadow.py b/loopx/control_plane/coordination/runtime_shadow.py
index 9b0a1b65ae..2e74450870 100644
--- a/loopx/control_plane/coordination/runtime_shadow.py
+++ b/loopx/control_plane/coordination/runtime_shadow.py
@@ -262,6 +262,35 @@ def capture_todo_archive_dependencies(todos: list[dict[str, Any]], state_text: s
     return result
 
 
+def todo_partition_projector(
+    goal: Mapping[str, Any] | None,
+    *,
+    state_path: Path,
+    rollout_events: list[dict[str, Any]] | None = None,
+) -> Callable[[str], dict[str, Any]]:
+    """Production projector: parse active-state text into the todos partition."""
+
+    from ..todos.handoff_mode import goal_handoff_mode
+    from .local_authority_shadow_projection import todo_partition_projection
+    from ..todos.goal_todo_projection import project_goal_todo_items
+
+    goal_record = dict(goal) if isinstance(goal, Mapping) else None
+    events = list(rollout_events or [])
+
+    def project(state_text: str) -> dict[str, Any]:
+        return todo_partition_projection(
+            handoff_mode=goal_handoff_mode(state_text),
+            todos=capture_todo_archive_dependencies(project_goal_todo_items(
+                goal_record,
+                state_text=state_text,
+                state_path=state_path,
+                rollout_events=events,
+            ), state_text),
+        )
+
+    return project
+
+
 def build_runtime_shadow_source_snapshot(
     *, goal: Mapping[str, Any], runtime_root: Path, state_path: Path,
     registry_path: Path,
diff --git a/loopx/control_plane/coordination/runtime_shadow_writer_adapter.py b/loopx/control_plane/coordination/runtime_shadow_writer_adapter.py
index ebc74eccd8..d63a35bae9 100644
--- a/loopx/control_plane/coordination/runtime_shadow_writer_adapter.py
+++ b/loopx/control_plane/coordination/runtime_shadow_writer_adapter.py
@@ -6,68 +6,17 @@
 from pathlib import Path
 from typing import Any
 
-from ...history import load_registry
-from ...registry import find_registry_goal
 from . import local_authority_shadow_outbox as outbox
 from .local_authority_shadow_projection import LEASE_PARTITION
+from .legacy_writer_fence import (
+    ActiveStateAuthorityMutationError as ActiveStateAuthorityMutationError,
+    require_prose_state_write_allowed as require_prose_state_write_allowed,
+)
 from .runtime_shadow import resolve_coordination_runtime_shadow_config
 from .shadow_goal_scope import shadow_goal_scope
 from .shadow_management import ShadowManagementError, read_shadow_capture_binding, require_shadow_primary_write_allowed
 
 
-class ActiveStateAuthorityMutationError(ValueError):
-    """A prose-only writer attempted to change canonical coordination state."""
-
-    code = "active_state_authority_mutation_forbidden"
-    payload = {"primary_writeback_preserved": True}
-
-
-def require_prose_state_write_allowed(
-    *, registry_path: Path, runtime_root: Path, goal_id: str, state_path: Path,
-    original_text: str, planned_text: str,
-) -> None:
-    """Under S, enforce source maintenance and the owned prose-only invariant."""
-
-    from .legacy_writer_fence import require_registry_source_write_allowed
-    require_registry_source_write_allowed(
-        registry_path=registry_path, runtime_root=runtime_root, goal_id=goal_id,
-        state_file=state_path, canonical_mutation=False,
-    )
-
-    from ...rollout_event_log import load_rollout_events, rollout_event_log_path
-    from ..todos.todo_index import MAX_TODO_INDEX_ROLLOUT_EVENTS_PER_GOAL
-    from .local_authority_shadow_adapter import todo_partition_projector
-    from .local_authority_shadow_projection import canonical_bytes, partition_comparison_view
-
-    try:
-        goal = find_registry_goal(load_registry(registry_path), goal_id)
-        events = load_rollout_events(
-            rollout_event_log_path(runtime_root, goal_id),
-            limit=MAX_TODO_INDEX_ROLLOUT_EVENTS_PER_GOAL,
-        )
-        projector = todo_partition_projector(
-            goal, state_path=state_path, rollout_events=events,
-        )
-        # Todo projections include the read-time resume-condition evaluation
-        # clock. Compare the same semantic view used by shadow continuity
-        # digests so an owned prose update is not rejected merely because the
-        # two projections were evaluated milliseconds apart.
-        original_projection = partition_comparison_view(projector(original_text))
-        planned_projection = partition_comparison_view(projector(planned_text))
-        # Python structural equality aliases JSON booleans with integers.
-        # Use the existing exact-byte identity contract before authorizing IO.
-        if canonical_bytes(original_projection) != canonical_bytes(planned_projection):
-            raise ActiveStateAuthorityMutationError(
-                "prose update would change canonical Todo or handoff state"
-            )
-    except ActiveStateAuthorityMutationError:
-        raise
-    except Exception as error:
-        raise ActiveStateAuthorityMutationError(
-            "prose update cannot prove that canonical coordination state is unchanged"
-        ) from error
-
-
 def begin_todo_runtime_shadow_capture(
     *,
     registry_path: Path,
diff --git a/loopx/control_plane/goals/acceptance.py b/loopx/control_plane/goals/acceptance.py
index f0fcb1eba5..84f4ab3ae1 100644
--- a/loopx/control_plane/goals/acceptance.py
+++ b/loopx/control_plane/goals/acceptance.py
@@ -15,7 +15,7 @@
 
 from ...agent_registry import load_goal_from_registry, registered_agent_ids_for_goal
 from ..coordination.local_authority import local_authority_is_promoted
-from ..coordination.local_authority_shadow_adapter import effective_runtime_root
+from ...paths import effective_runtime_root
 from ..effect_runtime import effect_runtime_result
 from ..todos.completion_validation import (
     _resolve_completion_validation_workspace,
diff --git a/loopx/control_plane/todos/provider_terminal_lifecycle.py b/loopx/control_plane/todos/provider_terminal_lifecycle.py
index fb70bc5f08..aab3e3e044 100644
--- a/loopx/control_plane/todos/provider_terminal_lifecycle.py
+++ b/loopx/control_plane/todos/provider_terminal_lifecycle.py
@@ -21,7 +21,7 @@
     LocalCoordinationAuthorityUnavailable,
     read_canonical_todos_if_promoted,
 )
-from ..coordination.local_authority_shadow_adapter import effective_runtime_root
+from ...paths import effective_runtime_root
 from ..effect_runtime import (
     CANONICAL_AUTHORITY_WRITE_TIMEOUT_SECONDS,
     effect_runtime_result,
diff --git a/loopx/control_plane/work_items/team_plan_adapter.py b/loopx/control_plane/work_items/team_plan_adapter.py
index 6560c02e08..1fbb0bd612 100644
--- a/loopx/control_plane/work_items/team_plan_adapter.py
+++ b/loopx/control_plane/work_items/team_plan_adapter.py
@@ -25,7 +25,7 @@
 from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result
 from ..runtime.public_safety import validate_public_safe_value
 from ..coordination.local_authority import read_canonical_todos_if_promoted
-from ..coordination.local_authority_shadow_adapter import effective_runtime_root
+from ...paths import effective_runtime_root
 from ..coordination.legacy_writer_fence import (
     legacy_coordination_todo_lock_path, legacy_todo_write_transaction,
 )
diff --git a/loopx/feedback.py b/loopx/feedback.py
index 22187e7aee..d342a63979 100644
--- a/loopx/feedback.py
+++ b/loopx/feedback.py
@@ -3,7 +3,7 @@
 from contextlib import nullcontext
 
 from .file_lock import exclusive_cross_runtime_file_lock
-from .control_plane.coordination.runtime_shadow_writer_adapter import require_prose_state_write_allowed
+from .control_plane.coordination.legacy_writer_fence import require_prose_state_write_allowed
 from .control_plane.runtime.document_io import atomic_write_state_text
 
 import re
diff --git a/loopx/paths.py b/loopx/paths.py
index 7dbbfbe0cb..7188752190 100644
--- a/loopx/paths.py
+++ b/loopx/paths.py
@@ -326,6 +326,28 @@ def resolve_runtime_root(
     return registry_project_root(registry_path) / runtime_root
 
 
+def effective_runtime_root(
+    registry_path: Path,
+    runtime_root_override: str | Path | None,
+) -> Path:
+    """Resolve the one runtime root every writer hook of a CLI call must share.
+
+    ``--runtime-root`` wins when given; otherwise the registry's
+    ``common_runtime_root`` applies, and a relative value resolves against the
+    registry's project root rather than the caller's working directory. Todo,
+    follow-up, handoff-mode, and task-lease hooks all consume this value so one
+    goal never splits into two candidate lineages.
+    """
+
+    # The codec imports path constants; load it at this call boundary to avoid
+    # a module cycle. No capture or drain owner participates in route selection.
+    from .control_plane.projects.registry_codec import load_registry
+
+    registry = load_registry(registry_path)
+    override = str(runtime_root_override) if runtime_root_override is not None else None
+    return resolve_runtime_root(registry, override, registry_path=registry_path)
+
+
 def rel_or_abs(path: Path, root: Path) -> str:
     try:
         return str(path.relative_to(root))
diff --git a/loopx/semantics/project_registry_io_manifest_v1.json b/loopx/semantics/project_registry_io_manifest_v1.json
index 32e6645289..2f2b8b8761 100644
--- a/loopx/semantics/project_registry_io_manifest_v1.json
+++ b/loopx/semantics/project_registry_io_manifest_v1.json
@@ -1079,23 +1079,23 @@
     },
     {
       "site": "loopx/control_plane/coordination/legacy_writer_fence.py::<module>.legacy_todo_write_transaction::codec_read:load_registry#1",
-      "line": 310,
+      "line": 362,
       "column": 9,
       "kind": "codec_read",
       "api": "load_registry",
       "classification": "codec_api"
     },
     {
-      "site": "loopx/control_plane/coordination/legacy_writer_fence.py::<module>.require_registry_source_write_allowed::codec_read:load_registry#1",
-      "line": 161,
-      "column": 16,
+      "site": "loopx/control_plane/coordination/legacy_writer_fence.py::<module>.require_prose_state_write_allowed::codec_read:load_registry#1",
+      "line": 219,
+      "column": 35,
       "kind": "codec_read",
       "api": "load_registry",
       "classification": "codec_api"
     },
     {
-      "site": "loopx/control_plane/coordination/local_authority_shadow_adapter.py::<module>.effective_runtime_root::codec_read:load_registry#1",
-      "line": 51,
+      "site": "loopx/control_plane/coordination/legacy_writer_fence.py::<module>.require_registry_source_write_allowed::codec_read:load_registry#1",
+      "line": 161,
       "column": 16,
       "kind": "codec_read",
       "api": "load_registry",
@@ -1103,7 +1103,7 @@
     },
     {
       "site": "loopx/control_plane/coordination/local_authority_shadow_adapter.py::<module>.local_authority_shadow_status::codec_read:load_registry#1",
-      "line": 301,
+      "line": 256,
       "column": 16,
       "kind": "codec_read",
       "api": "load_registry",
@@ -1111,20 +1111,12 @@
     },
     {
       "site": "loopx/control_plane/coordination/runtime_shadow.py::<module>.build_runtime_shadow_source_snapshot::codec_read:load_project_registry#1",
-      "line": 278,
+      "line": 307,
       "column": 20,
       "kind": "codec_read",
       "api": "load_project_registry",
       "classification": "codec_api"
     },
-    {
-      "site": "loopx/control_plane/coordination/runtime_shadow_writer_adapter.py::<module>.require_prose_state_write_allowed::codec_read:load_registry#1",
-      "line": 43,
-      "column": 35,
-      "kind": "codec_read",
-      "api": "load_registry",
-      "classification": "codec_api"
-    },
     {
       "site": "loopx/control_plane/coordination/shadow_goal_scope.py::<module>.shadow_goal_scope::codec_read:load_project_registry#1",
       "line": 60,
@@ -2101,6 +2093,14 @@
       "api": "load_registry",
       "classification": "codec_api"
     },
+    {
+      "site": "loopx/paths.py::<module>.effective_runtime_root::codec_read:load_registry#1",
+      "line": 346,
+      "column": 16,
+      "kind": "codec_read",
+      "api": "load_registry",
+      "classification": "codec_api"
+    },
     {
       "site": "loopx/project_map.py::<module>.read_only_project_map_run::codec_read:load_registry#1",
       "line": 477,
diff --git a/loopx/todos.py b/loopx/todos.py
index 2084d2a948..b5d709e18e 100644
--- a/loopx/todos.py
+++ b/loopx/todos.py
@@ -126,7 +126,7 @@
     enter_todo_ownership_handoff_gate,
     resolve_todo_completion_handoff,
 )
-from .control_plane.coordination.local_authority_shadow_adapter import effective_runtime_root
+from .paths import effective_runtime_root
 from .control_plane.work_items.task_lease import (
     enter_terminal_todo_lease_fence,
     hold_task_lease_mutation_fence,
diff --git a/tests/control_plane/test_chat_cold_source_import.py b/tests/control_plane/test_chat_cold_source_import.py
index ecb0683e88..173b109189 100644
--- a/tests/control_plane/test_chat_cold_source_import.py
+++ b/tests/control_plane/test_chat_cold_source_import.py
@@ -3,18 +3,27 @@
 
 import http.client
 import json
-import threading
+import subprocess
+import sys
+import time
 
 import pytest
 
-from loopx.chat_server import ChatHTTPServer, ChatRequestHandler
 from loopx.control_plane.effect_runtime import effect_runtime_result, restart_effect_runtime
 from test_cold_source_import_cli import workspace
 
 
+@pytest.fixture(params=[True, False], ids=["capture-present", "capture-absent"])
+def capture_present(request):
+    return request.param
+
+
 @pytest.fixture(params=[False, True], ids=["physical-registry", "external-registry-parent-alias"])
-def cold_api(tmp_path, monkeypatch, request):
-    _, state, _, body, runtime, _, _ = workspace(tmp_path, monkeypatch)
+def cold_api(tmp_path, monkeypatch, request, capture_present):
+    _, state, _, body, runtime, receiver, env = workspace(tmp_path, monkeypatch)
+    if not capture_present:
+        for name in ("runtime_shadow_writer_adapter.py", "local_authority_shadow_outbox.py"):
+            (receiver / "loopx/control_plane/coordination" / name).unlink()
     registry = tmp_path / "project/.loopx/registry.json"
     if request.param:
         # The configured registry may live outside conventional Goal folders.
@@ -24,16 +33,57 @@ def cold_api(tmp_path, monkeypatch, request):
         alias = tmp_path / "project-route"
         alias.symlink_to(tmp_path / "project", target_is_directory=True)
         registry = alias / "source-registry.json"
-    server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler)
-    server.runtime_root = runtime
-    server.registry_path = registry
-    server.runtime_root_override = str(runtime)
-    server.verbose = False
-    worker = threading.Thread(target=server.serve_forever, daemon=True)
-    worker.start()
+    ready = tmp_path / "server-ready.json"
+    script = """
+import json,sys,loopx
+from pathlib import Path
+from loopx.chat_server import ChatHTTPServer, ChatRequestHandler
+server = ChatHTTPServer(('127.0.0.1', 0), ChatRequestHandler)
+server.registry_path, server.runtime_root, ready = map(Path, sys.argv[1:])
+server.runtime_root_override = str(server.runtime_root)
+server.verbose = False
+pending = ready.with_suffix('.pending')
+pending.write_text(json.dumps({'port': server.server_port, 'package': loopx.__file__}))
+pending.replace(ready)
+server.serve_forever()
+"""
+    error_path = tmp_path / "server-errors.txt"
+    error_stream = error_path.open("w")
+    server = subprocess.Popen([sys.executable, "-u", "-c", script,
+        str(registry), str(runtime), str(ready)], cwd=tmp_path,
+        env={**env, "LOOPX_USAGE_PING": "0"}, stdout=subprocess.DEVNULL, stderr=error_stream)
+
+    def execute(script):
+        child = subprocess.run([sys.executable, "-c",
+            "import loopx; from pathlib import Path; "
+            f"assert Path(loopx.__file__) == Path({str(receiver / 'loopx/__init__.py')!r});\n" + script],
+            cwd=tmp_path, env={**env, "LOOPX_USAGE_PING": "0"},
+            capture_output=True, text=True, timeout=90)
+        assert child.returncode == 0, child.stdout + child.stderr
+        return json.loads(child.stdout)
+
+    def restart():
+        nonlocal server, port
+        server.terminate()
+        server.wait(timeout=10)
+        ready.unlink()
+        server = subprocess.Popen([sys.executable, "-u", "-c", script,
+            str(registry), str(runtime), str(ready)], cwd=tmp_path,
+            env={**env, "LOOPX_USAGE_PING": "0"}, stdout=subprocess.DEVNULL, stderr=error_stream)
+        port = wait_ready()
+
+    def wait_ready():
+        deadline = time.monotonic() + 30
+        while not ready.exists():
+            assert server.poll() is None, error_path.read_text()
+            assert time.monotonic() < deadline, "receiver HTTP server did not start"
+            time.sleep(.01)
+        startup = json.loads(ready.read_text())
+        assert startup["package"] == str(receiver / "loopx/__init__.py")
+        return startup["port"]
 
     def call(action="", payload=None, origin=None):
-        client = http.client.HTTPConnection("127.0.0.1", server.server_port, timeout=90)
+        client = http.client.HTTPConnection("127.0.0.1", port, timeout=90)
         path = "/api/chat/goal-storage" + (f"/import/{action}" if action else "?goal_id=cold")
         headers = {"Content-Type": "application/json"}
         if origin:
@@ -55,11 +105,15 @@ def read():
             "runtime_root": str(runtime), "goal_id": "cold", "role": None, "status": None,
             "todo_id": None, "agent_id": None, "limit": None})
 
-    yield call, read, state, body, runtime, registry
-    server.shutdown()
-    worker.join(5)
-    server.server_close()
-    restart_effect_runtime()
+    try:
+        port = wait_ready()
+        yield call, read, state, body, runtime, registry, execute, restart
+    finally:
+        if server.poll() is None:
+            server.terminate()
+        server.wait(timeout=10)
+        error_stream.close()
+        restart_effect_runtime()
 
 
 def prepare(call, provider="sqlite"):
@@ -74,7 +128,7 @@ def carrier(plan):
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 def test_cold_http_preview_reload_confirm_original_recovery(cold_api, provider):
-    call, read, state, body, runtime, registry = cold_api
+    call, read, state, body, runtime, registry, execute, restart = cold_api
     original = state.read_bytes()
     assert call()[1]["current"]["canonical"] is False
     plan = prepare(call, provider)
@@ -84,7 +138,7 @@ def test_cold_http_preview_reload_confirm_original_recovery(cold_api, provider):
     assert plan["complete_goal_backup_verified"] is False
     assert plan["authority_changed"] is False and not plan["legacy_writer_fenced"]
     saved = carrier(plan)
-    restart_effect_runtime()
+    restart()
     code, observed = call("recover", saved)
     assert code == 200 and observed["status"] == "prepared", observed
     assert observed["operation_id"] == plan["operation_id"]
@@ -100,12 +154,16 @@ def test_cold_http_preview_reload_confirm_original_recovery(cold_api, provider):
     rows = read()["todos"]
     assert next(row for row in rows if row["todo_id"] == "todo_current")["text"] == body
     assert next(row for row in rows if row["todo_id"] == "todo_archived")["evidence"] == "original"
-    from loopx.todos import add_goal_todo
-    add_goal_todo(registry_path=registry, goal_id="cold", role="agent", text="Keep later write",
-        claimed_by="agent-a", note="Original metadata")
+    result = execute(f"""
+import json
+from loopx.todos import add_goal_todo
+print(json.dumps(add_goal_todo(registry_path=Path({str(registry)!r}), goal_id='cold',
+    role='agent', text='Keep later write', claimed_by='agent-a', note='Original metadata')))
+""")
+    assert result["added"] is True
     later = read()
     state.unlink()
-    restart_effect_runtime()
+    restart()
     code, recovered = call("recover", saved)
     assert code == 200 and recovered["status"] == "replayed", recovered
     assert recovered["current"]["todo_count"] == 3
@@ -115,7 +173,7 @@ def test_cold_http_preview_reload_confirm_original_recovery(cold_api, provider):
 
 
 def test_cold_http_changed_source_backup_and_untrusted_inputs(cold_api):
-    call, read, state, _, runtime, _ = cold_api
+    call, read, state, _, runtime, _, _, _ = cold_api
     plan = prepare(call)
     saved = carrier(plan)
     assert call("apply", {**saved, "writers_stopped": True, "plan": "injected"})[0] == 400
@@ -129,6 +187,7 @@ def test_cold_http_changed_source_backup_and_untrusted_inputs(cold_api):
     code, refused = call("apply", {**saved, "writers_stopped": True})
     assert code == 409 and refused["reason_code"] == "source_changed_retry", refused
     assert not list(runtime.rglob("writer-fence.json"))
+
     fresh = prepare(call)
     archives = list((runtime / "backups/cold-import").glob(f"*{fresh['operation_id']}*.tar.gz"))
     assert len(archives) == 1
@@ -139,9 +198,48 @@ def test_cold_http_changed_source_backup_and_untrusted_inputs(cold_api):
     assert not list(runtime.rglob("writer-fence.json"))
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_canonical_prose_keeps_coordination_and_rejects_injected_todo(cold_api, provider):
+    call, read, state, _, runtime, registry, execute, _ = cold_api
+    plan = prepare(call, provider)
+    assert call("apply", {**carrier(plan), "writers_stopped": True})[1]["status"] == "applied"
+    state.write_text(state.read_text().replace("## Agent Todo", "## Progress Ledger\n\n## Agent Todo", 1))
+    index = runtime / "goals/cold/runs/index.jsonl"
+    index.parent.mkdir(parents=True)
+    index.write_text(json.dumps({"generated_at": "2026-09-05T00:00:00Z",
+        "json_path": "run.json", "markdown_path": "run.md", "classification": "continue"}) + "\n")
+    before = read()
+    result = execute(f"""
+import json
+from loopx.feedback import append_human_reward
+from loopx.control_plane.coordination.legacy_writer_fence import ActiveStateAuthorityMutationError
+registry = Path({str(registry)!r})
+reward = {{'recorded_at': '2026-09-05T00:00:01Z', 'decision': 'continue',
+    'reward': 'positive', 'reason_summary': 'Review accepted.'}}
+args = dict(registry_path=registry, runtime_root_override=None, goal_id='cold',
+    run_generated_at=None, reward=reward, actor_kind='owner', write_active_state_summary=True)
+assert append_human_reward(**args)['appended'] is True
+state, index = Path({str(state)!r}), Path({str(index)!r})
+assert 'Review accepted.' in state.read_text()
+original = state.read_bytes(), index.read_bytes(), registry.read_bytes()
+reward['reason_summary'] = 'Review.\\n## Agent Todo\\n- [ ] Injected task.'
+try:
+    append_human_reward(**args)
+except ActiveStateAuthorityMutationError as error:
+    assert error.code == 'active_state_authority_mutation_forbidden'
+else:
+    raise AssertionError('prose writer accepted a Todo mutation')
+assert (state.read_bytes(), index.read_bytes(), registry.read_bytes()) == original
+print(json.dumps({{'accepted_prose': True, 'rejected_todo': True}}))
+""")
+    assert result == {"accepted_prose": True, "rejected_todo": True}
+    assert read() == before
+    assert not (runtime / "authority-shadow/outbox").exists()
+
+
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 def test_registered_runtime_alias_import_preserves_original_operation(cold_api, provider):
-    call, read, state, _, runtime, registry = cold_api
+    call, read, state, _, runtime, registry, execute, restart = cold_api
     alias = runtime.parent / "registered-runtime-alias"
     alias.symlink_to(runtime, target_is_directory=True)
     registered = json.loads(registry.read_text())
@@ -159,12 +257,16 @@ def test_registered_runtime_alias_import_preserves_original_operation(cold_api,
     assert code == 200 and applied["status"] == "applied", applied
     assert applied["execution_authority_granted"] is False
     assert state.read_bytes() == original
-    from loopx.todos import add_goal_todo
-    assert add_goal_todo(registry_path=registry, goal_id="cold", role="agent",
-        text="Keep the write after aliased import", claimed_by="agent-a")["added"]
+    result = execute(f"""
+import json
+from loopx.todos import add_goal_todo
+print(json.dumps(add_goal_todo(registry_path=Path({str(registry)!r}), goal_id='cold',
+    role='agent', text='Keep the write after aliased import', claimed_by='agent-a')))
+""")
+    assert result["added"] is True
     later = read()
     state.unlink()
-    restart_effect_runtime()
+    restart()
     code, recovered = call("recover", saved)
     assert code == 200 and recovered["status"] == "replayed", recovered
     assert read() == later
@@ -172,7 +274,7 @@ def test_registered_runtime_alias_import_preserves_original_operation(cold_api,
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 def test_registered_runtime_alias_retarget_rejects_without_import(cold_api, provider):
-    call, _, state, _, runtime, registry = cold_api
+    call, _, state, _, runtime, registry, _, _ = cold_api
     alias = runtime.parent / "registered-runtime-alias"
     alias.symlink_to(runtime, target_is_directory=True)
     registered = json.loads(registry.read_text())
diff --git a/tests/control_plane/test_shadow_writer_boundaries.py b/tests/control_plane/test_shadow_writer_boundaries.py
index 2917ef84ca..2368b40a60 100644
--- a/tests/control_plane/test_shadow_writer_boundaries.py
+++ b/tests/control_plane/test_shadow_writer_boundaries.py
@@ -42,6 +42,28 @@ def fixture(tmp_path: Path) -> tuple[Path, Path, Path]:
     return registry, state, root
 
 
+@pytest.mark.parametrize("registry_directory", [".loopx", "standalone"])
+@pytest.mark.parametrize("override", [None, "", "override", Path("path-override")])
+def test_runtime_route_keeps_project_relative_precedence_without_writes(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, registry_directory: str, override,
+) -> None:
+    from loopx.paths import effective_runtime_root
+    from loopx.control_plane.coordination.local_authority_shadow_adapter import effective_runtime_root as compatibility_route
+
+    assert compatibility_route is effective_runtime_root
+    project = tmp_path / "project"
+    registry = project / registry_directory / "registry.json"
+    registry.parent.mkdir(parents=True)
+    original = json.dumps({"common_runtime_root": "declared"}).encode()
+    registry.write_bytes(original)
+    monkeypatch.chdir(tmp_path)
+    owner = project if registry_directory == ".loopx" else registry.parent
+    assert effective_runtime_root(registry, override) == owner / str(override or "declared")
+    assert registry.read_bytes() == original
+    assert list(registry.parent.iterdir()) == [registry]
+    assert not (owner / "declared").exists()
+
+
 def test_handoff_writer_refuses_a_fence_before_primary(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -466,7 +488,7 @@ def project_at_distinct_time(text):
         return project_at_distinct_time
 
     monkeypatch.setattr(
-        "loopx.control_plane.coordination.local_authority_shadow_adapter.todo_partition_projector",
+        "loopx.control_plane.coordination.runtime_shadow.todo_partition_projector",
         ticking_projector,
     )
     goal = json.loads(registry.read_text(encoding="utf-8"))["goals"][0]
@@ -499,7 +521,7 @@ def test_prose_guard_preserves_json_scalar_identity(
     original_value: bool, planned_value: int,
 ) -> None:
     """A full source projection must retain JSON types even if Python equates them."""
-    from loopx.control_plane.coordination import local_authority_shadow_adapter as adapter
+    from loopx.control_plane.coordination import runtime_shadow as adapter
     from loopx.control_plane.coordination.local_authority_shadow_projection import (
         todo_partition_projection,
     )
@@ -552,6 +574,81 @@ def test_prose_only_reward_holds_before_index_append_during_maintenance(tmp_path
     assert (state.read_bytes(), index.read_bytes()) == before
 
 
+def test_prose_guard_rejects_a_handoff_change(tmp_path: Path) -> None:
+    from loopx.control_plane.coordination.runtime_shadow_writer_adapter import (
+        ActiveStateAuthorityMutationError, require_prose_state_write_allowed,
+    )
+    registry, state, root = fixture(tmp_path)
+    original = state.read_text()
+    with pytest.raises(ActiveStateAuthorityMutationError, match="would change canonical"):
+        require_prose_state_write_allowed(
+            registry_path=registry, runtime_root=root, goal_id=GOAL, state_path=state,
+            original_text=original, planned_text=original.replace("handoff_mode: legacy", "handoff_mode: hard_lease"),
+        )
+    assert state.read_text() == original
+
+
+@pytest.mark.parametrize("maintenance", [False, True])
+def test_prose_projection_failure_preserves_source_and_reward_index(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, maintenance: bool,
+) -> None:
+    from loopx.feedback import append_human_reward
+    from loopx.control_plane.coordination import runtime_shadow as source
+    from loopx.control_plane.coordination.runtime_shadow_writer_adapter import ActiveStateAuthorityMutationError
+    from loopx.control_plane.coordination.shadow_management import ShadowManagementError, shadow_management_state_path
+
+    registry, state, index, reward = reward_fixture(tmp_path)
+    called = []
+
+    def unavailable(*_args, **_kwargs):
+        called.append(True)
+        raise ValueError("source projection unavailable")
+
+    monkeypatch.setattr(source, "todo_partition_projector", unavailable)
+    if maintenance:
+        path = shadow_management_state_path(tmp_path / "runtime", GOAL)
+        path.parent.mkdir(parents=True)
+        path.write_text("{}")
+    before = state.read_bytes(), index.read_bytes(), registry.read_bytes()
+    error = ShadowManagementError if maintenance else ActiveStateAuthorityMutationError
+    with pytest.raises(error):
+        append_human_reward(registry_path=registry, runtime_root_override=None,
+            goal_id=GOAL, run_generated_at=None, reward=reward, actor_kind="owner",
+            write_active_state_summary=True)
+    assert called == ([] if maintenance else [True])
+    assert (state.read_bytes(), index.read_bytes(), registry.read_bytes()) == before
+
+
+def test_configuration_and_prose_guard_load_without_capture_modules(tmp_path, monkeypatch):
+    from test_cold_source_import_cli import workspace
+
+    _, state, _, _, runtime, receiver, env = workspace(tmp_path, monkeypatch)
+    for name in ("runtime_shadow_writer_adapter.py", "local_authority_shadow_outbox.py"):
+        (receiver / "loopx/control_plane/coordination" / name).unlink()
+    registry = tmp_path / "project/.loopx/registry.json"
+    child = subprocess.run([sys.executable, "-c", """
+import sys,loopx
+from pathlib import Path
+import loopx.presentation.configuration_api
+from loopx.feedback import require_prose_state_write_allowed
+from loopx.control_plane.coordination.legacy_writer_fence import ActiveStateAuthorityMutationError
+registry,runtime,state,package = map(Path,sys.argv[1:])
+assert Path(loopx.__file__) == package / 'loopx/__init__.py'
+original = state.read_text()
+args = dict(registry_path=registry,runtime_root=runtime,goal_id='cold',state_path=state,original_text=original)
+require_prose_state_write_allowed(**args,planned_text=original+'\\nA narrative observation.\\n')
+try:
+    require_prose_state_write_allowed(**args,planned_text=original.replace('handoff_mode: legacy','handoff_mode: hard_lease'))
+except ActiveStateAuthorityMutationError:
+    pass
+else:
+    raise AssertionError('handoff mutation was accepted')
+assert state.read_text() == original
+""", str(registry), str(runtime), str(state), str(receiver)], cwd=tmp_path,
+        env=env, capture_output=True, text=True, timeout=60)
+    assert child.returncode == 0, child.stdout + child.stderr
+
+
 def test_override_root_is_the_only_maintenance_authority(tmp_path: Path) -> None:
     from loopx.control_plane.coordination.shadow_management import ShadowManagementError, shadow_management_state_path
     registry, state, root = fixture(tmp_path)