fix(upstream): add fast-fail concurrency guards

README.md

@@ -73,6 +73,7 @@ Then start `opencode-a2a` against that upstream:
 ```bash
 A2A_BEARER_TOKEN=dev-token \
 OPENCODE_BASE_URL=http://127.0.0.1:4096 \
+A2A_TASK_STORE_DATABASE_URL=sqlite+aiosqlite:///./opencode-a2a.db \
 A2A_HOST=127.0.0.1 \
 A2A_PORT=8000 \
 A2A_PUBLIC_URL=http://127.0.0.1:8000 \

docs/guide.md

@@ -69,6 +69,10 @@ Key variables to understand protocol behavior:
   `session.abort` in cancel flow.
 - `OPENCODE_TIMEOUT` / `OPENCODE_TIMEOUT_STREAM`: upstream request timeout and
   optional stream timeout override.
+- `OPENCODE_MAX_CONCURRENT_REQUESTS`: optional fast-fail concurrency limit for
+  unary/control upstream calls. `0` disables the limit.
+- `OPENCODE_MAX_CONCURRENT_STREAMS`: optional fast-fail concurrency limit for
+  long-lived upstream `/event` streams. `0` disables the limit.
 - `A2A_CLIENT_TIMEOUT_SECONDS`: outbound client timeout. Default: `30` seconds.
 - `A2A_CLIENT_CARD_FETCH_TIMEOUT_SECONDS`: outbound Agent Card fetch timeout.
   Default: `5` seconds.
@@ -110,10 +114,10 @@ Current client facade API:
 - `A2AClient.cancel_task()`
 - `A2AClient.resubscribe_task()`
 
-Server-side outbound peer calls use bearer auth only for now. Configure
-`A2A_CLIENT_BEARER_TOKEN` when the remote agent protects its runtime surface.
-CLI outbound calls may pass `--token` explicitly or use
-`A2A_CLIENT_BEARER_TOKEN`.
+Server-side outbound peer calls read outbound credentials from environment
+variables. Configure `A2A_CLIENT_BEARER_TOKEN` or `A2A_CLIENT_BASIC_AUTH` when
+the remote agent protects its runtime surface. CLI outbound calls follow the
+same environment-only model.
 
 Execution-boundary metadata is intentionally declarative deployment metadata:
 it is published through `RuntimeProfile`, Agent Card, OpenAPI, and `/health`,

src/opencode_a2a/config.py

@@ -80,6 +80,16 @@ class Settings(BaseSettings):
     opencode_variant: str | None = Field(default=None, alias="OPENCODE_VARIANT")
     opencode_timeout: float = Field(default=120.0, alias="OPENCODE_TIMEOUT")
     opencode_timeout_stream: float | None = Field(default=None, alias="OPENCODE_TIMEOUT_STREAM")
+    opencode_max_concurrent_requests: int = Field(
+        default=0,
+        ge=0,
+        alias="OPENCODE_MAX_CONCURRENT_REQUESTS",
+    )
+    opencode_max_concurrent_streams: int = Field(
+        default=0,
+        ge=0,
+        alias="OPENCODE_MAX_CONCURRENT_STREAMS",
+    )
 
     # A2A settings
     a2a_public_url: str = Field(default="http://127.0.0.1:8000", alias="A2A_PUBLIC_URL")

src/opencode_a2a/execution/executor.py

@@ -28,7 +28,11 @@
     TextPart,
 )
 
-from ..opencode_upstream_client import OpencodeUpstreamClient, UpstreamContractError
+from ..opencode_upstream_client import (
+    OpencodeUpstreamClient,
+    UpstreamConcurrencyLimitError,
+    UpstreamContractError,
+)
 from ..parts.mapping import (
     UnsupportedA2AInputError,
     extract_text_from_a2a_parts,
@@ -274,6 +278,17 @@ async def run(self) -> None:
                 error_type="UPSTREAM_PAYLOAD_ERROR",
                 streaming_request=self._prepared.streaming_request,
             )
+        except UpstreamConcurrencyLimitError as exc:
+            logger.warning("OpenCode request rejected by concurrency budget: %s", exc)
+            await self._executor._emit_error(
+                self._event_queue,
+                task_id=self._task_id,
+                context_id=self._context_id,
+                message=str(exc),
+                state=TaskState.failed,
+                error_type="UPSTREAM_BACKPRESSURE",
+                streaming_request=self._prepared.streaming_request,
+            )
         except Exception as exc:
             logger.exception("OpenCode request failed")
             await self._executor._emit_error(

src/opencode_a2a/jsonrpc/application.py

@@ -21,7 +21,11 @@
     PROVIDER_DISCOVERY_ERROR_BUSINESS_CODES,
     SESSION_QUERY_ERROR_BUSINESS_CODES,
 )
-from ..opencode_upstream_client import OpencodeUpstreamClient, UpstreamContractError
+from ..opencode_upstream_client import (
+    OpencodeUpstreamClient,
+    UpstreamConcurrencyLimitError,
+    UpstreamContractError,
+)
 from .error_responses import (
     interrupt_not_found_error,
     invalid_params_error,
@@ -344,6 +348,14 @@ async def _handle_session_query_request(
                 base_request.id,
                 upstream_unreachable_error(ERR_UPSTREAM_UNREACHABLE),
             )
+        except UpstreamConcurrencyLimitError as exc:
+            return self._generate_error_response(
+                base_request.id,
+                upstream_unreachable_error(
+                    ERR_UPSTREAM_UNREACHABLE,
+                    detail=str(exc),
+                ),
+            )
         except Exception as exc:
             logger.exception("OpenCode session query JSON-RPC method failed")
             return self._generate_error_response(
@@ -470,6 +482,15 @@ async def _handle_provider_discovery_request(
                     method=base_request.method,
                 ),
             )
+        except UpstreamConcurrencyLimitError as exc:
+            return self._generate_error_response(
+                base_request.id,
+                upstream_unreachable_error(
+                    ERR_DISCOVERY_UPSTREAM_UNREACHABLE,
+                    method=base_request.method,
+                    detail=str(exc),
+                ),
+            )
         except Exception as exc:
             logger.exception("OpenCode provider discovery JSON-RPC method failed")
             return self._generate_error_response(
@@ -702,6 +723,17 @@ def _log_shell_audit(outcome: str) -> None:
                     session_id=session_id,
                 ),
             )
+        except UpstreamConcurrencyLimitError as exc:
+            _log_shell_audit("upstream_backpressure")
+            return self._generate_error_response(
+                base_request.id,
+                upstream_unreachable_error(
+                    ERR_UPSTREAM_UNREACHABLE,
+                    method=base_request.method,
+                    session_id=session_id,
+                    detail=str(exc),
+                ),
+            )
         except UpstreamContractError as exc:
             _log_shell_audit("upstream_payload_error")
             return self._generate_error_response(
@@ -903,6 +935,15 @@ async def _handle_interrupt_callback_request(
                     request_id=request_id,
                 ),
             )
+        except UpstreamConcurrencyLimitError as exc:
+            return self._generate_error_response(
+                base_request.id,
+                upstream_unreachable_error(
+                    ERR_UPSTREAM_UNREACHABLE,
+                    request_id=request_id,
+                    detail=str(exc),
+                ),
+            )
         except Exception as exc:
             logger.exception("OpenCode interrupt callback JSON-RPC method failed")
             return self._generate_error_response(

src/opencode_a2a/jsonrpc/error_responses.py

@@ -93,6 +93,7 @@ def upstream_unreachable_error(
     method: str | None = None,
     session_id: str | None = None,
     request_id: str | None = None,
+    detail: str | None = None,
 ) -> JSONRPCError:
     data: dict[str, Any] = {"type": "UPSTREAM_UNREACHABLE"}
     if method is not None:
@@ -101,6 +102,8 @@ def upstream_unreachable_error(
         data["session_id"] = session_id
     if request_id is not None:
         data["request_id"] = request_id
+    if detail is not None:
+        data["detail"] = detail
     return JSONRPCError(code=code, message="Upstream OpenCode unreachable", data=data)