refactor(shadow): retain observations after source-producer exit

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

@@ -145,8 +145,16 @@ producer modules using the qualified CLI startup isolation above.
 
 Shadow diagnostics now have an independent read-only boundary: retained outbox
 disk codecs live in `local_authority_shadow_history.py`, and status/candidate
-read transport in `local_authority_shadow_readback.py`. The public status CLI
-does not load source producers; drain still uses its existing orchestration.
+read transport in `local_authority_shadow_readback.py`. The public status and enabled
+`coordination-shadow inspect`/`qualify`/`read-candidate` commands call that owner
+without loading source producers. Full source and normal-wheel observations
+retain matched/refused, default-off, source-drift and missing/corrupt-store
+recovery behavior with all four producer modules absent. Historical imports
+remain exact aliases; bootstrap, rollback and promotion retain their live adapter.
+The existing TS qualifier now propagates provider read refusals before lineage
+validation can mislabel corruption as `bootstrap_required`; valid/missing stores
+retain their behavior. This is an explicit failure-diagnostic correction, with
+no new RPC or migration authority; drain still uses its existing orchestration.
 Original entry identity, malformed evidence, cursor residue and missing-store
 behavior remain unchanged. Python retains historical filesystem decoding and
 presentation IO; TS retains history proof, delivery and cleanup authority.

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

@@ -116,7 +116,14 @@ producer module。
 
 shadow 诊断现在有独立只读边界：历史 outbox 磁盘 codec 位于
 `local_authority_shadow_history.py`，status／candidate read transport 位于
-`local_authority_shadow_readback.py`。公开 status CLI 不加载源 producer；drain
+`local_authority_shadow_readback.py`。公开 status 和启用的
+`coordination-shadow inspect`／`qualify`／`read-candidate` 直接调用保留 owner，
+不加载源 producer。源码和正常 wheel 物理删除四个 producer 后，完整读回保持
+匹配／拒绝、默认关闭、源漂移、store 缺失／损坏及恢复结果；历史 import 保留
+同一函数 alias，bootstrap／rollback／promotion 的活跃 adapter 继续受支持。
+既有 TS qualifier 保留 provider 读取失败，避免把损坏 journal 误称尚未
+bootstrap；正常／缺失状态保留原结果。这是明确的失败诊断修复，无新 RPC 或
+迁移权限；drain
 仍走现有 orchestration。原 entry 身份、损坏证据、cursor 残留和缺失 store 行为
 保持。Python 保留历史文件解码及展示 IO；TS 保留历史证明、投递与清理权限。
 旧 producer 的已有导入为支持调用方保留 alias，不复制实现。源码及正常 wheel

docs/reference/authority-observation-retirement.md

@@ -71,6 +71,22 @@ counts are diagnostic hints: they do not authorize cleanup or prove delivery.
 A missing candidate stays missing; status never creates a replacement identity,
 repairs bytes, drains an outbox or grants a writer permission.
 
+The enabled `coordination-shadow inspect`, `qualify` and `read-candidate`
+commands likewise call `local_authority_shadow_readback.py` directly. Their
+complete source snapshot, default-off gate, bounded qualification and refusal
+results retain the same TS owners. Source and normal-wheel checks remove all
+four old producers and compare byte-stable observations, insufficient coverage,
+missing Todo, source drift, unreadable/missing store and subsequent recovery.
+The historical `runtime_shadow.py` imports remain aliases to the same functions.
+The shared TS qualifier now preserves `provider_protocol_violation` and
+`provider_read_unavailable` rather than mislabelling a failed provider read as
+`bootstrap_required`. Consequently a corrupt journal makes inspection fail,
+instead of reporting a missing baseline; qualification and candidate reads
+retain the provider refusal too. Missing and valid stores retain their existing
+behavior. Recovery restores the original bytes, without creating a new lineage.
+Bootstrap, rollback and promotion still load their supported management adapter;
+this reader exit grants no writer cutoff, import or promotion authority.
+
 The retained Python filesystem codecs preserve the existing wire and errors;
 candidate proof and drain effects remain in their TypeScript owners. Supported
 unpromoted writes and `authority-shadow drain` still need their current producer

