1 files changed, 48 insertions, 0 deletions
diff --git a/packages/meshbay-cli/src/meshbay_cli/client/pool.py b/packages/meshbay-cli/src/meshbay_cli/client/pool.py
new file mode 100644
index 0000000..00e69fb
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/client/pool.py
@@ -0,0 +1,48 @@
+"""
+One connection per group for the whole run, and a ceiling on how many are being
+negotiated at once — the application's `ConnectionPool` rules
+(docs/MESHBAY_DESIGN.md §9.11).
+
+The ceiling is on concurrency, never a batch: a group whose node does not
+answer holds one place for its own deadline and delays nobody else. Six stays
+far below what the hub admits per account (32 pending offers), even with the
+person's application and a browser open at the same time.
+"""
+
+import asyncio
+import contextlib
+
+from .account import Account
+from .groups import open_group
+from .session import GroupSession
+
+MAX_IN_FLIGHT = 6
+
+
+class Pool:
+    def __init__(self, account: Account, max_in_flight: int = MAX_IN_FLIGHT):
+        self.account = account
+        self._gate = asyncio.Semaphore(max_in_flight)
+        self._sessions: dict[str, asyncio.Task] = {}
+
+    async def session(self, group: dict) -> GroupSession:
+        """The group's session, negotiated once and shared by every caller."""
+        task = self._sessions.get(group["id"])
+        if task is None:
+            task = asyncio.ensure_future(self._open(group))
+            self._sessions[group["id"]] = task
+        return await asyncio.shield(task)
+
+    async def _open(self, group: dict) -> GroupSession:
+        async with self._gate:
+            return await open_group(self.account, group)
+
+    async def close(self) -> None:
+        for task in self._sessions.values():
+            if task.done() and not task.cancelled() and task.exception() is None:
+                await task.result().close()
+            else:
+                task.cancel()
+                with contextlib.suppress(BaseException):
+                    await task
+        self._sessions.clear()