Keep saved research reads responsive during history loading

loopx/control_plane/status/goal_source.py

@@ -17,7 +17,7 @@
 import re
 import shlex
 import subprocess
-from threading import RLock
+from threading import Lock, RLock
 from time import monotonic
 from urllib.error import HTTPError, URLError
 from urllib.parse import parse_qs, urlencode, urlparse
@@ -241,6 +241,7 @@ def __init__(self, source, *, history_projector):
         self.source = source
         self._history_projector = history_projector
         self._lock = RLock()
+        self._page_lock = Lock()
         self._status = None
         self._read_at = 0.0
         self._selected = None
@@ -251,9 +252,11 @@ def _invalidate(self):
         self._pages = CompletedTodoPages()
 
     def status(self, goal_id=None, *, fresh=False):
+        # A bad selector belongs to this request, not the source's page snapshot.
+        selected = self.source.check_scope(goal_id)
         with self._lock:
             try:
-                selected = self.source.check_access(goal_id)
+                self.source.check_access(selected)
                 if (fresh or self._status is None or selected != self._selected
                         or monotonic() - self._read_at >= 2):
                     self._status = None
@@ -286,41 +289,40 @@ def goals(self, goal_id=None, *, fresh=False):
         return {"ok": True, "source": self.source.describe(),
                 "items": status["run_history"]["goals"]}
 
-    def history(self, goal_id, cursor=""):
-        with self._lock:
-            if goal_id is None:
-                raise SourceScopeError("Choose a Goal before reading history")
-            if not isinstance(cursor, str) or len(cursor) > 512:
-                raise ValueError("history_cursor_expired")
-            # Membership/SSH access is rechecked before any saved cursor is used.
-            self.status(goal_id)
+    def _read_page(self, goal_id, cursor, *, scope, load):
+        # Keep page loads ordered without holding up an independent status read.
+        with self._page_lock:
+            with self._lock:
+                self.status(goal_id)
+                pages = self._pages
             try:
-                page = self._pages.page(scope=(self.source.namespace, goal_id), cursor=cursor,
-                    load=lambda: self._history_projector(self.source.read_events(goal_id), goal_id=goal_id))
-                return {**page, "source": self.source.describe(), "goal_id": goal_id}
+                page = pages.page(scope=scope, cursor=cursor, load=load)
             except (ValueError, OSError, subprocess.SubprocessError) as exc:
-                if str(exc) != "history_cursor_expired":
-                    self._invalidate()
+                with self._lock:
+                    if pages is self._pages and str(exc) != "history_cursor_expired":
+                        self._invalidate()
                 raise
+            with self._lock:
+                if pages is not self._pages:
+                    raise SourceScopeError("The original Goal source became unavailable during the read")
+            return {**page, "source": self.source.describe(), "goal_id": goal_id}
+
+    def history(self, goal_id, cursor=""):
+        if goal_id is None:
+            raise SourceScopeError("Choose a Goal before reading history")
+        if not isinstance(cursor, str) or len(cursor) > 512:
+            raise ValueError("history_cursor_expired")
+        return self._read_page(goal_id, cursor, scope=(self.source.namespace, goal_id),
+            load=lambda: self._history_projector(self.source.read_events(goal_id), goal_id=goal_id))
 
     def findings(self, goal_id, cursor="", *, projector):
         """Read saved evidence; this transport has no complete task-list view."""
-        with self._lock:
-            if goal_id is None:
-                raise SourceScopeError("Choose a Goal before reading findings")
-            if not isinstance(cursor, str) or len(cursor) > 512:
-                raise ValueError("history_cursor_expired")
-            self.status(goal_id)
-            try:
-                page = self._pages.page(
-                    scope=(self.source.namespace, goal_id, "findings"), cursor=cursor,
-                    load=lambda: projector(self.source.read_events(goal_id),
-                                           None, goal_id=goal_id))
-                return {**page, "source": self.source.describe(), "goal_id": goal_id}
-            except (ValueError, OSError, subprocess.SubprocessError) as exc:
-                if str(exc) != "history_cursor_expired":
-                    self._invalidate()
-                raise
+        if goal_id is None:
+            raise SourceScopeError("Choose a Goal before reading findings")
+        if not isinstance(cursor, str) or len(cursor) > 512:
+            raise ValueError("history_cursor_expired")
+        return self._read_page(goal_id, cursor, scope=(self.source.namespace, goal_id, "findings"),
+            load=lambda: projector(self.source.read_events(goal_id), None, goal_id=goal_id))
 
 
 class _NoRedirect(HTTPRedirectHandler):

