diff --git a/loopx/control_plane/status/goal_source.py b/loopx/control_plane/status/goal_source.py index a79afc6e91..32bcb37789 100644 --- a/loopx/control_plane/status/goal_source.py +++ b/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): diff --git a/tests/presentation/test_linked_goal_sources.py b/tests/presentation/test_linked_goal_sources.py index 8a7a6d64da..15c7e10539 100644 --- a/tests/presentation/test_linked_goal_sources.py +++ b/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): diff --git a/tests/presentation/test_runtime_workspace.py b/tests/presentation/test_runtime_workspace.py index 29e286617c..640d36a986 100644 --- a/tests/presentation/test_runtime_workspace.py +++ b/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: