From 892a06b989dc1634a0576e74e3245634a6f4410b Mon Sep 17 00:00:00 2001 From: Adarsh Prashar Date: Fri, 5 Jun 2026 18:52:14 +0530 Subject: [PATCH] feat(sdk): resume an existing run after a crash MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The SDK could open a governed run (governed_run) but had no way to ATTACH to an existing one — so the crash-resume story, which the daemon fully supports server-side (it reloads non-terminal runs on restart with their already-spent budget), wasn't reachable from Python without dropping to the low-level client. Add Runtime.resume_run(run_id): a context manager that attaches to an existing run by id. Unlike governed_run it neither creates a new run nor cancels on error — the run keeps its server-side budget and spent usage, so a resumed agent can fetch its last checkpoint and continue where it left off, and can't overspend by restarting. Verified the server side end-to-end first: a run advanced 5 of 8 loop steps, then a SIGKILL of the daemon; on restart it reloaded the run (loops=5 + the checkpoint) and continued — allowing exactly 3 more steps before halting, proving no re-spend. Tests: a resume test (attach by id, current_run set, read the checkpoint, keep stepping the same run) plus the stub daemon now serves GET /v1/runs/{id}. Documented in the SDK README and CHANGELOG. --- CHANGELOG.md | 7 +++++++ sdks/python/README.md | 19 +++++++++++++++++++ sdks/python/riskkernel/runtime.py | 27 +++++++++++++++++++++++++++ sdks/python/tests/test_sdk.py | 19 +++++++++++++++++++ 4 files changed, 72 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index d136cbc..00bbc39 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,13 @@ surface is governed by [`COMPATIBILITY.md`](COMPATIBILITY.md). ## [Unreleased] +### Added +- **SDK: resume a run after a crash.** `Runtime.resume_run(run_id)` attaches to an + existing governed run (it neither creates a new run nor cancels on error), so a + Python agent can pick its work back up from the last checkpoint after a `SIGKILL`. + The run keeps its server-side budget and already-spent usage, so it can't + overspend by restarting. See the [SDK README](sdks/python/README.md#resume-after-a-crash). + ## [0.2.0] - 2026-06-04 A frictionless-adoption release: a one-line CLI install, three runnable key-free diff --git a/sdks/python/README.md b/sdks/python/README.md index eea4ede..8bc57a7 100644 --- a/sdks/python/README.md +++ b/sdks/python/README.md @@ -36,6 +36,25 @@ with rt.governed_run(name="research", When the governor halts the run (token / dollar / loop / time budget), the next `run.step()` — or a proxied model call — raises `rk.BudgetExceeded`. +## Resume after a crash + +The daemon reloads non-terminal runs on restart with the budget and usage they had +already spent, so a `SIGKILL`'d run keeps enforcing without re-spending. Reattach to +it by id with `resume_run` and pick your work back up from the last checkpoint: + +```python +with rt.resume_run(run_id) as run: # attaches; never creates or cancels + cp = run.latest_checkpoint() # the state you saved before the crash + start = cp["payload"]["cursor"] if cp else 0 + for i in range(start, total): # skip the steps you already paid for + run.step() # counts against the SAME budget + # ... your work ... + run.checkpoint("step", {"cursor": i + 1}) +``` + +The run resumes against whatever budget it had left, so it can't overspend by +restarting — `run.step()` still raises `rk.BudgetExceeded` at the original ceiling. + ## Human-in-the-loop tools Gate side-effecting tools on human approval (the daemon's policy decides what needs diff --git a/sdks/python/riskkernel/runtime.py b/sdks/python/riskkernel/runtime.py index d8dd856..b32d303 100644 --- a/sdks/python/riskkernel/runtime.py +++ b/sdks/python/riskkernel/runtime.py @@ -157,6 +157,33 @@ def governed_run(self, name: Optional[str] = None, finally: _current_run.reset(token) + @contextmanager + def resume_run(self, run_id: str): + """Attach to an existing governed run by id — the resume path after a crash. + + Unlike governed_run, this neither creates nor cancels the run: the daemon + reloads non-terminal runs on restart with the budget and usage they had + already spent, so enforcement continues without re-spending. Fetch + ``run.latest_checkpoint()`` to pick your work back up where it left off:: + + with rt.resume_run(run_id) as run: + cp = run.latest_checkpoint() + start = cp["payload"]["cursor"] if cp else 0 + for i in range(start, total): + run.step() # counts against the SAME budget + ... + run.checkpoint("step", {"cursor": i + 1}) + + Raises APIError(404) if the run id is unknown. + """ + data = self.client.get_run(run_id) + run = Run(self.client, data, self._poll, self._timeout) + token = _current_run.set(run) + try: + yield run + finally: + _current_run.reset(token) + # Module-level default runtime, configured from the environment, for the # decorator/convenience API. diff --git a/sdks/python/tests/test_sdk.py b/sdks/python/tests/test_sdk.py index f3d6a9d..2948f6f 100644 --- a/sdks/python/tests/test_sdk.py +++ b/sdks/python/tests/test_sdk.py @@ -58,6 +58,9 @@ def _read(self): def do_GET(self): p = self.path + if p == "/v1/runs/run-1": + return self._send(200, {"id": "run-1", "name": "t", "status": "running", + "usage": {"tokens": 0, "loops": STATE.steps}}) if p.startswith("/v1/memory/facts"): return self._send(200, [{"namespace": "dev", "key": "db", "value": "sqlite"}]) if p.startswith("/v1/memory/entry"): @@ -133,6 +136,22 @@ def test_governed_run_and_step_budget(self): run.step() self.assertEqual(cm.exception.reason, "loop_budget_exceeded") + def test_resume_run_attaches_without_creating(self): + # The post-crash path: resume an existing run by id. resume_run attaches a + # handle (no new run, no cancel) that keeps stepping/checkpointing against + # the SAME run, and reads the checkpoint it left off at. + with self.rt.governed_run(name="t", budget=self.rt.budget(loops=5)) as run: + run.step() # one step before the "crash" + run.checkpoint("before-crash", {"cursor": 1}) + rid = run.id + + with self.rt.resume_run(rid) as resumed: + self.assertEqual(resumed.id, rid) # same run, not a new one + self.assertEqual(rk.current_run().id, rid) # set as the current run + cp = resumed.latest_checkpoint() + self.assertEqual(cp["payload"]["cursor"], 1) # where we left off + self.assertEqual(resumed.step(), 2) # continues the same step count + def test_checkpoint_roundtrip(self): with self.rt.governed_run(name="t") as run: run.checkpoint("after", {"cursor": 7})