Allow opt-in concurrent Codex runs

Author: austinkennethtucker

.env.example

@@ -20,9 +20,14 @@ MAX_REQUEST_BODY_BYTES=262144
 MAX_MESSAGES=32
 MAX_TOTAL_TEXT_CHARS=80000
 
-# Local admission queue for short bursts while preserving one active Codex run.
+# Local admission queue for short bursts while all Codex run slots are busy.
 # 0 fails fast with wrapper_busy 429. Values above 0 wait up to that many
-# seconds for the active run to finish. The app clamps this to 0-5 seconds.
+# seconds for a run slot to open. The app clamps this to 0-5 seconds.
 QUEUE_WAIT_SECONDS=0
+
+# Optional provider-side parallelism. Keep 1 unless local wrapper_busy 429s are
+# the bottleneck and the signed-in Codex account tolerates two concurrent CLI
+# executions. The app clamps this to 1-2.
+MAX_CONCURRENT_CODEX_RUNS=1
 CORS_ALLOWED_ORIGINS=
 LOG_LEVEL=INFO

README.md

@@ -35,7 +35,8 @@ project.
 - `GET /v1/models` with the local model alias `codex-cli-default`.
 - `POST /v1/chat/completions` for text-only chat messages.
 - Non-streaming responses and final-only SSE streaming.
-- One active Codex execution at a time, with an optional short local wait queue.
+- One active Codex execution by default, with an optional second local execution
+  slot and short local wait queue.
 - A local operator dashboard at `http://127.0.0.1:8320/dashboard/`.
 
 ## What It Does Not Provide
@@ -228,8 +229,11 @@ For Obsidian LLM Wiki, use:
 required for the inspected OpenAI-compatible provider path.
 
 If near-simultaneous local requests collide, set `QUEUE_WAIT_SECONDS=2` or `3`
-in `.env` and restart the wrapper. The wrapper still permits only one active
-Codex run.
+in `.env` and restart the wrapper. If the collisions are confirmed
+`wrapper_busy` responses rather than upstream rate limits, you can also set
+`MAX_CONCURRENT_CODEX_RUNS=2` to permit a second local Codex CLI execution.
+Keep it at `1` if the signed-in Codex account starts returning
+`upstream_rate_limit`.
 
 ## Configuration
 
@@ -245,6 +249,7 @@ Most configuration lives in `.env`, copied from `.env.example`.
 | `MAX_MESSAGES` | Maximum number of chat messages. |
 | `MAX_TOTAL_TEXT_CHARS` | Maximum total text across messages. |
 | `QUEUE_WAIT_SECONDS` | Short local wait queue for bursty clients. Values are clamped to `0-5`. |
+| `MAX_CONCURRENT_CODEX_RUNS` | Number of provider-side Codex executions allowed at once. Values are clamped to `1-2`; default `1`. |
 | `CORS_ALLOWED_ORIGINS` | Optional comma-separated explicit origins. Requests without `Origin` are allowed. |
 | `LOG_LEVEL` | Python wrapper log level. |
 
@@ -360,11 +365,11 @@ docker compose restart
   `500000`, and restart.
 - `413` with `Request body too large`: increase `MAX_REQUEST_BODY_BYTES`, up to
   `2000000`, and restart.
-- `429` with `code: "wrapper_busy"`: wait for the active request to finish,
-  reduce client concurrency to `1`, increase batch delay, or set a small
-  `QUEUE_WAIT_SECONDS`.
+- `429` with `code: "wrapper_busy"`: wait for active requests to finish,
+  reduce client concurrency to `1`, increase batch delay, set a small
+  `QUEUE_WAIT_SECONDS`, or opt into `MAX_CONCURRENT_CODEX_RUNS=2`.
 - `429` with `code: "upstream_rate_limit"`: the signed-in upstream account is
-  rate limited; wait and retry later.
+  rate limited; wait and retry later, and keep `MAX_CONCURRENT_CODEX_RUNS=1`.
 - `502` or `/healthz` returning `503`: check
   `docker exec -it codex-cli-provider codex login status`, then re-run device
   login if needed.

src/dashboard_static/app.js

