Use running loop for asyncio futures

Signed-off-by: Miro <200482516+Mirochill@users.noreply.github.com>

Author: Mirochill

src/arcp/_client/handles.py

@@ -25,7 +25,7 @@ def __init__(self, job_id: str, accepted: JobAcceptedPayload) -> None:
         self.accepted = accepted
         self._events: asyncio.Queue[dict[str, Any] | None] = asyncio.Queue()
         self._chunks: asyncio.Queue[dict[str, Any] | None] = asyncio.Queue()
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         self._terminal: asyncio.Future[JobResultPayload] = loop.create_future()
 
     @property

src/arcp/_client/ops.py

@@ -69,7 +69,7 @@ async def submit_job(  # noqa: PLR0913
         trace_id=trace_id,
         payload=submit.model_dump(mode="json", exclude_none=True),
     )
-    accept_fut: asyncio.Future[JobAcceptedPayload] = asyncio.get_event_loop().create_future()
+    accept_fut: asyncio.Future[JobAcceptedPayload] = asyncio.get_running_loop().create_future()
     client._pending_accepts.append(accept_fut)
     await client._transport.send(env.to_wire())
     try:

src/arcp/_runtime/_accept.py

@@ -57,7 +57,7 @@ async def _accept_or_close(runtime: ARCPRuntime, transport: Transport) -> Sessio
 def _maybe_start_heartbeat(ctx: SessionContext) -> asyncio.Task[Any] | None:
     if not (ctx.has_feature("heartbeat") and ctx.state.heartbeat_interval_sec):
         return None
-    ctx.heartbeat_outcome = asyncio.get_event_loop().create_future()
+    ctx.heartbeat_outcome = asyncio.get_running_loop().create_future()
     return asyncio.create_task(
         heartbeat_loop(ctx, interval=float(ctx.state.heartbeat_interval_sec))
     )

tests/e2e/test_e2e_memory.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import warnings
 
 from arcp import ClientInfo, RuntimeInfo, pair_memory_transports
 from arcp.client import ARCPClient
@@ -70,3 +71,32 @@ async def echo(input_value, ctx: JobContext):
         await client.close()
         await rt.close()
         server.cancel()
+
+
+async def test_submit_does_not_emit_deprecation_warning() -> None:
+    rt = ARCPRuntime(
+        runtime=RuntimeInfo(name="r", version="1"),
+        bearer=StaticBearerVerifier({"tok": "p1"}),
+        heartbeat_interval_sec=60.0,
+    )
+
+    async def echo(input_value, ctx: JobContext):
+        return input_value
+
+    rt.register_agent("echo", echo)
+
+    a, b = pair_memory_transports()
+    server = asyncio.create_task(rt.accept(a))
+    client = ARCPClient(client=ClientInfo(name="c", version="1"), token="tok")
+    try:
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", DeprecationWarning)
+            await client.connect(b)
+            handle = await client.submit(agent="echo", input={"x": 1})
+            result = await handle.done
+        assert result.final_status == "success"
+        assert result.result == {"x": 1}
+    finally:
+        await client.close()
+        await rt.close()
+        server.cancel()