fix(runtime): allow offline cancellation of existing source analyses

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

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")

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

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: