fix(explore): keep research execution independent of the historic Todo facade

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

@@ -64,7 +64,7 @@ without claiming code retirement:
 bootstrap, handoff, team-plan and Monitor still have supported source callers;
 keep capture/outbox and historical recovery until their own exits. [Caller and
 rollback inventory](../../reference/canonical-todo-completion-update.md#canonical-mutation-entrypoint-isolation).
-Production Todo consumers now bypass the historic `loopx.todos` re-export:
+Production Todo consumers use the concrete owners instead of the historic `loopx.todos` re-export:
 CLI registration/actions, quota and planning, capability writeback, source lease
 reads and the Lark source projection import the same concrete mutation, full-read,
 path or Markdown display owners. No new wrapper, decision or RPC is introduced.
@@ -74,8 +74,18 @@ writer and four shadow producers together; real File/SQLite CLI lifecycle,
 once-only settlement, Monitor replay and App cold-import/action/restart journeys
 must still work, with missing-provider/lease and changed-source/backup refusals.
 Lark dry-run and Explore replay qualify local consumers only, not live external
-adoption. This closes the production re-export dependency, while last supported
-source writers, historical recovery, installed cohort and release exits remain.
+adoption. The fresh research execution path added by [#5280](https://github.com/loopx-project/loopx/pull/5280)
+reintroduced a function-local historic import after the earlier caller inventory.
+It now imports `todos/list_readback.py` directly, preserving the same typed
+current-Todo validation and the historical replay branch before that read.
+The qualification includes fresh research execution, actor/runnable/input
+refusals, exact lineage readback and replay after current input/Todo changes
+in the same facade-present/absent source and normal-wheel File/SQLite matrix.
+A whole-production Python import guard resolves relative imports and aliases;
+dynamic loading still requires the isolated real journeys. Keep this guard as
+new capabilities arrive, rather than treating a past inventory as a permanent
+exit. Last supported source writers, historical recovery, installed cohort
+and release exits remain independently open.
 
 The next T4 package is the [import-first normal-writer cutoff](../../reference/local-authority-provider-selection.md#proposed-compatibility-cutoff-and-release-sequence),
 not another re-export or producer-loading extraction. At `38cb407ed`, live

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

@@ -55,14 +55,21 @@ canonical Todo 入口隔离（T4）：五种 mutation facade 统一进入既有
 revision 保持。这为最后调用方删除建立边界，不宣称代码已退役：bootstrap、
 handoff、team-plan 和 Monitor 仍有受支持的旧源调用；capture/outbox 及历史恢复
 各等自己的出口。见[调用方及回滚清单](../../reference/canonical-todo-completion-update.md#canonical-mutation-entrypoint-isolation)。
-生产 Todo 消费者不再经历史 `loopx.todos` 重导出：CLI 注册／操作、quota 与规划、
+生产 Todo 消费者直接使用具体 owner：CLI 注册／操作、quota 与规划、
 能力写回、旧源租约读取及 Lark 源投影直接导入相同的 mutation、完整读取、路径或
 Markdown 显示 owner，不新增 wrapper、决策或 RPC。历史公开导入及其受支持的旧源
 IO 继续保留。隔离源码／正常 wheel 同时移除 facade、整个旧 Todo writer 和四个
 shadow producer，验证真实 File/SQLite CLI 生命周期、恰好一次结算、Monitor 重放
 及 App 冷导入／操作／重启；缺 provider／租约及源／备份变化须拒绝。Lark dry-run
-与 Explore 重放只验证本地消费者，不计 live 外部采用。此出口关闭生产重导出依赖，
-最后受支持的旧源 writer、历史恢复、安装 cohort 与发布验收继续保留。
+与 Explore 重放只验证本地消费者，不计 live 外部采用。
+此前调用方盘点后，[#5280](https://github.com/loopx-project/loopx/pull/5280) 新增的
+fresh research execution 又在函数内导入历史 facade；现直接导入既有
+`todos/list_readback.py`，保持相同 typed 当前 Todo 校验以及先于读取的历史重放。
+同一 facade 存在／缺席、源码／正常 wheel、File／SQLite 矩阵补 fresh 执行、
+actor／可运行状态／输入拒绝、精确 lineage 读回，以及当前输入／Todo 改变后的
+原观察重放。完整生产包 import 守卫解析相对导入与 alias，动态加载仍依赖真实
+隔离路径验证。新能力进入时持续守卫，不把历史库存当永久出口。最后受支持的
+旧源 writer、历史恢复、安装 cohort 与发布验收各自保持开放。
 
 保留的 import-compatible field writer 在既有 typed field plan 中统一等待条件
 归一化及矛盾 clear 拒绝；保持 Python 空白输入省略和锁内 IO，只退役重复的

loopx/capabilities/explore/research_evidence.py

@@ -114,7 +114,7 @@ def append_research_observation(
         if canonical.get("execution_lineage"):
             if registry_path is None or runtime_root is None:
                 raise ValueError("research execution lineage requires the selected registry and source runtime")
-            from ...todos import list_goal_todos
+            from ...control_plane.todos.list_readback import list_goal_todos
             from ...history import load_registry
             from ...materials import find_registry_goal
             from .research_frontier import build_research_composition_frontier, read_research_todo_history

tests/capabilities/test_explore_research_evidence.py

@@ -13,7 +13,7 @@
     build_explore_result_projection, explore_result_log_path, load_explore_result_events_strict,
 )
 from loopx.extensions.lark.presentation.explore_results import _node_record_values
-from loopx.todos import add_goal_todo, update_goal_todo
+from loopx.control_plane.todos.mutation_api import add_goal_todo, update_goal_todo
 
 
 GOAL = "research-fixture"

tests/control_plane/test_todo_consumer_host_retirement.py

@@ -97,3 +97,82 @@ def forbidden_runner(*args, **kwargs):
     assert result.returncode == 0, result.stdout + result.stderr
     assert json.loads(result.stdout) == {"todo_count": 1, "replayed": True}
     assert state.read_bytes() == original
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_fresh_research_execution_and_historical_replay_use_retained_owners(
+    tmp_path, provider, consumer_distribution,  # noqa: F811
+):
+    from canonical_authority_fixture import initialize_canonical_authority
+    from tests.capabilities import test_explore_research_evidence as research
+    from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+    from loopx.control_plane.todos.list_readback import list_goal_todos
+
+    # Seed through the existing source fixture, then use the selected copied
+    # distribution for every CLI decision. No editable finder can restore it.
+    registry, runtime, log, observation = research.execution_fixture(tmp_path)
+    state = tmp_path / "project/ACTIVE_GOAL_STATE.md"
+    todos = list_goal_todos(registry_path=registry, goal_id=research.GOAL, role="agent")["todos"]
+    initialize_canonical_authority(runtime, research.GOAL,
+        build_todo_runtime_shadow_projection(goal_id=research.GOAL, todos=todos, handoff_mode="soft_claim"),
+        state_path=state, provider=provider)
+    package, command = consumer_distribution
+    packet = tmp_path / "observation.json"
+    packet.write_text(json.dumps(observation))
+    registry_bytes = registry.read_bytes()
+
+    def cli(*args):
+        return subprocess.run([*command, "--registry", str(registry), "--runtime-root", str(runtime),
+                               "--format", "json", *args], cwd=tmp_path,
+                              capture_output=True, text=True, timeout=60)
+
+    def observe(*actor, expected_error=None):
+        before = log.read_bytes()
+        source_bytes = state.read_bytes()
+        result = cli("explore", "observe", "--goal-id", research.GOAL,
+                     "--observation-json", str(packet), *actor)
+        assert state.read_bytes() == source_bytes and registry.read_bytes() == registry_bytes
+        if expected_error:
+            assert result.returncode != 0 and expected_error in result.stdout + result.stderr
+            assert log.read_bytes() == before
+            return None
+        assert result.returncode == 0, result.stdout + result.stderr
+        return json.loads(result.stdout)
+
+    todo_id = observation["execution_lineage"]["successor_todo_id"]
+    observe(expected_error="actor differs")
+    observe("--agent-id", "other-agent", expected_error="actor differs")
+    deferred = cli("todo", "update", "--goal-id", research.GOAL, "--todo-id", todo_id,
+                   "--agent-id", "fixture-agent", "--status", "deferred",
+                   "--reason", "await-synthetic-capacity", "--resume-when", "capacity_available:fixture")
+    assert deferred.returncode == 0, deferred.stdout + deferred.stderr
+    observe("--agent-id", "fixture-agent", expected_error="runnable joint-probe")
+    reopened = cli("todo", "update", "--goal-id", research.GOAL, "--todo-id", todo_id,
+                   "--agent-id", "fixture-agent", "--status", "open", "--clear-resume-when")
+    assert reopened.returncode == 0, reopened.stdout + reopened.stderr
+    missing = json.loads(json.dumps(observation))
+    missing["execution_lineage"]["successor_todo_id"] = "todo_missing"
+    missing["progress"]["work_item_id"] = "todo_missing"
+    packet.write_text(json.dumps(missing))
+    observe("--agent-id", "fixture-agent", expected_error="runnable joint-probe")
+    packet.write_text(json.dumps(observation))
+    written = observe("--agent-id", "fixture-agent")
+    assert written["written"] and written["observation"]["execution_lineage"] == observation["execution_lineage"]
+    summary = cli("explore", "summary", "--goal-id", research.GOAL)
+    assert summary.returncode == 0, summary.stdout + summary.stderr
+    joint = next(node for node in json.loads(summary.stdout)["nodes"] if node["node_id"] == "joint")
+    assert joint["research_observation"]["execution_lineage"] == observation["execution_lineage"]
+    research.node(log, "a", status="open")
+    changed = json.loads(json.dumps(observation))
+    changed["progress"]["coverage_scope_id"] = "fresh-after-input-change"
+    packet.write_text(json.dumps(changed))
+    observe("--agent-id", "fixture-agent", expected_error="input")
+    changed_todo = cli("todo", "update", "--goal-id", research.GOAL, "--todo-id", todo_id,
+                       "--agent-id", "fixture-agent", "--action-kind", "inspect")
+    assert changed_todo.returncode == 0, changed_todo.stdout + changed_todo.stderr
+    packet.write_text(json.dumps(observation))
+    before = log.read_bytes()
+    replay = observe()
+    assert replay["replayed"] and not replay["written"] and log.read_bytes() == before
+    assert registry.read_bytes() == registry_bytes
+    assert (package / "control_plane/todos/legacy_mutation.py").exists() is False

tests/control_plane/test_todo_consumer_retirement.py

@@ -5,7 +5,10 @@
 """
 from __future__ import annotations
 
+import ast
+import importlib.util
 import json
+from pathlib import Path
 import subprocess
 import sys
 
@@ -17,6 +20,48 @@
 )
 
 
+def _historic_todo_imports(source, package):
+    """Resolve imports, including function-local relative imports and aliases."""
+    for node in ast.walk(ast.parse(source)):
+        targets = []
+        if isinstance(node, ast.Import):
+            targets = [alias.name for alias in node.names]
+        elif isinstance(node, ast.ImportFrom):
+            target = (importlib.util.resolve_name("." * node.level + (node.module or ""), package)
+                      if node.level else node.module or "")
+            targets = [target, *(target + "." + alias.name for alias in node.names)]
+        if any(target == "loopx.todos" or target.startswith("loopx.todos.") for target in targets):
+            yield node.lineno
+
+
+@pytest.mark.parametrize("source,package", [
+    ("import loopx.todos as historic", "loopx.capabilities.explore"),
+    ("from loopx import todos as historic", "loopx.capabilities.explore"),
+    ("def read():\n from ...todos import list_goal_todos", "loopx.capabilities.explore"),
+    ("from . import todos", "loopx"),
+])
+def test_import_guard_recognizes_historic_owner(source, package):
+    assert list(_historic_todo_imports(source, package))
+
+
+def test_import_guard_keeps_retained_todo_package_distinct():
+    assert not list(_historic_todo_imports("from ..todos import legacy_mutation", "loopx.control_plane.todos"))
+    assert not list(_historic_todo_imports(
+        "from ...control_plane.todos.list_readback import list_goal_todos", "loopx.capabilities.explore"))
+
+
+def test_historic_facade_has_no_production_consumers():
+    # An earlier caller inventory cannot cover imports added by a later feature.
+    # Dynamic imports still need the physically isolated production journeys.
+    root = Path(__file__).resolve().parents[2]
+    consumers = []
+    for path in sorted((root / "loopx").rglob("*.py")):
+        relative = path.relative_to(root)
+        package = ".".join(relative.parent.parts)
+        consumers.extend((str(relative), line) for line in _historic_todo_imports(path.read_text(encoding="utf-8"), package))
+    assert consumers == [], consumers
+
+
 @pytest.fixture(params=[True, False], ids=["facade-present", "facade-absent"])
 def consumer_distribution(request, isolated_todo_distribution):  # noqa: F811
     package, command = isolated_todo_distribution
@@ -33,7 +78,19 @@ def consumer_distribution(request, isolated_todo_distribution):  # noqa: F811
     provenance = json.loads(probe.stdout)
     assert provenance["package"] == str(package / "__init__.py")
     assert provenance["facade"] is request.param
-    return package, command
+    try:
+        yield package, command
+    finally:
+        # Both fixture seeding and the isolated CLI start owned runtimes under
+        # this fixture's private TMPDIR, never the shared developer service.
+        stopped = subprocess.run(
+            [sys.executable, "-S", "-c",
+             "from loopx.control_plane.effect_runtime import restart_effect_runtime;restart_effect_runtime()"],
+            cwd=package.parent, capture_output=True, text=True, timeout=15,
+        )
+        assert stopped.returncode == 0, stopped.stdout + stopped.stderr
+        from loopx.control_plane.effect_runtime import restart_effect_runtime
+        restart_effect_runtime()
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])