feat: update README and refactor module imports for worker pool

- Added documentation for the multi-process pool feature in the README, detailing usage with Rust workers and local pool management.
- Refactored import functions in main.py to load worker pool submodules from the `devsper.pool` namespace, improving clarity and organization.
- Updated Redis URL resolution and pool configuration loading to align with the new import structure.

Author: rithulkamesh

Dockerfile.local-pool

@@ -0,0 +1,22 @@
+# Local pool supervisor: runs `devsper.pool.local_pool`, which spawns N `devsper-worker` processes.
+# Build: docker build -f Dockerfile.local-pool -t devsper-local-pool .
+# Context must be the `runtime/` directory.
+
+FROM rust:1.82-slim AS rust-builder
+RUN apt-get update && apt-get install -y --no-install-recommends pkg-config libssl-dev \
+    && rm -rf /var/lib/apt/lists/*
+WORKDIR /build
+COPY worker/Cargo.toml worker/Cargo.lock ./
+COPY worker/src ./src
+RUN cargo build --release --features subprocess-executor
+
+FROM python:3.12-slim
+RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates libssl3 curl \
+    && rm -rf /var/lib/apt/lists/*
+COPY --from=rust-builder /build/target/release/devsper-worker /usr/local/bin/
+WORKDIR /app
+COPY pyproject.toml README.md LICENSE ./
+COPY devsper ./devsper
+RUN pip install --no-cache-dir -e ".[distributed]"
+ENV DEVSPER_PROFILE=local
+CMD ["python", "-m", "devsper.pool.local_pool", "--workers", "2"]

README.md

@@ -271,6 +271,8 @@ uv run python examples/distributed/run_controller.py "Your task" --parallel
 
 Rust workers: set `DEVSPER_WORKER_MODEL=github:gpt-4o` (or your model), `DEVSPER_PYTHON_BIN=.venv/bin/python`, `DEVSPER_RPC_PORT=0` for multiple workers on one host. Credentials load from keychain in the subprocess.
 
+**Multi-process pool (Rust default):** with `devsper[distributed]` installed, `python -m devsper.pool.local_pool --workers N` supervises *N* `devsper-worker` processes against Redis (`REDIS_URL`). Override the worker command in `devsper/pool/profiles/local.toml` (`local_worker_cmd`) if you need the Python `devsper.pool.worker_runner` for debugging. The monorepo builds a combined image via `runtime/Dockerfile.local-pool` (used as `pool-manager` in the root `docker-compose.yml`).
+
 ---
 
 ## Examples

devsper/cli/main.py

@@ -40,16 +40,9 @@ def _project_root() -> Path:
     return Path(__file__).resolve().parent.parent.parent
 
 
-def _import_platform_pool(submodule: str):
-    """Import ``platform.pool.<submodule>`` with repo root on sys.path; avoid stdlib ``platform``."""
-    root = _project_root().parent
-    if str(root) not in sys.path:
-        sys.path.insert(0, str(root))
-    if "platform" in sys.modules:
-        m = sys.modules["platform"]
-        if not getattr(m, "__path__", None):
-            del sys.modules["platform"]
-    return importlib.import_module(f"platform.pool.{submodule}")
+def _import_pool_submodule(submodule: str):
+    """Import ``devsper.pool.<submodule>`` (distributed worker pool; lives in the runtime package)."""
+    return importlib.import_module(f"devsper.pool.{submodule}")
 
 
 def _resolve_pool_redis_url(cli_override: str | None = None, profile: str | None = None) -> str:
@@ -61,7 +54,7 @@ def _resolve_pool_redis_url(cli_override: str | None = None, profile: str | None
     prof = (profile or os.environ.get("DEVSPER_PROFILE") or "").strip().lower()
     if prof == "local":
         try:
-            config_mod = _import_platform_pool("config")
+            config_mod = _import_pool_submodule("config")
             return config_mod.load_pool_config("local").redis_url
         except Exception:
             return "redis://127.0.0.1:6379"
@@ -544,12 +537,12 @@ def _run_swarm_via_local_pool(args: object) -> int:
     user_id = os.environ.get("DEVSPER_USER_ID", "local-user")
     redis_url = _resolve_pool_redis_url(profile="local")
 
-    pool_cfg = _import_platform_pool("config").load_pool_config("local")
+    pool_cfg = _import_pool_submodule("config").load_pool_config("local")
 
-    crypto_mod = _import_platform_pool("crypto")
-    manager_mod = _import_platform_pool("manager")
-    models_mod = _import_platform_pool("models")
-    store_mod = _import_platform_pool("store")
+    crypto_mod = _import_pool_submodule("crypto")
+    manager_mod = _import_pool_submodule("manager")
+    models_mod = _import_pool_submodule("models")
+    store_mod = _import_pool_submodule("store")
     encrypt_payload = crypto_mod.encrypt_payload
     generate_org_keypair = crypto_mod.generate_org_keypair
     PoolManager = manager_mod.PoolManager
@@ -1704,14 +1697,14 @@ def _run_pool_start(args) -> int:
     env["PYTHONPATH"] = root + os.pathsep + env.get("PYTHONPATH", "")
     if not env.get("REDIS_URL"):
         env["REDIS_URL"] = _resolve_pool_redis_url(profile="local")
-    cmd = [sys.executable, "-m", "platform.pool.local_pool", "--workers", str(workers)]
+    cmd = [sys.executable, "-m", "devsper.pool.local_pool", "--workers", str(workers)]
     return subprocess.call(cmd, env=env)
 
 
 def _run_org_keygen(args) -> int:
     """Generate org E2EE keypair and store private key in keyring."""
     try:
-        crypto_mod = _import_platform_pool("crypto")
+        crypto_mod = _import_pool_submodule("crypto")
         generate_org_keypair = crypto_mod.generate_org_keypair
         from devsper.credentials.store import CredentialStore
     except Exception as e:

devsper/pool/__init__.py

@@ -0,0 +1,2 @@
+"""Worker pool manager (platform-side)."""
+

devsper/pool/config.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import os
+from pathlib import Path
+import tomllib
+
+
+@dataclass
+class PoolConfig:
+    profile: str = "prod"
+    redis_url: str = "redis://localhost:6379"
+    max_tasks_per_minute: int = 60
+    heartbeat_interval: int = 30
+    worker_timeout_secs: int = 90
+    max_payload_bytes: int = 1_048_576
+    max_queue_depth: int = 100
+    local_workers: int = 0
+    local_worker_cmd: str = "devsper-worker"
+
+
+def load_pool_config(profile_override: str | None = None) -> PoolConfig:
+    profile = (profile_override or os.getenv("DEVSPER_PROFILE") or "").strip().lower() or "prod"
+    profile_path = Path(__file__).resolve().parent / "profiles" / f"{profile}.toml"
+    if not profile_path.exists():
+        profile = "prod"
+        profile_path = Path(__file__).resolve().parent / "profiles" / "prod.toml"
+
+    data = tomllib.loads(profile_path.read_text(encoding="utf-8"))
+    pool = data.get("pool", {})
+    limits = pool.get("limits", {}) if isinstance(pool.get("limits"), dict) else data.get("pool.limits", {})
+    # Our profile TOML uses [pool.limits] etc; tomllib nests them inside pool dict as "limits"
+    limits = pool.get("limits", {})
+
+    return PoolConfig(
+        profile=pool.get("profile", profile),
+        redis_url=pool.get("redis_url", "redis://localhost:6379"),
+        max_tasks_per_minute=int(pool.get("max_tasks_per_minute", 60)),
+        heartbeat_interval=int(pool.get("heartbeat_interval", 30)),
+        worker_timeout_secs=int(pool.get("worker_timeout_secs", 90)),
+        max_payload_bytes=int(limits.get("max_payload_bytes", 1_048_576)),
+        max_queue_depth=int(limits.get("max_queue_depth", 100)),
+        local_workers=int(pool.get("local_workers", 0)),
+        local_worker_cmd=str(pool.get("local_worker_cmd", "python -m devsper.agents.run_agent")),
+    )
+

devsper/pool/crypto.py

@@ -0,0 +1,75 @@
+"""
+E2EE for task payloads.
+
+Encryption scheme:
+  - Per-org X25519 key pair (private key stored in org keyring, public key registered with platform).
+  - For each task: ephemeral X25519 keypair -> shared secret -> HKDF-SHA256 -> AES-256-GCM.
+  - Wire format: ephemeral_pub (32B) || nonce (12B) || ciphertext+tag (len(pt)+16B)
+
+The pool stores and forwards ciphertext only; it never decrypts.
+"""
+
+from __future__ import annotations
+
+import os
+
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric.x25519 import (
+    X25519PrivateKey,
+    X25519PublicKey,
+)
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+from cryptography.hazmat.primitives.kdf.hkdf import HKDF
+
+
+def generate_org_keypair() -> tuple[bytes, bytes]:
+    priv = X25519PrivateKey.generate()
+    priv_b = priv.private_bytes(
+        serialization.Encoding.Raw,
+        serialization.PrivateFormat.Raw,
+        serialization.NoEncryption(),
+    )
+    pub_b = priv.public_key().public_bytes(
+        serialization.Encoding.Raw,
+        serialization.PublicFormat.Raw,
+    )
+    return priv_b, pub_b
+
+
+def encrypt_payload(plaintext: bytes, org_public_key_bytes: bytes) -> bytes:
+    eph_priv = X25519PrivateKey.generate()
+    eph_pub = eph_priv.public_key()
+    org_pub = X25519PublicKey.from_public_bytes(org_public_key_bytes)
+    shared = eph_priv.exchange(org_pub)
+    key = HKDF(
+        algorithm=hashes.SHA256(),
+        length=32,
+        salt=None,
+        info=b"devsper-task-v1",
+    ).derive(shared)
+    nonce = os.urandom(12)
+    ct = AESGCM(key).encrypt(nonce, plaintext, None)
+    eph_pub_b = eph_pub.public_bytes(
+        serialization.Encoding.Raw,
+        serialization.PublicFormat.Raw,
+    )
+    return eph_pub_b + nonce + ct
+
+
+def decrypt_payload(ciphertext: bytes, org_private_key_bytes: bytes) -> bytes:
+    if len(ciphertext) < 32 + 12 + 16:
+        raise ValueError("ciphertext too short")
+    eph_pub_b = ciphertext[:32]
+    nonce = ciphertext[32:44]
+    ct = ciphertext[44:]
+    org_priv = X25519PrivateKey.from_private_bytes(org_private_key_bytes)
+    eph_pub = X25519PublicKey.from_public_bytes(eph_pub_b)
+    shared = org_priv.exchange(eph_pub)
+    key = HKDF(
+        algorithm=hashes.SHA256(),
+        length=32,
+        salt=None,
+        info=b"devsper-task-v1",
+    ).derive(shared)
+    return AESGCM(key).decrypt(nonce, ct, None)
+

devsper/pool/local_pool.py

@@ -0,0 +1,126 @@
+from __future__ import annotations
+
+import asyncio
+import os
+import subprocess
+import uuid
+import logging
+import shlex
+import sys
+
+from .models import PoolTier, WorkerRecord, WorkerStatus
+from .manager import PoolManager
+from .store import RedisPoolStore, InMemoryPoolStore
+from .config import load_pool_config
+
+log = logging.getLogger("devsper.local_pool")
+
+
+class LocalWorkerPool:
+    def __init__(self, pool, config, redis_url: str):
+        self.pool = pool
+        self.config = config
+        self._redis_url = redis_url
+        self._procs: dict[str, subprocess.Popen] = {}
+
+    async def start(self, n: int | None = None):
+        n = n or getattr(self.config, "local_workers", 0) or 0
+        for _ in range(n):
+            await self._spawn_worker()
+
+    async def _spawn_worker(self):
+        worker_id = str(uuid.uuid4())
+        env = {
+            **os.environ,
+            "DEVSPER_WORKER_ID": worker_id,
+            "DEVSPER_PROFILE": "local",
+            # Workers must use the same Redis as registration / `devsper run` (defaults differ per module).
+            "REDIS_URL": self._redis_url,
+        }
+        cmd = shlex.split(str(getattr(self.config, "local_worker_cmd", "") or ""))
+        if not cmd:
+            cmd = ["python", "-c", "import time; time.sleep(3600)"]
+        if cmd and cmd[0] in ("python", "python3"):
+            cmd[0] = sys.executable
+        proc = subprocess.Popen(cmd, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        self._procs[worker_id] = proc
+
+        org_id = os.environ.get("DEVSPER_ORG_ID") or getattr(self.config, "org_id", None)
+        worker = WorkerRecord(
+            worker_id=worker_id,
+            node_id="local",
+            org_id=org_id,
+            tier=PoolTier.LOCAL,
+            status=WorkerStatus.IDLE,
+            profile="local",
+        )
+        await self.pool.register_worker(worker)
+        asyncio.create_task(self._monitor(worker_id, proc))
+
+    async def _monitor(self, worker_id: str, proc: subprocess.Popen):
+        while True:
+            await asyncio.sleep(5)
+            if proc.poll() is not None:
+                await self.pool.deregister_worker(worker_id)
+                self._procs.pop(worker_id, None)
+                await self._spawn_worker()
+                return
+            await self.pool.store.heartbeat(worker_id, ttl_secs=getattr(self.config, "worker_timeout_secs", 90))
+
+    async def stop(self):
+        for wid, proc in list(self._procs.items()):
+            try:
+                proc.terminate()
+            except Exception:
+                pass
+            await self.pool.deregister_worker(wid)
+        self._procs.clear()
+
+
+def main():
+    import argparse
+
+    parser = argparse.ArgumentParser(description="Start local worker pool (dev/test).")
+    parser.add_argument("--workers", type=int, default=None, help="Number of local workers to spawn")
+    args = parser.parse_args()
+
+    if not os.getenv("DEVSPER_PROFILE"):
+        os.environ["DEVSPER_PROFILE"] = "local"
+    cfg = load_pool_config()
+    # Prefer REDIS_URL env override for local dev/compose.
+    redis_url = os.getenv("REDIS_URL") or cfg.redis_url
+    os.environ.setdefault("REDIS_URL", redis_url)
+
+    use_memory = (os.getenv("DEVSPER_POOL_BACKEND", "").strip().lower() == "memory")
+    if use_memory:
+        store: RedisPoolStore | InMemoryPoolStore = InMemoryPoolStore()
+        log.warning("DEVSPER_POOL_BACKEND=memory: workers are in-process only; use TCP Redis for real E2E.")
+    else:
+        store = RedisPoolStore(redis_url)
+
+    class _Bus:
+        def __init__(self, url: str):
+            import redis as sync_redis
+
+            self._r = sync_redis.Redis.from_url(url, decode_responses=True)
+
+        def publish(self, channel: str, payload: dict):
+            import json as _json
+
+            self._r.publish(channel, _json.dumps(payload))
+
+    pool = PoolManager(store=store, bus=_Bus(redis_url), config=cfg)
+    lp = LocalWorkerPool(pool, cfg, redis_url=redis_url)
+
+    async def run():
+        await lp.start(args.workers)
+        # Run forever.
+        while True:
+            await asyncio.sleep(60)
+
+    asyncio.run(run())
+
+
+if __name__ == "__main__":
+    main()
+