Add offline JCZ cancellation recovery

docs/research/rayforge-jcz-cancel-recovery-spike.md

@@ -0,0 +1,43 @@
+# Rayforge JCZ offline cancellation and recovery spike
+
+- Date: 2026-10-19
+- Rayforge commit: `c28918c87f8273dd1d244f5a057addc192807fa5`
+- Scope: asynchronous no-device driver lifecycle; no transport
+
+## Result
+
+The external JCZ addon now exercises Rayforge's asynchronous `run`,
+`set_hold` and `cancel` driver seam with an in-memory state machine. A
+synthetic 260-line job produced two 3072-byte list buffers, entered running,
+hold and resumed states, then cancelled. Cancellation cleared all 6144 buffered
+bytes and moved to a terminal `cancelled` state.
+
+Restart without an explicit recovery transition was rejected. After recovery
+returned the session to idle, a separately encoded two-line job completed.
+The complete event sequence is checked in
+`tests/fixtures/rayforge-jcz-cancel-recovery.json`.
+
+Two runs produced identical output with SHA-256
+`943cb35432f58524c32fcf3bffb146ce1b57f6c6573870429158dd6e6550728a`.
+
+## Evidence and safety boundary
+
+The logical sequence follows the pinned BalorMK distinction between
+pause/resume and abort/reset, but it does not encode or issue BalorMK's
+`StopList`, `RestartList`, `StopExecute`, `ResetList`, port or source-control
+commands. Those are hardware behaviors and remain outside this offline spike.
+
+The mock records no hardware commands and has no discovery, USB, serial or
+socket transport. Windows asyncio internally requires a local event-loop
+wakeup socket pair, so this test may need to run outside the filesystem/network
+sandbox; that implementation detail is not reachable from the addon and does
+not contact a remote endpoint.
+
+## Remaining gap
+
+This proves the Rayforge cancellation contract and local buffer invalidation,
+not physical controller cancellation or recovery. A future M3 protocol must
+separately authorize and verify every real stop/reset/output-off command and
+its order before a transport implementation exists.
+
+No USB enumeration, controller command, motion, framing or emission occurred.

docs/roadmap.md

@@ -293,6 +293,12 @@ state-before-motion order derived from pinned BalorMK. Port commands, upload,
 execute, status and cancellation remain outside the encoder and cannot be
 reached by this spike.
 
+The cancellation half of item 5 now has an asynchronous no-device proof:
+hold/resume work through the Rayforge driver contract, cancel clears every
+pending list buffer, restart fails until explicit recovery, and a separately
+encoded follow-up job can then complete. Physical stop/reset commands remain
+absent. Graphical workflow assessment is still open.
+
 ## Decision rules
 
 - A simulator test precedes every equivalent hardware behavior.

scripts/spike_rayforge_jcz_cancel_recovery.py

@@ -0,0 +1,130 @@
+"""Exercise Rayforge-style async cancellation on the offline JCZ driver."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import subprocess
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+
+RAYFORGE_COMMIT = "c28918c87f8273dd1d244f5a057addc192807fa5"
+
+
+def _commit_of(source: Path) -> str:
+    result = subprocess.run(
+        ["git", "-C", str(source), "rev-parse", "HEAD"],
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    return result.stdout.strip()
+
+
+def _ops(Ops, line_count: int):
+    ops = Ops()
+    ops.job_start()
+    ops.set_power(0.5)
+    ops.set_feed_rate(15_000)
+    ops.set_rapid_rate(120_000)
+    ops.set_frequency(40_000)
+    ops.set_pulse_width(200)
+    ops.move_to(50, 50, 0)
+    for index in range(line_count):
+        ops.line_to(51 + index % 2, 50 + index % 2, 0)
+    ops.job_end()
+    return ops
+
+
+async def _exercise(Driver, Ops) -> dict[str, object]:
+    machine = MagicMock()
+    machine.coordinate_systems = {}
+    driver = Driver(MagicMock(), machine)
+    encoder = driver.create_encoder(machine)
+
+    long_ops = _ops(Ops, 260)
+    encoded = encoder.encode(long_ops)
+    buffered_before_cancel = sum(
+        len(packet) for packet in encoded.driver_data["list_buffers"]
+    )
+    run_task = asyncio.create_task(driver.run(encoded, None, long_ops))
+    await asyncio.sleep(0)
+    await driver.set_hold(True)
+    held_state = driver.offline_session.state.value
+    await driver.set_hold(False)
+    resumed_state = driver.offline_session.state.value
+    await driver.cancel()
+    await asyncio.wait_for(run_task, timeout=2)
+
+    cancelled_state = driver.offline_session.state.value
+    buffered_after_cancel = driver.offline_session.buffered_bytes
+    restart_rejected = False
+    try:
+        driver.offline_session.start()
+    except RuntimeError:
+        restart_rejected = True
+
+    driver.offline_session.recover()
+    recovered_state = driver.offline_session.state.value
+
+    recovery_ops = _ops(Ops, 2)
+    recovery_encoded = encoder.encode(recovery_ops)
+    await asyncio.wait_for(
+        driver.run(recovery_encoded, None, recovery_ops), timeout=2
+    )
+
+    return {
+        "transport": None,
+        "long_job_buffer_count": len(encoded.driver_data["list_buffers"]),
+        "buffered_bytes_before_cancel": buffered_before_cancel,
+        "held_state": held_state,
+        "resumed_state": resumed_state,
+        "cancelled_state": cancelled_state,
+        "buffered_bytes_after_cancel": buffered_after_cancel,
+        "restart_without_recovery_rejected": restart_rejected,
+        "recovered_state": recovered_state,
+        "second_job_state": driver.offline_session.state.value,
+        "events": driver.offline_session.events,
+        "hardware_commands": [],
+    }
+
+
+def run(source: Path, addon: Path) -> dict[str, object]:
+    commit = _commit_of(source)
+    if commit != RAYFORGE_COMMIT:
+        raise RuntimeError(f"expected Rayforge {RAYFORGE_COMMIT}, found {commit}")
+    sys.path.insert(0, str(source))
+    sys.path.insert(0, str(addon))
+
+    from pirate_ray_jcz.worker import PirateRayJCZNoDeviceDriver
+    from raygeo.ops import Ops
+
+    result = asyncio.run(_exercise(PirateRayJCZNoDeviceDriver, Ops))
+    result["rayforge_commit"] = commit
+    return result
+
+
+def main() -> None:
+    root = Path(__file__).parents[1]
+    parser = argparse.ArgumentParser()
+    parser.add_argument("source", type=Path, help="pinned Rayforge checkout")
+    parser.add_argument(
+        "--addon",
+        type=Path,
+        default=root / "spikes" / "rayforge-jcz-no-device-addon",
+    )
+    parser.add_argument("--output", type=Path)
+    args = parser.parse_args()
+    result = run(args.source.resolve(), args.addon.resolve())
+    rendered = json.dumps(result, indent=2) + "\n"
+    if args.output:
+        args.output.write_text(rendered, encoding="utf-8", newline="\n")
+    else:
+        print(rendered, end="")
+
+
+if __name__ == "__main__":
+    main()

scripts/spike_rayforge_jcz_equivalence.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import argparse
-import importlib.util
+import importlib
 import json
 import subprocess
 import sys
@@ -24,12 +24,8 @@ def _commit_of(source: Path) -> str:
 
 
 def _load_encoder(addon: Path):
-    worker = addon / "pirate_ray_jcz" / "worker.py"
-    spec = importlib.util.spec_from_file_location("tpr_jcz_worker", worker)
-    if spec is None or spec.loader is None:
-        raise RuntimeError("could not load JCZ spike encoder")
-    module = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(module)
+    sys.path.insert(0, str(addon))
+    module = importlib.import_module("pirate_ray_jcz.worker")
     return module.OfflineJCZEncoder
 
 

scripts/spike_rayforge_jcz_negative_cases.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import argparse
-import importlib.util
+import importlib
 import json
 import subprocess
 import sys
@@ -24,12 +24,8 @@ def _commit_of(source: Path) -> str:
 
 
 def _load_encoder(addon: Path):
-    path = addon / "pirate_ray_jcz" / "worker.py"
-    spec = importlib.util.spec_from_file_location("tpr_jcz_negative", path)
-    if spec is None or spec.loader is None:
-        raise RuntimeError("could not load JCZ spike encoder")
-    module = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(module)
+    sys.path.insert(0, str(addon))
+    module = importlib.import_module("pirate_ray_jcz.worker")
     return module.OfflineJCZEncoder
 
 

spikes/rayforge-jcz-no-device-addon/pirate_ray_jcz/offline_session.py

@@ -0,0 +1,112 @@
+"""Transport-free JCZ job lifecycle used by the offline driver spike."""
+
+from __future__ import annotations
+
+from enum import Enum
+from typing import Any
+
+
+class OfflineSessionState(str, Enum):
+    IDLE = "idle"
+    PREPARED = "prepared"
+    RUNNING = "running"
+    HOLD = "hold"
+    CANCELLED = "cancelled"
+    COMPLETE = "complete"
+
+
+class OfflineSessionError(RuntimeError):
+    """Raised when an invalid lifecycle transition is requested."""
+
+
+class OfflineJCZSession:
+    """Keep encoded buffers in memory and require explicit recovery."""
+
+    def __init__(self) -> None:
+        self.state = OfflineSessionState.IDLE
+        self._buffers: list[bytes] = []
+        self.events: list[str] = []
+
+    @property
+    def buffered_bytes(self) -> int:
+        return sum(len(packet) for packet in self._buffers)
+
+    def prepare(self, encoded: Any) -> None:
+        if self.state not in (
+            OfflineSessionState.IDLE,
+            OfflineSessionState.COMPLETE,
+        ):
+            raise OfflineSessionError(
+                f"cannot prepare from state {self.state.value}"
+            )
+        data = encoded.driver_data
+        if data.get("transport") is not None:
+            raise OfflineSessionError("offline session rejects transport")
+        buffers = data.get("list_buffers")
+        if not buffers or any(len(packet) != 0xC00 for packet in buffers):
+            raise OfflineSessionError("valid 0xC00 list buffers are required")
+        self._buffers = list(buffers)
+        self.state = OfflineSessionState.PREPARED
+        self.events.append("prepared")
+
+    def start(self) -> None:
+        if self.state is not OfflineSessionState.PREPARED:
+            raise OfflineSessionError(
+                f"cannot start from state {self.state.value}"
+            )
+        self.state = OfflineSessionState.RUNNING
+        self.events.append("started-offline")
+
+    def hold(self) -> None:
+        if self.state is not OfflineSessionState.RUNNING:
+            raise OfflineSessionError(
+                f"cannot hold from state {self.state.value}"
+            )
+        self.state = OfflineSessionState.HOLD
+        self.events.append("held-offline")
+
+    def resume(self) -> None:
+        if self.state is not OfflineSessionState.HOLD:
+            raise OfflineSessionError(
+                f"cannot resume from state {self.state.value}"
+            )
+        self.state = OfflineSessionState.RUNNING
+        self.events.append("resumed-offline")
+
+    def cancel(self) -> None:
+        if self.state not in (
+            OfflineSessionState.PREPARED,
+            OfflineSessionState.RUNNING,
+            OfflineSessionState.HOLD,
+        ):
+            raise OfflineSessionError(
+                f"cannot cancel from state {self.state.value}"
+            )
+        self._buffers.clear()
+        self.state = OfflineSessionState.CANCELLED
+        self.events.extend(
+            (
+                "cancel-requested-offline",
+                "pending-buffers-cleared",
+                "recovery-required",
+            )
+        )
+
+    def recover(self) -> None:
+        if self.state is not OfflineSessionState.CANCELLED:
+            raise OfflineSessionError(
+                f"cannot recover from state {self.state.value}"
+            )
+        if self._buffers:
+            raise OfflineSessionError("recovery requires empty buffers")
+        self.state = OfflineSessionState.IDLE
+        self.events.append("recovered-to-idle")
+
+    def complete(self) -> None:
+        if self.state is not OfflineSessionState.RUNNING:
+            raise OfflineSessionError(
+                f"cannot complete from state {self.state.value}"
+            )
+        self._buffers.clear()
+        self.state = OfflineSessionState.COMPLETE
+        self.events.extend(("completed-offline", "pending-buffers-cleared"))

spikes/rayforge-jcz-no-device-addon/pirate_ray_jcz/worker.py

@@ -5,6 +5,7 @@ specification, transport, USB import, or machine I/O.
 """
 
 import hashlib