@@ -78,14 +78,16 @@ function renderStatus(data) {
 
   setText("#runner-value", runner.busy ? "Busy" : runner.ready ? "Ready" : "Unready");
   setClass("#runner-value", runner.ready && !runner.busy ? "ok-text" : runner.busy ? "warn-text" : "bad-text");
-  setText("#runner-detail", data.provider.modelAlias);
+  setText("#runner-detail", `${runner.activeRuns}/${runner.maxConcurrentRuns} active - ${data.provider.modelAlias}`);
 
   setText("#event-value", String(events.total));
   setText("#event-detail", `${events.errors} errors, p95 ${events.p95DurationMs ?? "-"} ms`);
 
   setText("#limit-body", formatBytes(limits.maxBodyBytes));
   setText("#limit-text", `${limits.maxTotalTextChars.toLocaleString()} chars`);
   setText("#limit-timeout", `${limits.requestTimeoutSeconds}s`);
+  setText("#limit-concurrency", `${limits.maxConcurrentRuns} run${limits.maxConcurrentRuns === 1 ? "" : "s"}`);
+  setText("#limit-queue", `${limits.queueWaitSeconds}s`);
   setText("#last-refresh", `Updated ${new Date().toLocaleTimeString()}`);
 }
 

src/dashboard_static/index.html

@@ -86,6 +86,14 @@ <h2>Runtime Limits</h2>
               <dt>Timeout</dt>
               <dd id="limit-timeout">-</dd>
             </div>
+            <div>
+              <dt>Runs</dt>
+              <dd id="limit-concurrency">-</dd>
+            </div>
+            <div>
+              <dt>Queue</dt>
+              <dd id="limit-queue">-</dd>
+            </div>
           </dl>
         </article>
 

src/server.py

@@ -145,6 +145,7 @@ class AppSettings:
     max_total_text_chars: int = 80_000
     codex_request_timeout_seconds: int = 180
     queue_wait_seconds: float = 0.0
+    max_concurrent_codex_runs: int = 1
     dashboard_enabled: bool = True
 
     def __post_init__(self) -> None:
@@ -156,6 +157,7 @@ def __post_init__(self) -> None:
         self.max_total_text_chars = min(max(int(self.max_total_text_chars), 1), 500_000)
         self.codex_request_timeout_seconds = min(max(int(self.codex_request_timeout_seconds), 5), 900)
         self.queue_wait_seconds = min(max(float(self.queue_wait_seconds), 0.0), 5.0)
+        self.max_concurrent_codex_runs = min(max(int(self.max_concurrent_codex_runs), 1), 2)
 
     @classmethod
     def from_env(cls) -> "AppSettings":
@@ -175,6 +177,7 @@ def from_env(cls) -> "AppSettings":
             max_total_text_chars=int(os.environ.get("MAX_TOTAL_TEXT_CHARS", "80000")),
             codex_request_timeout_seconds=int(os.environ.get("CODEX_REQUEST_TIMEOUT_SECONDS", "180")),
             queue_wait_seconds=float(os.environ.get("QUEUE_WAIT_SECONDS", "0")),
+            max_concurrent_codex_runs=int(os.environ.get("MAX_CONCURRENT_CODEX_RUNS", "1")),
             dashboard_enabled=os.environ.get("DASHBOARD_ENABLED", "true").strip().lower() not in {"0", "false", "no", "off"},
         )
 
@@ -340,32 +343,31 @@ def summarize_dashboard_events(events: list[dict[str, Any]]) -> dict[str, Any]:
     }
 
 
-async def execute_single_flight(app: FastAPI, settings: AppSettings, prompt: str) -> str:
-    lock: asyncio.Lock = app.state.execution_lock
+def wrapper_busy_error() -> APIError:
+    return APIError(
+        429,
+        "Another Codex execution is already running",
+        "rate_limit_error",
+        code="wrapper_busy",
+        headers={"Retry-After": "3"},
+    )
+
+
+async def execute_with_admission(app: FastAPI, settings: AppSettings, prompt: str) -> str:
+    semaphore: asyncio.Semaphore = app.state.execution_semaphore
     acquired = False
-    if lock.locked():
+    if semaphore.locked():
         if settings.queue_wait_seconds == 0:
-            raise APIError(
-                429,
-                "Another Codex execution is already running",
-                "rate_limit_error",
-                code="wrapper_busy",
-                headers={"Retry-After": "3"},
-            )
+            raise wrapper_busy_error()
         try:
-            await asyncio.wait_for(lock.acquire(), timeout=settings.queue_wait_seconds)
+            await asyncio.wait_for(semaphore.acquire(), timeout=settings.queue_wait_seconds)
             acquired = True
         except asyncio.TimeoutError as exc:
-            raise APIError(
-                429,
-                "Another Codex execution is already running",
-                "rate_limit_error",
-                code="wrapper_busy",
-                headers={"Retry-After": "3"},
-            ) from exc
+            raise wrapper_busy_error() from exc
     else:
-        await lock.acquire()
+        await semaphore.acquire()
         acquired = True
+    app.state.active_executions += 1
     try:
         return await app.state.runner.execute(
             prompt,
@@ -376,7 +378,8 @@ async def execute_single_flight(app: FastAPI, settings: AppSettings, prompt: str
         raise map_runner_error(exc) from exc
     finally:
         if acquired:
-            lock.release()
+            app.state.active_executions = max(0, app.state.active_executions - 1)
+            semaphore.release()
 
 
 def create_app(settings: AppSettings | None = None, runner: Any | None = None) -> FastAPI:
@@ -385,7 +388,8 @@ def create_app(settings: AppSettings | None = None, runner: Any | None = None) -
     app = FastAPI(title="codex-cli-provider", docs_url=None, redoc_url=None, openapi_url=None)
     app.state.settings = settings
     app.state.runner = runner
-    app.state.execution_lock = asyncio.Lock()
+    app.state.execution_semaphore = asyncio.Semaphore(settings.max_concurrent_codex_runs)
+    app.state.active_executions = 0
     app.state.dashboard_events = deque(maxlen=DASHBOARD_EVENT_LIMIT)
 
     if settings.cors_allowed_origins:
@@ -475,6 +479,7 @@ async def dashboard_status() -> JSONResponse:
         require_dashboard_enabled()
         status = await runner.status()
         events = list(app.state.dashboard_events)
+        active_executions = int(app.state.active_executions)
         return dashboard_json({
             "time": now_iso(),
             "provider": {
@@ -485,13 +490,17 @@ async def dashboard_status() -> JSONResponse:
                 },
                 "runner": {
                     "ready": bool(status.get("ready")),
-                    "busy": bool(app.state.execution_lock.locked()),
+                    "busy": active_executions >= settings.max_concurrent_codex_runs,
+                    "activeRuns": active_executions,
+                    "maxConcurrentRuns": settings.max_concurrent_codex_runs,
                 },
                 "limits": {
                     "maxBodyBytes": settings.max_body_bytes,
                     "maxMessages": settings.max_messages,
                     "maxTotalTextChars": settings.max_total_text_chars,
                     "requestTimeoutSeconds": settings.codex_request_timeout_seconds,
+                    "queueWaitSeconds": settings.queue_wait_seconds,
+                    "maxConcurrentRuns": settings.max_concurrent_codex_runs,
                 },
             },
             "events": summarize_dashboard_events(events),
@@ -535,7 +544,7 @@ async def chat(payload: dict[str, Any] = Body(...), _: None = Depends(require_au
         prompt = build_codex_prompt(messages)
         start = time.perf_counter()
         try:
-            text = await execute_single_flight(app, settings, prompt)
+            text = await execute_with_admission(app, settings, prompt)
         finally:
             elapsed_ms = int((time.perf_counter() - start) * 1000)
             LOGGER.info("request complete route=/v1/chat/completions elapsed_ms=%s model=%s", elapsed_ms, MODEL_ALIAS)

tests/test_server.py

@@ -90,6 +90,14 @@ async def request(app, method, url, token=TEST_SECRET, **kwargs):
         return await client.request(method, url, headers=headers, **kwargs)
 
 
+async def wait_for_prompt_count(runner, count):
+    for _ in range(50):
+        if len(runner.prompts) >= count:
+            return
+        await asyncio.sleep(0.01)
+    raise AssertionError(f"expected {count} prompts, got {len(runner.prompts)}")
+
+
 def test_default_app_uses_local_runner():
     app = create_app(settings())
     assert isinstance(app.state.runner, LocalCodexRunner)
@@ -101,6 +109,15 @@ def test_app_settings_do_not_require_runner_socket():
     assert not hasattr(configured, "runner_api_key")
 
 
+def test_admission_settings_are_clamped():
+    high = AppSettings(proxy_api_key=TEST_SECRET, queue_wait_seconds=99, max_concurrent_codex_runs=99)
+    low = AppSettings(proxy_api_key=TEST_SECRET, queue_wait_seconds=-1, max_concurrent_codex_runs=-1)
+    assert high.queue_wait_seconds == 5.0
+    assert high.max_concurrent_codex_runs == 2
+    assert low.queue_wait_seconds == 0.0
+    assert low.max_concurrent_codex_runs == 1
+
+
 @pytest.mark.asyncio
 async def test_health_ready_and_unready():
     ready_app = create_app(settings(), FakeRunner(ready=True))
@@ -149,6 +166,19 @@ async def test_dashboard_status_is_sanitized_and_unauthenticated():
     assert TEST_SECRET not in response.text
 
 
+@pytest.mark.asyncio
+async def test_dashboard_status_reports_admission_limits():
+    app = create_app(settings(max_concurrent_codex_runs=2, queue_wait_seconds=3), FakeRunner())
+    response = await request(app, "GET", "/dashboard/api/status", token=None)
+    assert response.status_code == 200
+    body = response.json()
+    assert body["provider"]["runner"]["activeRuns"] == 0
+    assert body["provider"]["runner"]["maxConcurrentRuns"] == 2
+    assert body["provider"]["runner"]["busy"] is False
+    assert body["provider"]["limits"]["queueWaitSeconds"] == 3
+    assert body["provider"]["limits"]["maxConcurrentRuns"] == 2
+
+
 @pytest.mark.asyncio
 async def test_dashboard_can_be_disabled():
     app = create_app(settings(dashboard_enabled=False), FakeRunner())
@@ -477,6 +507,47 @@ async def test_single_flight_queue_timeout_returns_wrapper_busy_429():
     assert second.json()["error"]["code"] == "wrapper_busy"
 
 
+@pytest.mark.asyncio
+async def test_two_execution_slots_allow_two_parallel_requests():
+    runner = BlockingRunner()
+    app = create_app(settings(max_concurrent_codex_runs=2, queue_wait_seconds=0), runner)
+    payload = {"model": "codex-cli-default", "messages": [{"role": "user", "content": "hello"}]}
+
+    first = asyncio.create_task(request(app, "POST", "/v1/chat/completions", json=payload))
+    second = asyncio.create_task(request(app, "POST", "/v1/chat/completions", json=payload))
+    await wait_for_prompt_count(runner, 2)
+
+    status = await request(app, "GET", "/dashboard/api/status", token=None)
+    assert status.json()["provider"]["runner"]["activeRuns"] == 2
+    assert status.json()["provider"]["runner"]["busy"] is True
+
+    runner.release.set()
+    first_response = await first
+    second_response = await second
+    assert first_response.status_code == 200
+    assert second_response.status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_concurrency_limit_returns_429_when_all_slots_busy():
+    runner = BlockingRunner()
+    app = create_app(settings(max_concurrent_codex_runs=2, queue_wait_seconds=0), runner)
+    payload = {"model": "codex-cli-default", "messages": [{"role": "user", "content": "hello"}]}
+
+    first = asyncio.create_task(request(app, "POST", "/v1/chat/completions", json=payload))
+    second = asyncio.create_task(request(app, "POST", "/v1/chat/completions", json=payload))
+    await wait_for_prompt_count(runner, 2)
+
+    third = await request(app, "POST", "/v1/chat/completions", json=payload)
+    runner.release.set()
+    await first
+    await second
+
+    assert third.status_code == 429
+    assert third.headers["retry-after"] == "3"
+    assert third.json()["error"]["code"] == "wrapper_busy"
+
+
 @pytest.mark.asyncio
 async def test_runner_errors_are_mapped():
     cases = [