JuliusBrussee/caveman · warning · MiddlewareError

code

Error message

code

What it means

In strict mode, _bypass() re-raises the bypass code as MiddlewareError(code) after invoking the diagnostic callback, instead of returning a bypassed Optimization. Callers of optimize()/decline() configured with strict=True get an exception for any bypass condition (capacity, off, unsupported_version, etc.).

Solutions

  1. Catch MiddlewareError from optimize()/decline() and fall back to unoptimized behavior.
  2. Set strict=False if bypasses should be returned as Optimization("bypassed", ...) rather than raised.
  3. Fix the underlying cause indicated by the code attribute (e.g. restore service capacity, enable the middleware).

Example fix

// before
result = runtime.optimize(scope, prompt)  # strict mode -> raises
// after
try:
    result = runtime.optimize(scope, prompt)
except MiddlewareError as e:
    result = None  # proceed without optimization
Defensive patterns

Strategy: try-catch

Validate before calling

def strict_enabled(rt) -> bool:
    return bool(getattr(rt, "strict", False))

Type guard

None

Try / catch

try:
    result = rt.optimize(scope, prompt)
except MiddlewareError as e:
    result = None  # strict-mode bypass surfaced as error; proceed unoptimized

Prevention

When it happens

Trigger: Running with strict=True and hitting a bypass path: capacity exhausted, mode off, or decline("unsupported_version") with diagnostic=True; also when the diagnostic callback itself raises (it is swallowed) but strict still escalates.

Common situations: Production strict deployments where the middleware service reports no capacity; teams enabling strict mode without monitoring the diagnostic hook.

Related errors


AI-assisted analysis of JuliusBrussee/caveman@3ee70a1026 (2026-09-20). Data as JSON: /api/errors/7a101ba09e27ae64. Report an issue: GitHub.

Appendix: source

Thrown at packages/sdk/python/caveman_cloud/middleware/runtime.py:352

    def __exit__(self, *_):
        self.close()

    def decline(self, reason: str) -> Optimization:
        """Report an untested native framework without sending content or I/O."""
        if reason != "unsupported_version":
            raise ValueError("Unsupported adapter diagnostic")
        if self.mode == "off":
            return Optimization("off", "disabled", (), None, None, "off")
        return self._bypass(reason)

    def _bypass(self, code: str, diagnostic: bool = True) -> Optimization:
        if diagnostic and self._diagnostic:
            try:
                self._diagnostic({"code": code, "cache_continuity": "unavailable"})
            except Exception:
                pass
        if self.strict and diagnostic:
            raise MiddlewareError(code)
        return Optimization("bypassed", code)

    def _http(self, path: str, body: str | None, timeout: float) -> Any:
        if self._closed:
            raise MiddlewareError("closed")
        headers = {"Content-Type": "application/json"}
        if self._token:
            headers["Authorization"] = f"Bearer {self._token}"
        started = time.monotonic()
        cls = http.client.HTTPSConnection if self._url.scheme == "https" else http.client.HTTPConnection
        connection = cls(self._url.hostname, self._url.port, timeout=max(0.001, timeout))
        with self._lock:
            if self._closed:
                raise MiddlewareError("closed")
            self._connections.add(connection)
        try:
            connection.connect()
            sock = connection.sock

View on GitHub (pinned to 3ee70a1026)