Pixellab library expansion

pixellab_api.py

@@ -3,7 +3,8 @@ Minimal PixelLab v2 client used by pixellab_library.py.
 
 - Token from PIXELLAB_API_KEY (environment or .env next to this file).
 - Retries rate limits (429/529) with backoff, polls async jobs (edit endpoints return a job id),
-  and runs one job at a time across all processes (the account's concurrency limit is 1).
+  and caps concurrent jobs across ALL processes at the account limit (PIXELLAB_CONCURRENCY,
+  default 10) with one file lock per slot, so parallel workers and agents queue instead of 429.
 - Appends one line per paid call to pixellab_src/ledger.jsonl so spend can be tracked across
   runs and parallel workers: `python pixellab_library.py cost`.
 """
@@ -22,7 +23,8 @@ from PIL import Image
 
 API = "https://api.pixellab.ai/v2"
 LEDGER = Path(__file__).with_name("pixellab_src") / "ledger.jsonl"
-LOCK = LEDGER.with_name(".api.lock")
+LOCK_DIR = LEDGER.with_name(".api_slots")
+CONCURRENCY = int(os.environ.get("PIXELLAB_CONCURRENCY", "10"))
 RETRIES = 10
 JOB_POLL_SECONDS = 4
 JOB_TIMEOUT_SECONDS = 900
@@ -105,15 +107,22 @@ class Client:
     def run(self, endpoint, payload, label):
         """POST, wait for the job if there is one, return (image, usage). Logs the call to the ledger.
 
-        The account allows one job at a time, so jobs from every process (parallel workers, other
-        terminals) queue on a file lock instead of failing with 429."""
-        LOCK.parent.mkdir(parents=True, exist_ok=True)
-        with LOCK.open("a") as lock:
-            fcntl.flock(lock, fcntl.LOCK_EX)
-            try:
-                return self._run(endpoint, payload, label)
-            finally:
-                fcntl.flock(lock, fcntl.LOCK_UN)
+        Holds one of CONCURRENCY slot locks for the whole job (POST + polling)."""
+        LOCK_DIR.mkdir(parents=True, exist_ok=True)
+        while True:
+            for i in random.sample(range(CONCURRENCY), CONCURRENCY):
+                f = (LOCK_DIR / f"slot{i}").open("a")
+                try:
+                    fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
+                except BlockingIOError:
+                    f.close()
+                    continue
+                try:
+                    return self._run(endpoint, payload, label)
+                finally:
+                    fcntl.flock(f, fcntl.LOCK_UN)
+                    f.close()
+            time.sleep(1 + random.random())
 
     def _run(self, endpoint, payload, label):
         resp = self.call("POST", endpoint, payload)