From 2ffc25e152b44dceea3a12a75fd38164627342b8 Mon Sep 17 00:00:00 2001 From: song Date: Mon, 26 Oct 2026 10:55:25 +0800 Subject: [PATCH] fix(runtime): allow scoped analysis cancellation while source is offline Keep namespace, original Goal and existing pack/job identity checks for local cancellation. Preserve fresh source membership checks for starts and retries, and add HTTP/process regressions for offline cancellation. Related to #6878. Signed-off-by: Codex --- docs/reference/trajectory-assessment.md | 4 + loopx/control_plane/status/goal_source.py | 7 +- loopx/presentation/runtime_workspace.py | 14 ++-- .../test_research_analysis_api.py | 73 +++++++++++++++++++ 4 files changed, 92 insertions(+), 6 deletions(-) diff --git a/docs/reference/trajectory-assessment.md b/docs/reference/trajectory-assessment.md index d69ec10b1d..f86cc0fd6e 100644 --- a/docs/reference/trajectory-assessment.md +++ b/docs/reference/trajectory-assessment.md @@ -790,6 +790,10 @@ It grants no source-Goal lifecycle, conversation, scoring or solver writes. Stop and restart the reader without the option to disable local execution; historical packs remain on disk. Cancel running analysis through its existing action before stopping the server. +Cancellation uses the prepared source namespace, original Goal and existing +local pack/job identity, so it remains available when the original source is +offline. Starting or retrying analysis still requires a fresh source read and +current registered membership. The trusted curator registers authorized inputs through the existing `trajectory-pack-register` command. For each source, use runtime root diff --git a/loopx/control_plane/status/goal_source.py b/loopx/control_plane/status/goal_source.py index 92c2e5a724..f133ea243a 100644 --- a/loopx/control_plane/status/goal_source.py +++ b/loopx/control_plane/status/goal_source.py @@ -143,12 +143,17 @@ def namespace(self): # A reused display id cannot retain another authority's cache or cursor. return sha256(json.dumps(asdict(self), sort_keys=True).encode()).hexdigest() - def check_access(self, goal_id=None): + def check_scope(self, goal_id=None): + """Validate the prepared Goal selection without accessing its connection.""" selected = self.goal_id if goal_id is None else goal_id if selected is not None and (not isinstance(selected, str) or not _GOAL_ID.fullmatch(selected)): raise SourceScopeError("Invalid source Goal id") if self.goal_id is not None and selected != self.goal_id: raise SourceScopeError("Goal is not in this source") + return selected + + def check_access(self, goal_id=None): + selected = self.check_scope(goal_id) # Recheck even on cached reads; a removed alias is not a retained grant. if self.connection == "ssh" and self.host_alias not in configured_ssh_host_aliases(): raise SourceScopeError("The configured SSH source is unavailable") diff --git a/loopx/presentation/runtime_workspace.py b/loopx/presentation/runtime_workspace.py index b347623a91..9390ea338f 100644 --- a/loopx/presentation/runtime_workspace.py +++ b/loopx/presentation/runtime_workspace.py @@ -104,12 +104,16 @@ def _read_json(self, *, max_bytes): return parse_strict_json_object(self.rfile.read(length)) def _research_analysis_scope(self, goal_id, *, cancel=False): - # Check current access and actual registered membership before local - # pack reads or provider effects. No host registry entry is created. try: - # Display reads may share their short cache. A provider operation - # must check the original source again before admitting any work. - self.server.reader.registered_status(goal_id, fresh=self.command == "POST") + if cancel: + # The shared handler pins the namespace; the toolkit still + # verifies the local pack/key and existing job before stopping + # it. Source availability is admission for new work only. + self.server.source.check_scope(goal_id) + else: + # Display reads may share their short cache. Starts and retries + # must check the original source before admitting new work. + self.server.reader.registered_status(goal_id, fresh=self.command == "POST") except SourceScopeError: self._send_error("Goal or connection is not available in this source.", status=404) return None diff --git a/tests/presentation/test_research_analysis_api.py b/tests/presentation/test_research_analysis_api.py index b2d18b6e4a..22977c0c3b 100644 --- a/tests/presentation/test_research_analysis_api.py +++ b/tests/presentation/test_research_analysis_api.py @@ -533,6 +533,79 @@ def test_source_goal_analysis_reads_cancel_retry_and_reuse_do_not_mutate_source( assert changed.value.code == 409 +@pytest.mark.parametrize("registered,failed,code", [(True, True, 503), (False, False, 404)]) +def test_source_analysis_cancel_does_not_require_live_source(workspace, registered, failed, code): + w = workspace + with source_analysis_workspace(w, "offline-cancel") as s: + read, operation, source = s["request"], s["operation"], s["server"].source + with ThreadPoolExecutor(max_workers=1) as pool: + running = pool.submit(read, body=operation) + until = time.monotonic() + 8 + while not w["count"].exists() and time.monotonic() < until: + time.sleep(.05) + assert w["count"].exists(), "Local analysis process did not start" + reads = source.reads + source.registered, source.failed = registered, failed + try: + assert read(body=operation, suffix="/cancel")["job"]["cancel_requested"] + assert running.result(timeout=8)["job"]["status"] == "cancelled" + assert source.reads == reads, "Cancelling a local job must not read its original source" + for retry in (False, True): + with pytest.raises(HTTPError) as denied: + read(body={**operation, "retry": retry}) + assert denied.value.code == code + assert w["count"].read_text().splitlines() == ["started"] + finally: + # The same test must clean up its child when run against the + # defective baseline, where offline cancellation is rejected. + source.registered, source.failed = True, False + read(body=operation, suffix="/cancel") + running.result(timeout=8) + assert read()["pack"]["job"]["status"] == "cancelled" + for path, original in w["before"].items(): + assert path.read_bytes() == original + + +def test_source_analysis_offline_cancel_requires_pinned_job_identity(workspace): + w = workspace + with source_analysis_workspace(w, "cancel-identity") as s: + read, operation, source = s["request"], s["operation"], s["server"].source + source.failed = True + with pytest.raises(HTTPError): + read(body=operation, suffix="/cancel") + assert not w["count"].exists(), "Cancellation cannot create a missing local job" + source.failed = False + with ThreadPoolExecutor(max_workers=1) as pool: + running = pool.submit(read, body=operation) + until = time.monotonic() + 8 + while not w["count"].exists() and time.monotonic() < until: + time.sleep(.05) + assert w["count"].exists(), "Local analysis process did not start" + source.failed = True + try: + for namespace in (None, "0" * 64): + with pytest.raises(HTTPError) as denied: + read(body=operation, suffix="/cancel", namespace=namespace) + assert denied.value.code == 400 + for field, value, code in ( + ("goal_id", "another-goal", 404), + ("pack_id", "0" * 64, 409), + ("analysis_key", "0" * 64, 409), + ): + with pytest.raises(HTTPError) as denied: + read(body={**operation, field: value}, suffix="/cancel") + assert denied.value.code == code + assert not list(s["server"].analysis_runtime_root.rglob("cancel.request")) + assert not running.done(), "A mismatched selector must not cancel the running job" + assert read(body=operation, suffix="/cancel")["job"]["cancel_requested"] + assert running.result(timeout=8)["job"]["status"] == "cancelled" + assert w["count"].read_text().splitlines() == ["started"] + finally: + source.failed = False + read(body=operation, suffix="/cancel") + running.result(timeout=8) + + def test_source_analysis_requires_exact_namespace_goal_and_live_membership(workspace): w = workspace with source_analysis_workspace(w, "retained") as a, source_analysis_workspace(w, "fresh") as b: