JuliusBrussee/caveman · error · TypeError

Synchronous Agno recovery requires MiddlewareRuntime

Error message

Synchronous Agno recovery requires MiddlewareRuntime

What it means

The synchronous Agno recover callback delegates retrieval to a synchronous MiddlewareRuntime. If the adapter was constructed only with an async runtime, `self.sync` is None and synchronous recovery cannot proceed, so a TypeError is raised.

Solutions

1. Pass a synchronous MiddlewareRuntime when constructing the adapter so self.sync is set
2. Call the async `arecover` variant instead of sync `recover`
3. Wrap the sync call path to construct/reuse a MiddlewareRuntime

Example fix

# before
adapter = AgnoMiddleware(async_runtime=AsyncMiddlewareRuntime(...))
recover(handle, run_context)  # TypeError
# after
from caveman_cloud.middleware import MiddlewareRuntime
adapter = AgnoMiddleware(async_runtime=AsyncMiddlewareRuntime(...), sync=MiddlewareRuntime(...))
recover(handle, run_context)

if adapter_sync_runtime is None:
    raise RuntimeError('sync recovery needs a MiddlewareRuntime')

from caveman_cloud.middleware import MiddlewareRuntime
def supports_sync_recover(adapter) -> bool:
    return adapter.sync is not None and isinstance(adapter.sync, MiddlewareRuntime)

try:
    payload = recover(handle, run_context)
except TypeError:
    payload = await arecover(handle, run_context)  # async fallback

When it happens

Trigger: Registering the adapter with only `async_runtime` (or a MiddlewareRuntime lacking a sync counterpart) and then calling the sync `recover(handle, run_context, ...)` function.

Common situations: An async-only application (e.g. FastAPI/Agno async agents) that later invokes the sync recover path from a tool or CLI script.

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.

Related errors

• ASGI requires AsyncMiddlewareRuntime
• Match the sync/async middleware runtime to the native client
• Synchronous Agno calls require MiddlewareRuntime
• ASGI context must come from authenticated server state
• AutoGen requires a stable Caveman Scope for each agent or…

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

Appendix: source

Thrown at packages/middleware/python/caveman_middleware/agno.py:147

    def __init__(self, runtime, scope):
        self.sync = runtime if isinstance(runtime, MiddlewareRuntime) else None
        self.async_runtime = runtime.as_async() if self.sync else runtime
        self.scope, self.recovery_tool = scope, None
        self.active = contextvars.ContextVar("caveman_agno_run", default=None)
        self.version_supported = matches_framework(("agno", "3.0", "4"))
        if not self.version_supported and runtime.mode != "off":
            runtime.decline("unsupported_version")

    def passive(self, runtime, reason):
        return Attempt(runtime, None, str(uuid.uuid4()), str(uuid.uuid4()), passive=True, reason=reason, adapter=ADAPTER.id), {}, None, None

    def register(self):
        def recover(handle: str, run_context: RunContext, offset: int = 0, limit: int = 262144, query: str = ""):
            frame = self.active.get()
            if frame is None:
                raise MiddlewareError("recovery_unavailable")
            if self.sync is None:
                raise TypeError("Synchronous Agno recovery requires MiddlewareRuntime")
            if run_context is not None:
                raise_if_cancelled(run_context.run_id)
            return json.dumps(self.sync.retrieve(frame.scope, handle=handle, offset=offset, limit=limit, query=query),
                              ensure_ascii=False, separators=(",", ":"))

        async def arecover(handle: str, run_context: RunContext, offset: int = 0, limit: int = 262144, query: str = ""):
            frame = self.active.get()
            if frame is None:
                raise MiddlewareError("recovery_unavailable")
            if run_context is not None:
                await araise_if_cancelled(run_context.run_id)
            return json.dumps(await self.async_runtime.retrieve(frame.scope, handle=handle, offset=offset, limit=limit, query=query),
                              ensure_ascii=False, separators=(",", ":"))

        # Agno preserves a Function's entrypoint identity in its per-run copy.
        # Explicit processing avoids schema rewriting and keeps RunContext hidden.
        self.recovery_tool = Function(name="caveman_retrieve", description=RECOVERY_DESCRIPTION,
            parameters=copy.deepcopy(RECOVERY_SCHEMA), entrypoint=recover, skip_entrypoint_processing=True)

View on GitHub (pinned to 3ee70a1026)