tests/presentation/test_linked_goal_sources.py

@@ -9,6 +9,7 @@
 import os
 from pathlib import Path
 import sys
+from threading import Event
 import time
 from urllib.error import HTTPError
 from urllib.parse import urlencode
@@ -187,6 +188,64 @@ def test_app_parent_directory_reads_original_tasks_history_and_explicit_analysis
     assert all(path.read_bytes() == data for path, data in before.items())
 
 
+@pytest.mark.parametrize(("view", "source_fails"), [
+    ("history", False), ("findings", False), ("history", True),
+])
+def test_linked_analysis_does_not_wait_for_history_page_io(workspace, monkeypatch, view, source_fails):
+    raw, request, before = native_source(workspace)
+    request('/api/chat/goal-sources', {'parent_goal_id': 'research-owner', 'source': raw})
+    namespace = GoalSource.parse(raw).namespace
+    analysis_runtime = workspace['tmp_path'] / 'runtime' / 'goal-sources' / namespace
+    pack = register_pack(runtime_root=analysis_runtime, owner_goal_id='solver-goal',
+        input_path=workspace['tmp_path'] / 'input.json', source_root=workspace['sources'],
+        title='Original source evidence')
+    query = {'goal_id': 'solver-goal', 'source_namespace': namespace}
+    catalog = '/api/chat/research-analysis?' + urlencode(query)
+    history = '/api/chat/explore-results?' + urlencode({**query, 'view': view})
+    fresh = '/api/chat/explore-results?' + urlencode({**query, 'view': 'history-timing'})
+    entered, resume = Event(), Event()
+    read_events = GoalSource.read_events
+
+    def slow_events(self, goal_id):
+        if self.namespace == namespace:
+            entered.set()
+            assert resume.wait(25)
+        return read_events(self, goal_id)
+
+    monkeypatch.setattr(GoalSource, 'read_events', slow_events)
+    source_registry = Path(raw['registry'])
+    try:
+        with ThreadPoolExecutor(max_workers=2) as clients:
+            pending_page = clients.submit(request, history)
+            try:
+                assert entered.wait(12)
+                if source_fails:
+                    source_registry.unlink()
+                    with pytest.raises(HTTPError) as unavailable:
+                        clients.submit(request, fresh).result(timeout=7)
+                    assert unavailable.value.code == 503
+                else:
+                    analysis = clients.submit(request, catalog).result(timeout=7)
+                    assert analysis['pack']['pack_id'] == pack['pack_id']
+                assert not pending_page.done()
+            finally:
+                resume.set()
+            if source_fails:
+                with pytest.raises(HTTPError) as invalidated:
+                    pending_page.result(timeout=12)
+                assert invalidated.value.code == 409
+            else:
+                page = pending_page.result(timeout=12)
+                assert page['source']['namespace'] == namespace
+                assert page['total'] == (3 if view == 'history' else 0)
+    finally:
+        if source_fails:
+            source_registry.write_bytes(before[source_registry])
+    assert request(catalog)['pack']['pack_id'] == pack['pack_id']
+    assert all(path.read_bytes() == data for path, data in before.items())
+    assert not workspace['count'].exists()
+
+
 @pytest.mark.parametrize('retry', [False, True])
 @pytest.mark.parametrize('failure', ['removed', 'unavailable'])
 def test_app_warm_read_cannot_admit_analysis_after_original_goal_disappears(workspace, retry, failure):

tests/presentation/test_runtime_workspace.py

@@ -1,12 +1,13 @@
 """Real HTTP boundary: source identity, read-only access and full history."""
 
+from concurrent.futures import ThreadPoolExecutor
 from contextlib import contextmanager
 import json
 import os
 from pathlib import Path
 import subprocess
 import sys
-from threading import Thread
+from threading import Event, Thread
 from urllib.error import HTTPError
 from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
 from urllib.request import Request, urlopen
