prime-agent-runtime: create bash completion notices on the serving loop from executor threads
795 unmodified linesExpand all7967977987998008018028038048058068078088098108118128138148158168178188198201888 unmodified linesExpand all795 unmodified linesExpand all reply_ok(self.repl, request) inspected = self.repl.execute("bash-inspect", "handle.poll().text()") self.assertIn("detached", one(inspected, "result")["text"])
def test_executor_thread_handle_notifies_on_the_serving_loop(self): # asyncio.to_thread copies the cell context but has no running loop of its own; # the notice task still belongs to the serving loop. events = self.repl.execute( "off-loop-bash", "from rlm import bash\nimport asyncio\nbox = {}\n" "def worker():\n box['handle'] = bash('printf off-loop')\n" "await asyncio.to_thread(worker)\nbox['handle'].pid", ) request = wait_for_host_request(self.repl, events) self.assertEqual(int(one(events, "result")["text"]), request["data"]["pid"]) self.assertEqual( (request["data"]["type"], request["data"]["command"], request["data"]["exitCode"]), ("bash.completed", "printf off-loop", 0), ) reply_ok(self.repl, request)
def test_result_read_after_notice_requests_one_withdrawal(self): reads = ( ("poll", "handle.poll().text()"), ("await", "(await handle).text()"),1888 unmodified linesExpand all175 unmodified linesExpand all1761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062071031 unmodified linesExpand all12391240124112421243124412451220124612471248124912501251125212531254122312241255125612571258125912601261126212631264126512661267126812691270127112721644 unmodified linesExpand all175 unmodified linesExpand all if not task.cancelled(): task.exception()
def _start_background_completion_task( handle: BashHandle, loop: asyncio.AbstractEventLoop, notice: Any, activity: dict[str, Any],) -> None: """Create a bash completion notice task on the loop's own thread.
A handle may be started from an executor thread, which has no loop to create a task on; this runs there instead, and mirrors the inline construction path's rollback when the task cannot be created. """ from . import repl
try: task = loop.create_task(notice) except BaseException: handle.kill(signal.SIGKILL if _IS_POSIX else signal.SIGTERM) notice.close() repl.emit({"application/vnd.prime-agent.bash-activity+json": {**activity, "active": False}}) raise task.add_done_callback(_consume_notice_task)
def _completion_reaches( start: asyncio.Future[Any], targets: tuple[asyncio.Future[Any], ...]) -> bool: """Follow asyncio's wrapper and TaskGroup ownership callbacks."""1031 unmodified linesExpand all def _schedule_background_completion_notice(self) -> None: cell_finished = self._creating_cell_finished if cell_finished is None: return from . import repl
try: loop = asyncio.get_running_loop() running = asyncio.get_running_loop() except RuntimeError: running = None # A creating thread without a loop of its own (an executor worker) still owns # this notice: the task belongs on the serving loop, which owns the cell. serving = getattr(repl, "serving_loop", None) loop = running if running is not None else (serving() if serving is not None else None) if loop is None: return from . import repl
activity = {"id": secrets.token_hex(16), "pid": self._pid, "active": True} # Publish synchronously before bash() returns and the creating cell can end. repl.emit({"application/vnd.prime-agent.bash-activity+json": activity}) notice = self._notify_background_completion(cell_finished, activity) if running is None: try: # Only the loop's own thread may create a task on it. loop.call_soon_threadsafe(_start_background_completion_task, self, loop, notice, activity) except RuntimeError: # Loop already closed: the notice is undeliverable, but the activity # event that was already published must not look permanently active. notice.close() repl.emit({"application/vnd.prime-agent.bash-activity+json": {**activity, "active": False}}) return try: task = loop.create_task(notice) except BaseException: self.kill(signal.SIGKILL if _IS_POSIX else signal.SIGTERM)1644 unmodified linesExpand all