fix(research): restore original history and saved analysis on shared Core

loopx/chat_completed_todos.py

@@ -11,6 +11,7 @@
 from urllib.parse import parse_qs, urlparse
 
 from .paths import resolve_runtime_root
+from .capabilities.native_chat.project_context import coordination_runtime_root
 from .status_server import is_loopback_host
 
 
@@ -128,9 +129,10 @@ def _goal_result_scope(self, goal_id):
         if not self._require_loopback_origin():
             return None
         registry, _goal = self._registry_and_goal(goal_id)
-        return resolve_runtime_root(
-            registry, self.server.runtime_root_override,
-            registry_path=self.server.registry_path,
+        return coordination_runtime_root(
+            self.server.registry_path,
+            resolve_runtime_root(registry, self.server.runtime_root_override,
+                                 registry_path=self.server.registry_path),
         )
 
     def _goal_results(self) -> None:
@@ -193,13 +195,20 @@ def _completed_todos(self) -> None:
             if len(search) > 256:
                 raise ValueError("history_search_too_long")
             search = search.casefold()
-            self._registry_and_goal(goal_id)
+            registry, _goal = self._registry_and_goal(goal_id)
+            runtime_root_arg = self.server.runtime_root_override
+            if registry.get("common_runtime_root"):
+                runtime_root_arg = str(coordination_runtime_root(
+                    self.server.registry_path,
+                    resolve_runtime_root(registry, runtime_root_arg,
+                                         registry_path=self.server.registry_path),
+                ))
 
             def load():
                 payload = list_goal_todos(
                     registry_path=self.server.registry_path, goal_id=goal_id,
                     role="agent", status="done", read_scope="completed_history",
-                    runtime_root_arg=self.server.runtime_root_override,
+                    runtime_root_arg=runtime_root_arg,
                 )
                 ordered = sorted(enumerate(payload["todos"]), key=lambda pair: (str(pair[1].get("completed_at") or ""), pair[0]), reverse=True)
                 return [

loopx/presentation/goal_source_routing.py

@@ -15,7 +15,10 @@ def goal_source_catalog(server):
     catalog = getattr(server, "goal_source_catalog", None)
     if catalog is None:
         from ..capabilities.explore.result_log import explore_history_rows
-        catalog = GoalSourceCatalog(server.registry_path, server.runtime_root, history_projector=explore_history_rows)
+        from ..capabilities.native_chat.project_context import coordination_runtime_root
+        catalog = GoalSourceCatalog(server.registry_path,
+            coordination_runtime_root(server.registry_path, server.runtime_root),
+            history_projector=explore_history_rows)
         server.goal_source_catalog = catalog
     return catalog
 

tests/presentation/test_linked_goal_sources.py

@@ -77,9 +77,13 @@ def request(path, body=None, origin=None):
     return raw, request, {path: path.read_bytes() for path in (registry, state, log)}
 
 
-def test_app_parent_directory_reads_original_tasks_history_and_explicit_analysis(workspace):
+@pytest.mark.parametrize("separate_chat_store", [False, True])
+def test_app_parent_directory_reads_original_tasks_history_and_explicit_analysis(workspace, separate_chat_store):
     w = workspace
     raw, request, before = native_source(w)
+    if separate_chat_store:
+        w['server'].runtime_root = w['tmp_path'] / 'private-chat'
+        w['server'].runtime_root_override = str(w['server'].runtime_root)
     assert request('/api/chat/goal-sources', {'parent_goal_id': 'research-owner', 'source': raw})['ok']
     namespace = GoalSource.parse(raw).namespace
     directory = request('/status.json?view=workspace-directory')
@@ -99,7 +103,7 @@ def test_app_parent_directory_reads_original_tasks_history_and_explicit_analysis
         'goal_id': 'solver-goal', 'source_namespace': namespace, 'view': 'history'}))
     assert history['total'] == 3 and any(row.get('edge_type') == 'depends_on' for row in history['items'])
 
-    analysis_runtime = w['server'].runtime_root / 'goal-sources' / namespace
+    analysis_runtime = w['tmp_path'] / 'runtime' / 'goal-sources' / namespace
     pack = register_pack(runtime_root=analysis_runtime, owner_goal_id='solver-goal',
         input_path=w['tmp_path'] / 'input.json', source_root=w['sources'], title='Original source evidence')
     query = urlencode({'goal_id': 'solver-goal', 'source_namespace': namespace})
@@ -118,9 +122,7 @@ def test_app_parent_directory_reads_original_tasks_history_and_explicit_analysis
     host_registry = json.loads(w['server'].registry_path.read_text())
     assert [goal['id'] for goal in host_registry['goals']] == ['research-owner', 'other-owner']
     # A new App service reads the persisted relationship, not a browser URL or in-memory mirror.
-    from loopx.control_plane.status.goal_source import GoalSourceCatalog
-    w['server'].goal_source_catalog = GoalSourceCatalog(w['server'].registry_path, w['server'].runtime_root,
-                                                     history_projector=explore_history_rows)
+    del w['server'].goal_source_catalog
     assert request(endpoint + '?' + query)['pack']['job']['status'] == 'succeeded'
     assert len(w['count'].read_text().splitlines()) == 1
     bound = w['server'].registry_path.read_bytes()

tests/presentation/test_research_analysis_api.py

@@ -24,7 +24,7 @@
 
 
 @pytest.fixture
-def workspace(tmp_path, monkeypatch):
+def workspace(tmp_path, monkeypatch, request):
     runtime, sources = tmp_path / "runtime", tmp_path / "sources"
     sources.mkdir()
     evidence = b"Prior question\nOnly the diagnostic was measured; native utility remains untested.\n"
@@ -75,7 +75,9 @@ def workspace(tmp_path, monkeypatch):
     registry.write_text(json.dumps({"common_runtime_root": str(runtime), "goals": [{"id": goal, "repo": str(tmp_path), "state_file": "active.md"} for goal in ("research-owner", "other-owner")]}))
     before = {path: path.read_bytes() for path in (state, registry, input_path, sources / "facts.txt")}
     server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler)
-    server.registry_path, server.runtime_root_override, server.verbose = registry, None, False
+    server.registry_path, server.runtime_root_override, server.verbose = (
+        registry, str(tmp_path / "private-chat") if getattr(request, "param", False) else None, False
+    )
     worker = threading.Thread(target=server.serve_forever, daemon=True)
     worker.start()
     base = f"http://127.0.0.1:{server.server_port}/api/chat/research-analysis"
@@ -97,6 +99,7 @@ def request(*, goal="research-owner", body=None, suffix="", headers=None, **quer
         server.server_close()
 
 
+@pytest.mark.parametrize("workspace", [False, True], indirect=True)
 def test_shared_excerpts_reverify_through_http_without_dispatch(workspace, monkeypatch):
     w = workspace
     path = w["tmp_path"] / "input.json"
@@ -335,6 +338,7 @@ def test_registered_http_cancel_failure_retry_reuse_and_source_return(workspace)
     assert changed.value.code == 409
 
 
+@pytest.mark.parametrize("workspace", [False, True], indirect=True)
 def test_http_rejects_other_owner_paths_free_configuration_and_wrong_origin(workspace):
     w = workspace
     read, operation = w["request"], w["operation"]
@@ -354,6 +358,7 @@ def test_http_rejects_other_owner_paths_free_configuration_and_wrong_origin(work
     assert not w["count"].exists()
 
 
+@pytest.mark.parametrize("workspace", [False, True], indirect=True)
 def test_native_outcomes_and_bound_citations_read_over_http_without_analysis(workspace):
     w = workspace
     data = json.loads((w["tmp_path"] / "input.json").read_text())

tests/test_chat_completed_todos.py

@@ -182,7 +182,7 @@ def read(**kwargs):
 
 @pytest.mark.parametrize("count", [85, 4087])
 @pytest.mark.parametrize("archived", [False, True])
-def test_http_history_reads_real_markdown_without_writes(tmp_path, count, archived):
+def test_http_history_reads_full_retained_content_without_writes(tmp_path, count, archived):
     state = tmp_path / "active.md"
     text = f"Inspect {tmp_path}/results.txt " + " ".join(["complete task description"] * 25)
     evidence = f"Verified output at {tmp_path}/results.txt"
@@ -203,7 +203,18 @@ def test_http_history_reads_real_markdown_without_writes(tmp_path, count, archiv
             lines.insert(first + 2, "\n## Completed Work Archive\n")
             state.write_text("\n".join(lines).replace("status=done ", "role=agent status=done "))
     registry = tmp_path / "registry.json"
-    registry.write_text(json.dumps({"runtime_root": str(tmp_path / "runtime"), "goals": [{"id": "history-goal", "repo": str(tmp_path), "state_file": "active.md"}]}))
+    runtime = tmp_path / "runtime"
+    registry.write_text(json.dumps({"common_runtime_root": str(runtime), "goals": [{"id": "history-goal", "repo": str(tmp_path), "state_file": "active.md"}]}))
+    from tests.control_plane.canonical_authority_fixture import initialize_canonical_authority
+    from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+    from loopx.control_plane.todos.active_state_todo_parser import parse_todo_source
+    active, history, _ = parse_todo_source(state.read_text())
+    rows = active["agent"] + history
+    for row in rows:
+        row["schema_version"] = "todo_item_v0"
+    initialize_canonical_authority(runtime, "history-goal",
+        build_todo_runtime_shadow_projection(goal_id="history-goal", todos=rows, handoff_mode="soft_claim"),
+        state_path=state, provider="file")
     before = state.read_bytes()
     server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler)
     server.registry_path = registry
@@ -244,7 +255,8 @@ def test_http_history_reads_real_markdown_without_writes(tmp_path, count, archiv
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
-def test_canonical_history_includes_archives_without_widening_active_lists(tmp_path, monkeypatch, provider):
+@pytest.mark.parametrize("separate_chat_store", [False, True])
+def test_canonical_history_includes_archives_without_widening_active_lists(tmp_path, monkeypatch, provider, separate_chat_store):
     from pathlib import Path
     monkeypatch.syspath_prepend(str(Path(__file__).parent / "control_plane"))
     from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
@@ -303,7 +315,9 @@ def test_canonical_history_includes_archives_without_widening_active_lists(tmp_p
     with pytest.raises(ValueError, match="read_scope must be"):
         list_goal_todos(registry_path=registry, goal_id="history-goal", read_scope="unknown")
     server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler)
-    server.registry_path, server.runtime_root_override, server.verbose = registry, runtime, False
+    server.registry_path, server.runtime_root_override, server.verbose = (
+        registry, tmp_path / "private-chat" if separate_chat_store else runtime, False
+    )
     worker = threading.Thread(target=server.serve_forever, daemon=True)
     worker.start()
     try:

tests/test_chat_explore_results.py

@@ -19,7 +19,8 @@
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
-def test_ninth_link_readback_pagination_scope_failure_and_recovery(tmp_path, monkeypatch, provider):
+@pytest.mark.parametrize("separate_chat_store", [False, True])
+def test_ninth_link_readback_pagination_scope_failure_and_recovery(tmp_path, monkeypatch, provider, separate_chat_store):
     registry = tmp_path / "registry.json"
     runtime = tmp_path / "runtime"
     (tmp_path / "active.md").write_text("# Synthetic Goal\n")
@@ -88,7 +89,8 @@ def test_ninth_link_readback_pagination_scope_failure_and_recovery(tmp_path, mon
     state_before = state.read_bytes()
     server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler)
     server.registry_path = registry
-    server.runtime_root_override = None
+    server.runtime_root_override = str(tmp_path / "private-chat") if separate_chat_store else None
+    server.runtime_root = tmp_path / "private-chat" if separate_chat_store else runtime
     server.verbose = False
     worker = threading.Thread(target=server.serve_forever, daemon=True)
     worker.start()
@@ -99,6 +101,8 @@ def read(url):
     try:
         page = read(base + "evidence-goal")
         assert page["total"] == 41 and len(page["items"]) == 40
+        history = read(base + "evidence-goal&view=history")
+        assert history["total"] == 42 and len(history["items"]) == 40
         item = page["items"][0]
         assert item["scope"] == "Inputs with a finite prefix"
         assert item["status"] == "refuted"

tests/test_managed_research_team.py

@@ -134,7 +134,7 @@ def test_canonical_delivery_requires_completed_current_dependencies(team, monkey
     assert result_read["result"]["sha256"] == lead_completion["completion_result"]["sha256"]
     server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler)
     server.registry_path = root / "registry.json"
-    server.runtime_root_override = str(root / "runtime")
+    server.runtime_root_override = str(root / "separate-chat-store")
     server.completed_todo_pages = CompletedTodoPages()
     server.verbose = False
     worker = threading.Thread(target=server.serve_forever, daemon=True)