@@ -120,9 +121,9 @@ def test_failed_refresh_does_not_serve_old_success_and_recovers(tmp_path):
         assert get(server, f"/api/chat/explore-results?view=history&goal_id=shared-goal&cursor={cursor}")[0] == 409
 
 
-@pytest.mark.parametrize("invalid_kind", ["null", "event-pairs"])
-def test_invalid_native_history_clears_only_its_source_and_recovers(tmp_path, invalid_kind):
-    """Exercise actual native CLI/log IO through both shared HTTP consumers."""
+@pytest.fixture
+def native_history(tmp_path):
+    """A pinned source backed by the native CLI and a disposable SQLite Goal."""
     project, runtime = tmp_path / "project", tmp_path / "runtime"
     project.mkdir()
     (project / "goal.md").write_text(
@@ -145,8 +146,16 @@ def test_invalid_native_history_clears_only_its_source_and_recovers(tmp_path, in
                                      title=f"Question {i}") for i in range(43)]
     log = runtime / "goals" / "shared-goal" / "explore-result-log.jsonl"
     log.parent.mkdir(parents=True, exist_ok=True)
-    valid = "\n".join(json.dumps(event) for event in events).encode()
-    log.write_bytes(valid)
+    log.write_text("\n".join(json.dumps(event) for event in events))
+    return source, log, events
+
+
+@pytest.mark.parametrize("invalid_kind", ["null", "event-pairs"])
+def test_invalid_native_history_clears_only_its_source_and_recovers(tmp_path, invalid_kind, native_history):
+    """Exercise actual native CLI/log IO through both shared HTTP consumers."""
+    source, log, events = native_history
+    project = Path(source.registry).parent
+    valid = log.read_bytes()
     routes = ("/api/goal-source/history?goal_id=shared-goal",
               "/api/chat/explore-results?view=history&goal_id=shared-goal")
     with serving(source, tmp_path) as server, serving(Source("untouched"), tmp_path) as other:
@@ -178,6 +187,46 @@ def test_invalid_native_history_clears_only_its_source_and_recovers(tmp_path, in
         assert log.read_bytes() == valid
 
 
+@pytest.mark.parametrize("wrong_goal", ["another-goal", ""])
+def test_invalid_selector_does_not_invalidate_authorized_history(tmp_path, native_history, monkeypatch, wrong_goal):
+    source, log, events = native_history
+    registry = Path(source.registry)
+    original = {path: path.read_bytes() for path in (log, registry)}
+    history = "/api/goal-source/history?goal_id=shared-goal"
+    entered, release = Event(), Event()
+    read_events = GoalSource.read_events
+
+    def hold_after_native_read(self, goal_id):
+        rows = read_events(self, goal_id)
+        if self is source:
+            entered.set()
+            assert release.wait(20)
+        return rows
+
+    with serving(source, tmp_path) as server:
+        code, first = get(server, history, timeout=30)
+        assert code == 200 and first["total"] == 43 and first["next_cursor"]
+        monkeypatch.setattr(GoalSource, "read_events", hold_after_native_read)
+        with ThreadPoolExecutor(max_workers=1) as clients:
+            pending = clients.submit(get, server, history, timeout=30)
+            try:
+                assert entered.wait(12)
+                code, error = get(server, "/status.json?" + urlencode({"goal_id": wrong_goal}))
+                assert code == 404 and error["ok"] is False and "items" not in error
+                assert not pending.done()
+            finally:
+                release.set()
+            code, latest = pending.result(timeout=15)
+        assert code == 200 and latest["total"] == first["total"]
+        assert latest["items"] == first["items"] and latest["next_cursor"]
+        for page in (first, latest):
+            code, tail = get(server, f'{history}&cursor={page["next_cursor"]}', timeout=30)
+            assert code == 200 and tail["next_cursor"] is None
+            assert {row["event_id"] for row in page["items"] + tail["items"]} == {
+                event["event_id"] for event in events}
+    assert all(path.read_bytes() == data for path, data in original.items())
+
+
 def test_cli_agent_and_app_use_the_same_reader_and_source_binding(tmp_path):
     source = Source("original")
     with serving(source, tmp_path) as server: