Enhance PVM runtime with continuation support and new features

- Added support for continuation modes (FSM and checkpoint) in the PVM runtime, allowing for more flexible execution control. - Introduced new methods for sending messages, scheduling timers, and canceling timers in the `FsHost` struct. - Updated `HostContext` to include additional fields for actor address, message ID, and nonce. - Enhanced the `ExecutionOptions` struct to accommodate continuation options and updated related functions for better configurability. - Refactored demo scripts to showcase new continuation features and provide clearer usage instructions. - Improved error handling and state management during checkpoint operations, ensuring robust execution flow.
RustPython · yusufyian · Dec 18, 2025 · Dec 18, 2025 · Dec 19, 2025 · Dec 19, 2025
commit 9f3c462d667722ee0895dec6d9c04e98c965d296
diff --git a/.gitignore b/.gitignore
@@ -37,4 +37,4 @@ examples/breakpoint_resume_demo/demo.rpsnap
 examples/breakpoint_resume_demo/actor_complex_demo.rpsnap
 examples/breakpoint_resume_demo/comprehensive_demo.rpsnap
 tmp/
-
+.DS_Store
diff --git a/Lib/pvm_sdk/__init__.py b/Lib/pvm_sdk/__init__.py
@@ -1,5 +1,24 @@
 from . import pvm_random
 from . import pvm_sys
 from . import pvm_time
+from . import runtime
+from . import continuation
+from . import runner
+from . import actor
+from . import verify
+from . import types
 
-__all__ = ["pvm_random", "pvm_sys", "pvm_time"]
+capture = continuation.capture
+
+__all__ = [
+    "pvm_random",
+    "pvm_sys",
+    "pvm_time",
+    "runtime",
+    "continuation",
+    "runner",
+    "actor",
+    "verify",
+    "types",
+    "capture",
+]
diff --git a/Lib/pvm_sdk/actor.py b/Lib/pvm_sdk/actor.py
@@ -0,0 +1,28 @@
+from . import runtime
+
+
+def continuation(*_args, **_kwargs):
+    def decorator(func):
+        return func
+    return decorator
+
+
+class ActorRef:
+    def __init__(self, address):
+        self.address = address
+
+    def async_call(self, method, *args, **kwargs):
+        if runtime.mode() != "checkpoint":
+            raise RuntimeError("actor async is only supported in checkpoint mode without FSM")
+        return _ActorAwaitable(self.address, method, *args, **kwargs)
+
+
+class _ActorAwaitable:
+    def __init__(self, address, method, *args, **kwargs):
+        self.address = address
+        self.method = method
+        self.args = args
+        self.kwargs = kwargs
+
+    def __await__(self):
+        raise RuntimeError("actor await not implemented")
diff --git a/Lib/pvm_sdk/continuation.py b/Lib/pvm_sdk/continuation.py
@@ -0,0 +1,158 @@
+import hashlib
+import json
+
+import pvm_host
+
+
+def _encode_value(value):
+    if isinstance(value, bytes):
+        return {"__bytes__": value.hex()}
+    if isinstance(value, bytearray):
+        return {"__bytes__": bytes(value).hex()}
+    if isinstance(value, dict):
+        return {str(k): _encode_value(v) for k, v in value.items()}
+    if isinstance(value, list):
+        return [_encode_value(v) for v in value]
+    if value is None or isinstance(value, (bool, int, str)):
+        return value
+    raise TypeError("unsupported capture value type")
+
+
+def _decode_value(value):
+    if isinstance(value, dict) and "__bytes__" in value:
+        return bytes.fromhex(value["__bytes__"])
+    if isinstance(value, dict):
+        return {k: _decode_value(v) for k, v in value.items()}
+    if isinstance(value, list):
+        return [_decode_value(v) for v in value]
+    return value
+
+
+def _encode_json(value):
+    try:
+        return json.dumps(
+            value,
+            sort_keys=True,
+            separators=(",", ":"),
+            ensure_ascii=True,
+        ).encode("ascii")
+    except AttributeError as exc:
+        if "check_circular" not in str(exc):
+            raise
+        try:
+            import importlib
+            import json as _json
+            importlib.reload(_json)
+            return _json.dumps(
+                value,
+                sort_keys=True,
+                separators=(",", ":"),
+                ensure_ascii=True,
+            ).encode("ascii")
+        except Exception:
+            raise exc
+
+
+def _decode_json(data):
+    try:
+        return json.loads(data.decode("utf-8"))
+    except TypeError as exc:
+        if "Pattern" not in str(exc):
+            raise
+        try:
+            import importlib
+            import re as _re
+            import json as _json
+            importlib.reload(_re)
+            importlib.reload(_json)
+            return _json.loads(data.decode("utf-8"))
+        except Exception:
+            raise exc
+
+
+class Capture:
+    def __init__(self):
+        object.__setattr__(self, "_data", {})
+
+    def __getattr__(self, name):
+        data = object.__getattribute__(self, "_data")
+        if name in data:
+            return data[name]
+        raise AttributeError(name)
+
+    def __setattr__(self, name, value):
+        data = object.__getattribute__(self, "_data")
+        data[name] = value
+
+    def to_dict(self):
+        return dict(self._data)
+
+    @classmethod
+    def from_dict(cls, value):
+        inst = cls()
+        for k, v in value.items():
+            inst._data[k] = v
+        return inst
+
+
+def capture():
+    return Capture()
+
+
+def new_cid(self_obj, name):
+    ctx = pvm_host.context()
+    seed = b""
+    tx_hash = ctx.get("tx_hash")
+    if isinstance(tx_hash, (bytes, bytearray)):
+        seed += bytes(tx_hash)
+    sender = ctx.get("sender")
+    if isinstance(sender, (bytes, bytearray)):
+        seed += bytes(sender)
+    seed += str(name).encode("utf-8")
+    return hashlib.sha256(seed).digest()
+
+
+def _cont_key(cid):
+    return b"__continuation:" + cid
+
+
+def save_cont(cid, state, ctx, handler, timeout_blocks=0, guard_unchanged=None):
+    if isinstance(ctx, Capture):
+        ctx_dict = ctx.to_dict()
+    else:
+        ctx_dict = dict(ctx)
+    guard_value = guard_unchanged
+    if isinstance(guard_value, Capture):
+        guard_value = guard_value.to_dict()
+    payload = {
+        "state": int(state),
+        "ctx": _encode_value(ctx_dict),
+        "handler": str(handler),
+        "timeout_blocks": int(timeout_blocks),
+        "guard_unchanged": _encode_value(guard_value),
+    }
+    pvm_host.set_state(_cont_key(cid), _encode_json(payload))
+
+
+def load_cont(cid):
+    raw = pvm_host.get_state(_cont_key(cid))
+    if raw is None:
+        raise RuntimeError("continuation state missing")
+    data = _decode_json(raw)
+    ctx = _decode_value(data.get("ctx") or {})
+    data["ctx"] = Capture.from_dict(ctx)
+    if "guard_unchanged" in data:
+        data["guard_unchanged"] = _decode_value(data["guard_unchanged"])
+    return data
+
+
+def delete_cont(cid):
+    pvm_host.delete_state(_cont_key(cid))
+
+
+def encode_payload(value):
+    return _encode_json(_encode_value(value))
+
+
+def decode_payload(data):
+    return _decode_value(_decode_json(data))
diff --git a/Lib/pvm_sdk/runner.py b/Lib/pvm_sdk/runner.py
@@ -0,0 +1,82 @@
+import pvm_host
+from . import continuation as _continuation
+from . import runtime
+
+try:
+    import rustpython_checkpoint as _checkpoint
+except Exception:
+    _checkpoint = None
+
+
+RUNNER_ADDRESS = b"__runner__"
+
+
+def continuation(*_args, **_kwargs):
+    def decorator(func):
+        return func
+    return decorator
+
+
+def _send_job(job_type, cid, reply_handler, *args, **kwargs):
+    payload = {
+        "kind": "runner_job",
+        "job_type": job_type,
+        "payload": {
+            "args": list(args),
+            "kwargs": kwargs,
+        },
+        "cid": cid,
+        "reply_handler": reply_handler,
+    }
+    pvm_host.send_message(RUNNER_ADDRESS, _continuation.encode_payload(payload))
+
+
+def _result_key(cid):
+    return b"__runner_result:" + cid
+
+
+def _try_get_result(cid):
+    raw = pvm_host.get_state(_result_key(cid))
+    if raw is None:
+        return None
+    pvm_host.delete_state(_result_key(cid))
+    return _continuation.decode_payload(raw)
+
+
+class _RunnerAwaitable:
+    def __init__(self, job_type, *args, **kwargs):
+        self.job_type = job_type
+        self.args = args
+        self.kwargs = kwargs
+        self.cid = _continuation.new_cid(None, job_type)
+
+    def __await__(self):
+        if runtime.mode() != "checkpoint":
+            raise RuntimeError("runner await is only supported in checkpoint mode without FSM")
+        if False:
+            yield None
+        try_get_result = _try_get_result
+        send_job = _send_job
+        checkpoint = _checkpoint
+        while True:
+            result = try_get_result(self.cid)
+            if result is not None:
+                return result
+            send_job(self.job_type, self.cid, "", *self.args, **self.kwargs)
+            if checkpoint is None:
+                raise RuntimeError("checkpoint support missing")
+            checkpoint.checkpoint_bytes()
+
+
+def _request_checkpoint():
+    if _checkpoint is None:
+        raise RuntimeError("checkpoint support missing")
+    _checkpoint.checkpoint_bytes()
+
+
+def llm(*args, **kwargs):
+    return _RunnerAwaitable("llm", *args, **kwargs)
+
+
+def http(*args, **kwargs):
+    return _RunnerAwaitable("http", *args, **kwargs)
diff --git a/Lib/pvm_sdk/runtime.py b/Lib/pvm_sdk/runtime.py
@@ -0,0 +1,6 @@
+import pvm_host
+
+
+def mode():
+    cfg = pvm_host.runtime_config()
+    return cfg.get("continuation_mode", "fsm")
diff --git a/Lib/pvm_sdk/types.py b/Lib/pvm_sdk/types.py
@@ -0,0 +1,3 @@
+class SoftFloat(str):
+    def __new__(cls, value):
+        return str.__new__(cls, str(value))
diff --git a/Lib/pvm_sdk/verify.py b/Lib/pvm_sdk/verify.py
@@ -0,0 +1,45 @@
+class VerifyBuilder:
+    def __init__(self):
+        self._data = {
+            "mode": "none",
+            "runners": 1,
+            "threshold": 1,
+            "checks": [],
+        }
+
+    def mode(self, value):
+        self._data["mode"] = value
+        return self
+
+    def runners(self, value):
+        self._data["runners"] = int(value)
+        return self
+
+    def threshold(self, value):
+        self._data["threshold"] = int(value)
+        return self
+
+    def check(self, value):
+        self._data["checks"].append(value)
+        return self
+
+    def build(self):
+        return dict(self._data)
+
+
+class Verify:
+    @staticmethod
+    def builder():
+        return VerifyBuilder()
+
+    @staticmethod
+    def json_schema_valid(schema):
+        return {"kind": "json_schema_valid", "schema": schema}
+
+    @staticmethod
+    def structured_match(fields):
+        return {"kind": "structured_match", "fields": list(fields)}
+
+    @staticmethod
+    def majority_vote(field):
+        return {"kind": "majority_vote", "field": field}
diff --git a/crates/codegen/Cargo.toml b/crates/codegen/Cargo.toml
@@ -13,6 +13,7 @@ rustpython-compiler-core = { workspace = true }
 rustpython-literal = {workspace = true }
 rustpython-wtf8 = { workspace = true }
 ruff_python_ast = { workspace = true }
+ruff_python_parser = { workspace = true }
 ruff_text_size = { workspace = true }
 
 ahash = { workspace = true }
@@ -28,7 +29,6 @@ memchr = { workspace = true }
 unicode_names2 = { workspace = true }
 
 [dev-dependencies]
-ruff_python_parser = { workspace = true }
 insta = { workspace = true }
 
 [lints]