+import asyncio
 import json
 import math
 import struct
@@ -19,6 +20,11 @@ from rayforge.pipeline.encoder.base import (
     OpsEncoder,
 )
 
+from .offline_session import (
+    OfflineJCZSession,
+    OfflineSessionState,
+)
+
 
 class OfflineJCZEncoder(OpsEncoder):
     """Encode Rayforge Ops to a deterministic, transport-free JCZ trace."""
@@ -294,11 +300,45 @@ class PirateRayJCZNoDeviceDriver(NoDeviceDriver):
     uses_gcode = False
     DISCOVERY = None
 
+    def __init__(self, context, machine) -> None:
+        super().__init__(context, machine)
+        self.offline_session = OfflineJCZSession()
+
     @classmethod
     def create_encoder(cls, machine) -> OpsEncoder:
         del machine
         return OfflineJCZEncoder()
 
+    async def run(
+        self, encoded, doc, ops, on_command_done=None
+    ) -> None:
+        del doc, ops
+        self.offline_session.prepare(encoded)
+        self.offline_session.start()
+        op_count = encoded.op_map.op_count if encoded.op_map else 0
+        for op_index in range(op_count):
+            await asyncio.sleep(0)
+            while self.offline_session.state is OfflineSessionState.HOLD:
+                await asyncio.sleep(0)
+            if self.offline_session.state is OfflineSessionState.CANCELLED:
+                return
+            if on_command_done is not None:
+                result = on_command_done(op_index)
+                if hasattr(result, "__await__"):
+                    await result
+        self.offline_session.complete()
+        self.job_finished.send(self)
+
+    async def set_hold(self, hold: bool = True) -> None:
+        if hold:
+            self.offline_session.hold()
+        else:
+            self.offline_session.resume()
+
+    async def cancel(self, emergency: bool = False) -> None:
+        del emergency
+        self.offline_session.cancel()
+
 
 def _register() -> None:
     name = PirateRayJCZNoDeviceDriver.__name__

