What causes "ASGI context must come from authenticated server state" in JuliusBrussee/caveman?

Wrap the authenticated data in ASGIContext(scope= , recovery=...) before returning it from resolve_context

JuliusBrussee/caveman · error · TypeError

ASGI context must come from authenticated server state

Error message

ASGI context must come from authenticated server state

What it means

The middleware accepts a resolve_context callback but strictly requires that, when it returns a non-None value, that value is an ASGIContext instance built from authenticated server state. Returning any other object (a raw dict, a framework request/user object, a tuple) raises this TypeError during request handling. It enforces the library's security invariant that projected LLM context can never come from caller-controlled data.

Solutions

Wrap the authenticated data in ASGIContext(scope=<Scope>, recovery=...) before returning it from resolve_context
Return None instead of a substitute object when authentication/context is unavailable — None cleanly declines to passthrough
If returning an awaitable, await-able result, ensure the awaited value (not the coroutine wrapper) is the ASGIContext
Check the isinstance: the check is exact (isinstance against caveman_middleware.asgi.ASGIContext); subclasses are fine, duck-typed lookalikes are not

Example fix

// before
def resolve_context(scope):
    return {'scope': current_scope(), 'recovery': current_recovery()}  # TypeError
// after
from caveman_middleware.asgi import ASGIContext
def resolve_context(scope):
    s = current_scope()
    if s is None:
        return None
    return ASGIContext(scope=s, recovery=current_recovery())

Defensive patterns

Strategy: type-guard

Validate before calling

from caveman_middleware.asgi import ASGIContext
def resolve_context(scope):
    ctx = build_context(scope)
    if ctx is not None and not isinstance(ctx, ASGIContext):
        raise TypeError('resolver returned non-ASGIContext')
    return ctx

Type guard

def is_asgi_context(value) -> bool:
    from caveman_middleware.asgi import ASGIContext
    return value is None or isinstance(value, ASGIContext)

Try / catch

try:
    await middleware(scope, receive, send)
except TypeError as e:
    if 'authenticated server state' in str(e):
        logging.error('resolve_context returned wrong type: %s', e)
    raise

Prevention

Always construct the return value with ASGIContext(...); never return raw dicts or framework user objects
Return None to decline instead of a placeholder object
Add a unit test asserting resolve_context returns None or ASGIContext for every auth outcome

When it happens

Trigger: At request time, resolve_context(scope) (sync or awaited) returns a non-None value that is not an ASGIContext — e.g. returning scope['user'], a dict like {'scope': ..., 'recovery': ...}, a dataclass with a similar shape, or the result of a builder that wraps ASGIContext.

Common situations: Developers implement resolve_context by returning their framework's auth/user object directly; they deserialize a token/JSON body into a context-like dict; they refactor ASGIContext construction into a helper that accidentally returns .__dict__; or they confuse ASGIContext with the caveman_cloud Scope it wraps.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Authentication and authorization failures — expired tokens, bad credentials, and missing scopes.

Related errors

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

Appendix: source

Thrown at packages/middleware/python/caveman_middleware/asgi.py:124

            return await passthrough(receive, "protected_request")
        types = [value.lower().split(b";", 1)[0].strip() for key, value in headers if key.lower() == b"content-type"]
        lengths = [value for key, value in headers if key.lower() == b"content-length"]
        if types != [b"application/json"] or len(lengths) > 1:
            return await passthrough(receive, "unsupported_shape")
        if lengths:
            try:
                if not 0 <= int(lengths[0]) <= self.max_body_bytes:
                    return await passthrough(receive, "payload_limit")
            except ValueError:
                return await passthrough(receive, "unsupported_shape")

        context = self.resolve_context(scope)
        if inspect.isawaitable(context):
            context = await context
        if context is None:
            return await passthrough(receive, "scope_unavailable")
        if not isinstance(context, ASGIContext):
            raise TypeError("ASGI context must come from authenticated server state")

        buffered, parts, size = deque(), [], 0

        async def replay():
            return buffered.popleft() if buffered else await receive()

        while True:
            message = await receive()
            buffered.append(message)
            if message.get("type") != "http.request" or set(message) - {"type", "body", "more_body"}:
                return await passthrough(replay, "request_interrupted")
            body = message.get("body", b"")
            if type(body) is not bytes:
                return await passthrough(replay, "unsupported_shape")
            size += len(body)
            if size > self.max_body_bytes or len(buffered) > self.max_request_chunks:
                return await passthrough(replay, "payload_limit")
            parts.append(body)

View on GitHub (pinned to 3ee70a1026)