loopx/cli_commands/coordination_shadow.py

@@ -10,6 +10,7 @@
 
 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 import local_authority_shadow_readback as observation
 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
@@ -344,8 +345,6 @@ 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)
@@ -357,7 +356,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 shadow.inspect_coordination_runtime_shadow(
+        inspection = {"status": "not_evaluated"} if args.coordination_shadow_command == "rollback" else observation.inspect_coordination_runtime_shadow(
             goal=goal,
             runtime_root=runtime_root,
             goal_id=args.goal_id,
@@ -388,6 +387,8 @@ def handle_coordination_shadow_command(
             "decision_read_from_shadow": False,
         }
         if args.coordination_shadow_command == "bootstrap" and args.execute:
+            from ..control_plane.coordination import runtime_shadow as shadow
+
             from ..control_plane.coordination.shadow_management import read_shadow_management_state
             management = read_shadow_management_state(runtime_root, args.goal_id)
             if management is not None and management["status"] in {"bootstrapping", "active"}:
@@ -412,7 +413,7 @@ def handle_coordination_shadow_command(
             payload["executed"] = True
             payload["bootstrap"] = bootstrap
             if bootstrap.get("status") in {"applied", "replayed", "recovered"}:
-                payload["inspection"] = shadow.inspect_coordination_runtime_shadow(
+                payload["inspection"] = observation.inspect_coordination_runtime_shadow(
                     goal=goal,
                     runtime_root=runtime_root,
                     goal_id=args.goal_id,
@@ -426,7 +427,7 @@ def handle_coordination_shadow_command(
                 and final_inspection.get("status") == "matched"
             )
         if args.coordination_shadow_command == "qualify":
-            qualification = shadow.qualify_coordination_runtime_shadow(
+            qualification = observation.qualify_coordination_runtime_shadow(
                 goal=goal,
                 runtime_root=runtime_root,
                 goal_id=args.goal_id,
@@ -438,7 +439,7 @@ def handle_coordination_shadow_command(
             payload["qualification"] = qualification
             payload["ok"] = qualification.get("status") == "qualified"
         if args.coordination_shadow_command == "read-candidate":
-            read_candidate = shadow.read_coordination_runtime_shadow_todo_candidate(
+            read_candidate = observation.read_coordination_runtime_shadow_todo_candidate(
                 goal=goal,
                 runtime_root=runtime_root,
                 goal_id=args.goal_id,
@@ -453,6 +454,8 @@ def handle_coordination_shadow_command(
                 and read_candidate.get("decision_read_from_shadow") is False
             )
         if args.coordination_shadow_command == "promote":
+            from ..control_plane.coordination import runtime_shadow as shadow
+
             # A saved plan owns qualification policy, so the flag defaults to
             # `None`; normalize it only for the freshly reviewed path.
             minimum_operations = args.minimum_operations if args.minimum_operations is not None else 3
@@ -499,6 +502,8 @@ def handle_coordination_shadow_command(
             payload["expected_provider_revision"] = provider_revision
             payload["expected_bootstrap_operation_id"] = pending_bootstrap
             if args.execute:
+                from ..control_plane.coordination import runtime_shadow as 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}",

loopx/control_plane/coordination/local_authority_shadow_readback.py

@@ -5,6 +5,7 @@
 """
 from __future__ import annotations
 
+from collections.abc import Callable, Mapping
 from dataclasses import asdict
 from pathlib import Path
 from typing import Any
@@ -15,10 +16,18 @@
 from ..effect_runtime import effect_runtime_result
 from . import local_authority_shadow_history as outbox
 from .coordination_state_contract_generated import (
+    COORDINATION_RUNTIME_SHADOW_INSPECT_REQUEST_SCHEMA as RUNTIME_SHADOW_INSPECT_REQUEST_SCHEMA_VERSION,
+    COORDINATION_RUNTIME_SHADOW_INSPECT_RESULT_SCHEMA,
+    COORDINATION_RUNTIME_SHADOW_QUALIFY_REQUEST_SCHEMA as RUNTIME_SHADOW_QUALIFY_REQUEST_SCHEMA_VERSION,
+    COORDINATION_RUNTIME_SHADOW_QUALIFY_RESULT_SCHEMA,
+    COORDINATION_RUNTIME_SHADOW_TODO_READ_REQUEST_SCHEMA as RUNTIME_SHADOW_TODO_READ_REQUEST_SCHEMA_VERSION,
+    COORDINATION_RUNTIME_SHADOW_TODO_READ_RESULT_SCHEMA,
     LOCAL_AUTHORITY_SHADOW_READ_REQUEST_SCHEMA,
     LOCAL_AUTHORITY_SHADOW_READ_RESULT_SCHEMA,
 )
-from .local_authority_shadow_projection import head_digest
+from .local_authority_shadow_projection import (
+    head_digest, source_effect_runtime_result as source_effect_runtime_result,
+)
 from .configuration import resolve_coordination_runtime_shadow_config, local_authority_shadow_summary
 from .shadow_management import read_shadow_capture_binding
 
@@ -180,3 +189,166 @@ def local_authority_shadow_status(
         if store_bytes is not None
         else None,
     }
+
+
+RuntimeInvoker = Callable[..., object]
+RUNTIME_SHADOW_INSPECT_METHOD = "coordination.runtime_shadow.inspect"
+RUNTIME_SHADOW_QUALIFY_METHOD = "coordination.runtime_shadow.qualify"
+RUNTIME_SHADOW_TODO_READ_METHOD = "coordination.runtime_shadow.todo_read_candidate"
+
+
+def inspect_coordination_runtime_shadow(
+    *,
+    goal: Mapping[str, Any] | None,
+    runtime_root: Path,
+    goal_id: str,
+    projection: Mapping[str, Any],
+    source_snapshot: Mapping[str, Any] | None = None,
+    runtime_invoker: RuntimeInvoker = source_effect_runtime_result,
+) -> dict[str, object]:
+    """Read parity evidence without allowing the shadow to drive decisions."""
+
+    config = resolve_coordination_runtime_shadow_config(goal)
+    if not config.enabled:
+        return {
+            "schema_version": COORDINATION_RUNTIME_SHADOW_INSPECT_RESULT_SCHEMA,
+            "status": "disabled",
+            "reason_code": config.reason_code,
+            "parity_matches": False,
+            "bootstrap_required": False,
+            "decision_read_from_shadow": False,
+        }
+    request = {
+        "schema_version": RUNTIME_SHADOW_INSPECT_REQUEST_SCHEMA_VERSION,
+        "runtime_root": str(runtime_root.expanduser().absolute()),
+        "goal_id": goal_id,
+        "source_snapshot": dict(source_snapshot or {}),
+        "projection": dict(projection),
+    }
+    try:
+        result = runtime_invoker(RUNTIME_SHADOW_INSPECT_METHOD, request)
+    except Exception as exc:
+        return {
+            "schema_version": COORDINATION_RUNTIME_SHADOW_INSPECT_RESULT_SCHEMA,
+            "status": "failed",
+            "reason_code": "shadow_runtime_unavailable",
+            "reason": str(exc),
+            "parity_matches": False,
+            "bootstrap_required": False,
+            "decision_read_from_shadow": False,
+        }
+    if not isinstance(result, Mapping):
+        return {
+            "schema_version": COORDINATION_RUNTIME_SHADOW_INSPECT_RESULT_SCHEMA,
+            "status": "failed",
+            "reason_code": "shadow_runtime_result_invalid",
+            "parity_matches": False,
+            "bootstrap_required": False,
+            "decision_read_from_shadow": False,
+        }
+    return dict(result)
+
+def qualify_coordination_runtime_shadow(
+    *,
+    goal: Mapping[str, Any] | None,
+    runtime_root: Path,
+    goal_id: str,
+    projection: Mapping[str, Any],
+    minimum_operations: int,
+    required_event_kinds: list[str],
+    source_snapshot: Mapping[str, Any] | None = None,
+    runtime_invoker: RuntimeInvoker = source_effect_runtime_result,
+) -> dict[str, object]:
+    """Qualify coverage across a shadow lineage without serving from it."""
+
+    config = resolve_coordination_runtime_shadow_config(goal)
+    if not config.enabled:
+        return {
+            "schema_version": COORDINATION_RUNTIME_SHADOW_QUALIFY_RESULT_SCHEMA,
+            "status": "disabled",
+            "reason_code": config.reason_code,
+            "qualified": False,
+            "primary_writeback_preserved": True,
+            "decision_read_from_shadow": False,
+        }
+    request = {
+        "schema_version": RUNTIME_SHADOW_QUALIFY_REQUEST_SCHEMA_VERSION,
+        "runtime_root": str(runtime_root.expanduser().absolute()),
+        "goal_id": goal_id,
+        "source_snapshot": dict(source_snapshot or {}),
+        "projection": dict(projection),
+        "minimum_operations": minimum_operations,
+        "required_event_kinds": list(required_event_kinds),
+    }
+    try:
+        result = runtime_invoker(RUNTIME_SHADOW_QUALIFY_METHOD, request)
+    except Exception as exc:
+        return {
+            "schema_version": COORDINATION_RUNTIME_SHADOW_QUALIFY_RESULT_SCHEMA,
+            "status": "failed",
+            "reason_code": "shadow_qualification_runtime_unavailable",
+            "reason": str(exc),
+            "qualified": False,
+            "primary_writeback_preserved": True,
+            "decision_read_from_shadow": False,
+        }
+    if not isinstance(result, Mapping):
+        return {
+            "schema_version": COORDINATION_RUNTIME_SHADOW_QUALIFY_RESULT_SCHEMA,
+            "status": "failed",
+            "reason_code": "shadow_qualification_runtime_result_invalid",
+            "qualified": False,
+            "primary_writeback_preserved": True,
+            "decision_read_from_shadow": False,
+        }
+    return dict(result)
+
+def read_coordination_runtime_shadow_todo_candidate(
+    *,
+    goal: Mapping[str, Any] | None,
+    runtime_root: Path,
+    goal_id: str,
+    todo_id: str,
+    projection: Mapping[str, Any],
+    source_snapshot: Mapping[str, Any] | None = None,
+    runtime_invoker: RuntimeInvoker = source_effect_runtime_result,
+) -> dict[str, object]:
+    """Read one parity-matched file Todo as pre-promotion evidence only."""
+
+    config = resolve_coordination_runtime_shadow_config(goal)
+    if not config.enabled:
+        return {
+            "schema_version": COORDINATION_RUNTIME_SHADOW_TODO_READ_RESULT_SCHEMA,
+            "status": "disabled",
+            "reason_code": config.reason_code,
+            "read_candidate_qualified": False,
+            "decision_read_from_shadow": False,
+        }
+    request = {
+        "schema_version": RUNTIME_SHADOW_TODO_READ_REQUEST_SCHEMA_VERSION,
+        "runtime_root": str(runtime_root.expanduser().absolute()),
+        "goal_id": goal_id,
+        "source_snapshot": dict(source_snapshot or {}),
+        "todo_id": todo_id,
+        "projection": dict(projection),
+    }
+    try:
+        result = runtime_invoker(RUNTIME_SHADOW_TODO_READ_METHOD, request)
+    except Exception as exc:
+        return {
+            "schema_version": COORDINATION_RUNTIME_SHADOW_TODO_READ_RESULT_SCHEMA,
+            "status": "failed",
+            "reason_code": "shadow_todo_read_runtime_unavailable",
+            "reason": str(exc),
+            "read_candidate_qualified": False,
+            "decision_read_from_shadow": False,
+        }
+    if not isinstance(result, Mapping):
+        return {
+            "schema_version": COORDINATION_RUNTIME_SHADOW_TODO_READ_RESULT_SCHEMA,
+            "status": "failed",
+            "reason_code": "shadow_todo_read_runtime_result_invalid",
+            "read_candidate_qualified": False,
+            "decision_read_from_shadow": False,
+        }
+    return dict(result)