fix(coding-agent): dispatcher-owned host-reply envelope (#1836)
* fix host reply result envelope * docs(coding-agent): drop a stale handler-contract sentence from the merge
115 unmodified linesExpand all116117118119120120121122123124653 unmodified linesExpand all7787797807817827827837847847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538541221 unmodified linesExpand all115 unmodified linesExpand all self.ready_event, self.ready_ms = self.repl.ready()
def test_ready_handshake_and_startup_time(self): self.assertEqual(self.ready_event["event"], "ready") self.assertEqual(self.ready_event["protocol"], 2) self.assertEqual(self.ready_event["protocol"], 3) major, minor = sys.version_info[:2] self.assertTrue(self.ready_event["python"].startswith(f"{major}.{minor}.")) # Loose bound for loaded CI machines; still catches an order-of-magnitude regression. print(f"\n[startup] spawn -> ready: {self.ready_ms:.0f} ms")653 unmodified linesExpand all request = self.repl.read_event() while request.get("event") != "host_request": request = self.repl.read_event() self.assertEqual(request["data"], {"type": "demo", "value": 7}) envelope = {"status": "ok", "result": {"status": "weird", "answer": 42}} self.repl.send({"type": "host_reply", "id": request["id"], "data": {"status": "ok", "answer": 42}}) self.repl.send({"type": "host_reply", "id": request["id"], "data": envelope}) events = self.repl.until_done("hr") self.assertEqual(one(events, "result")["text"], "{'status': 'ok', 'answer': 42}") self.assertEqual(one(events, "result")["text"], repr(envelope)) self.assertEqual(one(events, "done")["status"], "ok")
def test_typed_host_request_unwraps_exact_handler_result(self): code = "\n".join( [ "from rlm import host_request", "reply = await host_request('demo')", "reply", ] ) self.repl.send({"type": "execute", "id": "typed-hr", "code": code}) request = self.repl.read_event() while request.get("event") != "host_request": request = self.repl.read_event() handler_result = {"status": "weird", "answer": 42} self.repl.send( { "type": "host_reply", "id": request["id"], "data": {"status": "ok", "result": handler_result}, } ) events = self.repl.until_done("typed-hr") self.assertEqual(one(events, "result")["text"], repr(handler_result)) self.assertEqual(one(events, "done")["status"], "ok")
def test_typed_host_request_raises_host_error(self): code = "from rlm import host_request\nawait host_request('demo')" self.repl.send({"type": "execute", "id": "typed-error", "code": code}) request = self.repl.read_event() while request.get("event") != "host_request": request = self.repl.read_event() self.repl.send( { "type": "host_reply", "id": request["id"], "data": {"status": "error", "error": "handler exploded"}, } ) events = self.repl.until_done("typed-error") self.assertEqual(one(events, "error")["ename"], "RuntimeError") self.assertEqual(one(events, "error")["evalue"], "handler exploded") self.assertEqual(one(events, "done")["status"], "error")
def test_typed_host_request_rejects_unexpected_envelope_status(self): code = "from rlm import host_request\nawait host_request('demo')" self.repl.send({"type": "execute", "id": "typed-unexpected", "code": code}) request = self.repl.read_event() while request.get("event") != "host_request": request = self.repl.read_event() self.repl.send( { "type": "host_reply", "id": request["id"], "data": {"status": "partial", "result": {}}, } ) events = self.repl.until_done("typed-unexpected") self.assertEqual(one(events, "error")["ename"], "RuntimeError") self.assertEqual( one(events, "error")["evalue"], "host request demo returned unexpected status: 'partial'", ) self.assertEqual(one(events, "done")["status"], "error")
def test_host_reply_for_unknown_id_dropped(self): self.repl.send({"type": "host_reply", "id": "no-such-request", "data": {"status": "ok"}}) events = self.repl.execute("ok", "'alive'") self.assertEqual(one(events, "result")["text"], "'alive'")1221 unmodified linesExpand all1- Namespaced kernel host handler results so handler fields cannot overwrite host reply protocol metadata.122 unmodified linesExpand all1231241251261271271281291301311 unmodified line133134135136137138139140137138139140143141142143122 unmodified linesExpand all expect(reroute.status).toBe("ok"); expect(reroute.stdout.trim()).toBe('RuntimeError: host request type "goal.get" is not available in this session'); });
it("rejects replies with an unexpected status instead of hanging", async () => { it("round-trips handler records with a status field", async () => { provisioner = new IpythonKernelProvisioner(tempDir, { pythonSkills: [bundledGoalSkill()], hostHandlers: { "goal.get": async () => ({ status: "partial" }),1 unmodified line });
const manager = await provisioner.ensure(); const result = await manager.execute(`try: await goal.get()except RuntimeError as error: print(f"RuntimeError: {error}")import jsonprint(json.dumps(await goal.get(), sort_keys=True))`); expect(result.status).toBe("ok"); expect(result.stdout.trim()).toBe("RuntimeError: host request goal.get returned unexpected status: 'partial'"); expect(JSON.parse(result.stdout.trim())).toEqual({ status: "partial" }); });});115 unmodified linesExpand all1161171181191201201211221231244 unmodified lines12913013113213313313413513613713813914014114214390 unmodified lines115 unmodified linesExpand all }, { "test.slow": async () => { await handlerBlocked; return { answer: 42 }; return { answer: 42, status: "weird" }; }, }, );
4 unmodified lines releaseHandler?.(); await dispose;
expect(internals.inFlightHostRequests.size).toBe(0); expect(sentReplies).toEqual([{ type: "host_reply", id: "hr-1", data: { status: "ok", answer: 42 } }]); expect(sentReplies).toEqual([ { type: "host_reply", id: "hr-1", data: { status: "ok", result: { answer: 42, status: "weird" } }, }, ]); expect(internals.child).toBeUndefined(); });
it("shutdown during an in-flight host request leaves no dangling task once the handler settles", async () => {90 unmodified lines47 unmodified lines4849505152525354555637 unmodified lines47 unmodified lines const errorSpy = vi.spyOn(console, "error").mockImplementation(() => {}); const manager = new ReplKernelManager({ python, cwd: tempDir });
try { await expect(manager.execute("print(1)")).rejects.toThrow(/speaks protocol 1, expected 2/); await expect(manager.execute("print(1)")).rejects.toThrow(/speaks protocol 1, expected 3/); } finally { errorSpy.mockRestore(); await manager.dispose(); }37 unmodified lines55 unmodified lines56575859606061626364235 unmodified linesExpand all55 unmodified lines
def _parse_host_reply(request_type: str, reply: dict[str, Any]) -> dict[str, Any]: status = reply.get("status") if status == "ok": return {k: v for k, v in reply.items() if k != "status"} return reply["result"] if status == "error": raise RuntimeError(str(reply.get("error") or f"host request {request_type} failed")) raise RuntimeError(f"host request {request_type} returned unexpected status: {status!r}")
235 unmodified linesExpand all1 unmodified line2345667891017 unmodified lines282930313232333435364 unmodified lines41424344454546474849119 unmodified linesExpand all1 unmodified line
`python -m rlm.repl` starts a CPython REPL runtime that executes code cells inone persistent `__main__` namespace on a single asyncio event loop. The wireformat is newline-delimited JSON: one object per line, UTF-8, no other framing.The current protocol version is `2`; the runtime announces it in the `ready`The current protocol version is `3`; the runtime announces it in the `ready`event.
## Channels
17 unmodified lines| Request | Fields ||---|---|| `execute` | `{"type":"execute","id":str,"code":str}` || `interrupt` | `{"type":"interrupt","id"?:str}` — no reply || `host_reply` | `{"type":"host_reply","id":str,"data":{...}}` — no reply || `host_reply` | `{"type":"host_reply","id":str,"data":{"status":"ok","result":{...}}}` or an error envelope — no reply || `snapshot` | `{"type":"snapshot","id":str,"path":str,"manifest_path":str,"max_bytes"?:int,"max_variable_bytes"?:int,"prune_oversized"?:bool}` || `restore` | `{"type":"restore","id":str,"path":str}` || `list_names` | `{"type":"list_names","id":str}` || `shutdown` | `{"type":"shutdown","id"?:str}` |4 unmodified linesruntime keeps serving. Closing stdin is equivalent to `shutdown`.
## Events
- `{"event":"ready","protocol":2,"python":"3.13.11"}` — sent once at startup;- `{"event":"ready","protocol":3,"python":"3.13.11"}` — sent once at startup; the handshake. No banner precedes it.- `{"event":"stdout"|"stderr","id":str|null,"text":str}` — captured output. `id` is the cell whose Python execution context performed the write; asyncio tasks inherit the spawning cell's id (even after that cell finished). `null`119 unmodified linesExpand all29 unmodified lines303132333434353637381127 unmodified linesExpand all29 unmodified linesfrom typing import Any
from .bash import _kill_live_handles
PROTOCOL_VERSION = 2PROTOCOL_VERSION = 3
DEFAULT_SNAPSHOT_MAX_BYTES = 256 * 1024 * 1024DEFAULT_SNAPSHOT_MAX_VARIABLE_BYTES = 16 * 1024 * 1024
1127 unmodified linesExpand all