tests/fixtures/rayforge-addon-spike.json

@@ -17,6 +17,7 @@
   ],
   "packaged_files": [
     "pirate_ray_jcz/__init__.py",
+    "pirate_ray_jcz/offline_session.py",
     "pirate_ray_jcz/profile_import.py",
     "pirate_ray_jcz/worker.py",
     "rayforge-addon.yaml"

tests/fixtures/rayforge-jcz-cancel-recovery.json

@@ -0,0 +1,28 @@
+{
+  "transport": null,
+  "long_job_buffer_count": 2,
+  "buffered_bytes_before_cancel": 6144,
+  "held_state": "hold",
+  "resumed_state": "running",
+  "cancelled_state": "cancelled",
+  "buffered_bytes_after_cancel": 0,
+  "restart_without_recovery_rejected": true,
+  "recovered_state": "idle",
+  "second_job_state": "complete",
+  "events": [
+    "prepared",
+    "started-offline",
+    "held-offline",
+    "resumed-offline",
+    "cancel-requested-offline",
+    "pending-buffers-cleared",
+    "recovery-required",
+    "recovered-to-idle",
+    "prepared",
+    "started-offline",
+    "completed-offline",
+    "pending-buffers-cleared"
+  ],
+  "hardware_commands": [],
+  "rayforge_commit": "c28918c87f8273dd1d244f5a057addc192807fa5"
+}

tests/test_rayforge_addon_spike.py

@@ -12,6 +12,9 @@ EQUIVALENCE_FIXTURE = (
 NEGATIVE_FIXTURE = (
     ROOT / "tests" / "fixtures" / "rayforge-jcz-negative-cases.json"
 )
+CANCEL_FIXTURE = (
+    ROOT / "tests" / "fixtures" / "rayforge-jcz-cancel-recovery.json"
+)
 
 
 class RayforgeAddonSpikeTests(unittest.TestCase):
@@ -115,6 +118,20 @@ class RayforgeAddonSpikeTests(unittest.TestCase):
             {"rejected"}, {case["result"] for case in result["cases"]}
         )
 
+    def test_cancel_requires_recovery_and_clears_all_buffers(self):
+        result = json.loads(CANCEL_FIXTURE.read_text(encoding="utf-8"))
+        self.assertIsNone(result["transport"])
+        self.assertEqual([], result["hardware_commands"])
+        self.assertEqual(2, result["long_job_buffer_count"])
+        self.assertEqual(6144, result["buffered_bytes_before_cancel"])
+        self.assertEqual(0, result["buffered_bytes_after_cancel"])
+        self.assertEqual("hold", result["held_state"])
+        self.assertEqual("running", result["resumed_state"])
+        self.assertEqual("cancelled", result["cancelled_state"])
+        self.assertTrue(result["restart_without_recovery_rejected"])
+        self.assertEqual("idle", result["recovered_state"])
+        self.assertEqual("complete", result["second_job_state"])
+
 
 if __name__ == "__main__":
     unittest.main()