refactor(coordination): decouple CLI and cold import from shadow producers

docs/architecture/rfcs/typescript-control-plane-migration-v0.md

@@ -112,6 +112,20 @@ caller-isolation exit, not deletion of the live legacy writer or completion of
 receiver adoption, Host execution, PostgreSQL or D2 qualification.
 [Contract and retained duties](../../reference/canonical-todo-completion-update.md#team-plan-source-writer-isolation).
 
+Public CLI isolation now keeps shadow configuration in `coordination/configuration.py`
+and complete source capture in `coordination/authority_source_capture.py`. Cold import,
+source inspection and historical promotion recovery reuse those retained owners;
+only explicit enabled shadow actions load the old producers. Ordinary new-Goal
+SQLite creation, leased work, once-only settlement and reversible exit, plus cold
+File/SQLite import and original-operation replay, run through the real CLI with
+all four shadow producers physically absent in disposable source/normal-wheel
+copies. The old module still re-exports the retained APIs for current consumers;
+enabled capture, outbox and legacy commands remain supported. Preserve exact
+source identity, archive dependencies, safe lease reads and later-write recovery.
+This removes an eager-import blocker; deleting live producers still requires
+their last-caller exit. App/Lark contracts, provider transactions, cohort adoption
+and frozen D2 evidence remain separate.
+
 Existing Goal migration, two-policy ownership retirement and storage-format
 upgrade retain separate receipts and exits. Original-receipt recovery does not
 justify retaining `legacy` as a live policy. Required migration readers remain.

docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md

@@ -91,6 +91,16 @@ canonical team-plan 只在受支持的未迁移分支加载 capture 和 Markdown
 不等于删除活跃旧 writer，或完成 receiver adoption、Host 执行、PostgreSQL／D2
 资格。见[合同及保留义务](../../reference/canonical-todo-completion-update.md#team-plan-source-writer-isolation)。
 
+公开 CLI 的 shadow 配置归入 `coordination/configuration.py`，完整源捕获归入
+`coordination/authority_source_capture.py`；冷导入、源检查和历史 promotion 恢复
+复用保留 owner，只有显式开启的 shadow 动作加载旧 producer。隔离源码及正常
+wheel 的副本中物理删除四个 shadow producer 后，真实 CLI 仍覆盖新 Goal 默认
+SQLite、租约工作、一次结算及可回退出口，以及冷 File／SQLite 导入和原操作重放。
+旧模块为现有消费者保留 API 重导出；启用的 capture、outbox 和旧命令仍受支持。
+精确源身份、归档依赖、安全 lease 读取和后续写入恢复均须保留。本批移除提前导入
+阻塞；删除活跃 producer 仍须其最后调用方退出。App／Lark 合同、provider 事务、
+cohort 采用及冻结 D2 证据保持独立。
+
 存量 Goal 迁移、两策略退役和格式升级仍各有独立回执及出口；原回执恢复不能成为
 保留 legacy 活跃策略的理由，必要迁移 reader 保留。
 

loopx/capabilities/machine_configuration/goal_storage.py

@@ -75,7 +75,7 @@ def initialize_goal_storage_target(runtime_root: Path, goal: Mapping[str, Any],
         # owner can request a complete source capture for unfinished creation.
         if result.get("source_capture_required") is not True:
             return result
-        from ...control_plane.coordination.runtime_shadow import build_runtime_shadow_source_snapshot
+        from ...control_plane.coordination.authority_source_capture import build_runtime_shadow_source_snapshot
         projection, snapshot = build_runtime_shadow_source_snapshot(goal=goal, runtime_root=runtime_root,
             state_path=state_path, registry_path=registry_path)
         request.update(creation_operation_id=goal.get("creation_operation_id"), projection=projection, source_snapshot=snapshot)

loopx/cli_commands/authority_shadow.py

@@ -4,17 +4,12 @@
 from collections.abc import Callable
 from pathlib import Path
 
-from ..control_plane.coordination.local_authority_shadow_adapter import (
-    CLI_DRAIN_LOCK_TIMEOUT_SECONDS,
-    drain_local_authority_shadow_outbox,
-    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
 
 
 AUTHORITY_SHADOW_CLI_SCHEMA = "loopx_authority_shadow_cli_v0"
+CLI_DRAIN_LOCK_TIMEOUT_SECONDS = 5.0
 CLI_DRAIN_MAX_ENTRIES = 256
 CLI_DRAIN_BUDGET_SECONDS = 30.0
 
@@ -152,43 +147,50 @@ def handle_authority_shadow_command(
     action = str(getattr(args, "authority_shadow_command", None))
     payload: dict[str, object]
     try:
-        # The same resolver every writer hook uses, so drain and status address
-        # the lineage those hooks wrote.
-        runtime_root = effective_runtime_root(registry_path, runtime_root_arg)
-        if action == "drain":
-            if args.max_entries < 1:
-                raise ValueError("--max-entries must be at least 1")
-            result = drain_local_authority_shadow_outbox(
-                registry_path=registry_path,
-                runtime_root=runtime_root,
-                goal_id=args.goal_id,
-                max_entries=args.max_entries,
-                budget_seconds=args.budget_seconds,
-                lock_timeout_seconds=args.lock_timeout_seconds,
-            )
-            payload = {
-                "schema_version": AUTHORITY_SHADOW_CLI_SCHEMA,
-                "action": "drain",
-                **result.to_payload(),
-            }
-        else:
-            payload = {
-                "schema_version": AUTHORITY_SHADOW_CLI_SCHEMA,
-                **local_authority_shadow_status(
+        from ..control_plane.coordination.local_authority_shadow_adapter import (
+            drain_local_authority_shadow_outbox,
+            local_authority_shadow_status,
+        )
+        from ..control_plane.coordination.local_authority_shadow_outbox import OutboxError
+
+        try:
+            # The same resolver every writer hook uses, so drain and status address
+            # the lineage those hooks wrote.
+            runtime_root = effective_runtime_root(registry_path, runtime_root_arg)
+            if action == "drain":
+                if args.max_entries < 1:
+                    raise ValueError("--max-entries must be at least 1")
+                result = drain_local_authority_shadow_outbox(
                     registry_path=registry_path,
                     runtime_root=runtime_root,
                     goal_id=args.goal_id,
-                ),
+                    max_entries=args.max_entries,
+                    budget_seconds=args.budget_seconds,
+                    lock_timeout_seconds=args.lock_timeout_seconds,
+                )
+                payload = {
+                    "schema_version": AUTHORITY_SHADOW_CLI_SCHEMA,
+                    "action": "drain",
+                    **result.to_payload(),
+                }
+            else:
+                payload = {
+                    "schema_version": AUTHORITY_SHADOW_CLI_SCHEMA,
+                    **local_authority_shadow_status(
+                        registry_path=registry_path,
+                        runtime_root=runtime_root,
+                        goal_id=args.goal_id,
+                    ),
+                }
+        except OutboxError as exc:
+            payload = {
+                "ok": False,
+                "schema_version": AUTHORITY_SHADOW_CLI_SCHEMA,
+                "action": action,
+                "goal_id": args.goal_id,
+                "error": str(exc),
+                "error_code": exc.reason_code,
             }
-    except OutboxError as exc:
-        payload = {
-            "ok": False,
-            "schema_version": AUTHORITY_SHADOW_CLI_SCHEMA,
-            "action": action,
-            "goal_id": args.goal_id,
-            "error": str(exc),
-            "error_code": exc.reason_code,
-        }
     except LockAcquireTimeoutError as exc:
         payload = {
             "ok": False,

loopx/cli_commands/coordination_shadow.py

@@ -8,20 +8,8 @@
 
 from ..agent_registry import registered_agent_ids_for_goal
 
-# The projection builder is reached through this module by
-# tests that seed and read the shadow through the command surface; keep them
-# importable here even when the command does not call them directly.
-from ..control_plane.coordination.runtime_shadow import (  # noqa: F401
-    bootstrap_coordination_runtime_shadow,
-    build_runtime_shadow_source_snapshot,
-    build_todo_runtime_shadow_projection,
-    inspect_coordination_runtime_shadow,
-    qualify_coordination_runtime_shadow,
-    read_coordination_runtime_shadow_todo_candidate,
-    review_local_coordination_authority_promotion,
-    resolve_coordination_runtime_shadow_config,
-    rollback_coordination_runtime_shadow,
-)
+from ..control_plane.coordination.authority_source_capture import build_runtime_shadow_source_snapshot
+from ..control_plane.coordination.configuration import resolve_coordination_runtime_shadow_config
 from ..control_plane.coordination.promotion_review import execute_reviewed_coordination_promotion
 from ..control_plane.coordination.shadow_goal_scope import shadow_goal_scope
 from ..control_plane.projects.registry_codec import load_project_registry
@@ -356,6 +344,9 @@ def handle_coordination_shadow_command(
             }
             print_payload(payload, output_format(args), _render)
             return 1
+        # Only explicit, enabled legacy shadow actions load the producer.
+        from ..control_plane.coordination import runtime_shadow as shadow
+
         _, _, state_path = resolve_goal_state(registry=registry, goal_id=args.goal_id,
             project_override=args.project, state_file_override=args.state_file)
         if args.coordination_shadow_command == "rollback":
@@ -366,7 +357,7 @@ def handle_coordination_shadow_command(
         projection_version = _projection_version(projection)
         projected_todos = projection.get("todos")
         projected_leases = projection.get("leases")
-        inspection = {"status": "not_evaluated"} if args.coordination_shadow_command == "rollback" else inspect_coordination_runtime_shadow(
+        inspection = {"status": "not_evaluated"} if args.coordination_shadow_command == "rollback" else shadow.inspect_coordination_runtime_shadow(
             goal=goal,
             runtime_root=runtime_root,
             goal_id=args.goal_id,
@@ -408,7 +399,7 @@ def handle_coordination_shadow_command(
                 operation_id = f"shadow-bootstrap:{args.goal_id}:{operation_digest}"
             with shadow_goal_scope(registry_path, goal_id=args.goal_id) as scope:
                 goal = scope.goal
-                bootstrap = bootstrap_coordination_runtime_shadow(
+                bootstrap = shadow.bootstrap_coordination_runtime_shadow(
                     goal=goal,
                     runtime_root=runtime_root,
                     goal_id=args.goal_id,
@@ -421,7 +412,7 @@ def handle_coordination_shadow_command(
             payload["executed"] = True
             payload["bootstrap"] = bootstrap
             if bootstrap.get("status") in {"applied", "replayed", "recovered"}:
-                payload["inspection"] = inspect_coordination_runtime_shadow(
+                payload["inspection"] = shadow.inspect_coordination_runtime_shadow(
                     goal=goal,
                     runtime_root=runtime_root,
                     goal_id=args.goal_id,
@@ -435,7 +426,7 @@ def handle_coordination_shadow_command(
                 and final_inspection.get("status") == "matched"
             )
         if args.coordination_shadow_command == "qualify":
-            qualification = qualify_coordination_runtime_shadow(
+            qualification = shadow.qualify_coordination_runtime_shadow(
                 goal=goal,
                 runtime_root=runtime_root,
                 goal_id=args.goal_id,
@@ -447,7 +438,7 @@ def handle_coordination_shadow_command(
             payload["qualification"] = qualification
             payload["ok"] = qualification.get("status") == "qualified"
         if args.coordination_shadow_command == "read-candidate":
-            read_candidate = read_coordination_runtime_shadow_todo_candidate(
+            read_candidate = shadow.read_coordination_runtime_shadow_todo_candidate(
                 goal=goal,
                 runtime_root=runtime_root,
                 goal_id=args.goal_id,
@@ -484,7 +475,7 @@ def handle_coordination_shadow_command(
                         "registered_agents": registered_agents,
                     }
                 )
-                promotion = review_local_coordination_authority_promotion(
+                promotion = shadow.review_local_coordination_authority_promotion(
                     goal=goal,
                     runtime_root=runtime_root,
                     goal_id=args.goal_id,
@@ -508,7 +499,7 @@ def handle_coordination_shadow_command(
             payload["expected_provider_revision"] = provider_revision
             payload["expected_bootstrap_operation_id"] = pending_bootstrap
             if args.execute:
-                rollback = rollback_coordination_runtime_shadow(
+                rollback = shadow.rollback_coordination_runtime_shadow(
                     goal=goal, runtime_root=runtime_root, goal_id=args.goal_id,
                     operation_id=f"shadow-rollback:{args.goal_id}:{provider_revision or pending_bootstrap}",
                     expected_provider_revision=provider_revision,

loopx/configure_goal.py

@@ -49,7 +49,7 @@
 )
 from .control_plane.agents.supervisor import normalize_peer_supervisor
 from .control_plane.agents.work_mode import normalize_agent_work_modes
-from .control_plane.coordination import runtime_shadow as shadow
+from .control_plane.coordination import configuration as shadow
 from .control_plane.coordination.configuration import normalize_goal_write_scope
 from .control_plane.operator_inbox_binding import local_private_config_digest
 from .control_plane.projects.registry_codec import (