JuliusBrussee/caveman · error · TypeError

Synchronous recovery requires MiddlewareRuntime

Error message

Synchronous recovery requires MiddlewareRuntime

What it means

Raised by the sync recovery path in the LangChain middleware: synchronous recovery (the recover tool invoked outside async context) needs a real MiddlewareRuntime, but the registration was built without one (or with an unsupported runtime that declined). Sync recovery cannot be simulated without the engine-backed runtime.

Solutions

1. Provide a MiddlewareRuntime when constructing the connection/adapter.
2. Use the async agent path (arecover/ainvoke) with the async runtime instead.
3. Disable the recovery tool binding for sync-only deployments lacking a sync runtime.

Example fix

// before
conn = _Connection(async_runtime, scope)
agent.invoke({...})  # caveman_recover tool -> TypeError
// after
conn = _Connection(MiddlewareRuntime(...), scope)
agent.invoke({...})

if connection.sync is None:
    raise RuntimeError('caveman_recover requires a sync MiddlewareRuntime; bind arecover instead')

def can_recover_sync(conn): return getattr(conn, 'sync', None) is not None

try:
    data = recover(handle, config)
except TypeError as e:
    if 'Synchronous recovery' in str(e):
        data = await arecover(handle, config)
    else:
        raise

When it happens

Trigger: Agent executes the caveman_recover tool during a synchronous run whose connection has sync=None (built from an async-only runtime).

Common situations: Binding recovery tools while constructing the connection with an async runtime, then running a sync agent executor; enabling recovery in mixed sync/async pipelines.

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

• Synchronous calls require MiddlewareRuntime
• Synchronous document compression requires MiddlewareRuntime
• Synchronous LangChain calls require MiddlewareRuntime
• Expected a native LangChain BaseChatModel
• Invalid caveman_branch_id

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

Appendix: source

Thrown at packages/middleware/python/caveman_middleware/langchain.py:174

    with_caveman_agent() performs collision-safe model/tool registration together.
    scope may be a fixed Scope or a trusted Callable[[RunnableConfig], Scope].
    """
    tools = ()

    def __init__(self, *, runtime, scope):
        self.connection = _Connection(runtime, scope)
        self.recovery_tool = None
        if not _supported(runtime):
            if runtime.mode != "off":
                runtime.decline("unsupported_version")
            return
        if runtime.mode == "off":
            return

        def recover(handle: str, config: RunnableConfig, offset: int = 0, limit: int = 262144, query: str = ""):
            if self.connection.sync is None:
                raise TypeError("Synchronous recovery requires MiddlewareRuntime")
            return json.dumps(self.connection.sync.retrieve(_scope(scope, config), handle=handle, offset=offset, limit=limit, query=query), ensure_ascii=False, separators=(",", ":"))

        async def arecover(handle: str, config: RunnableConfig, offset: int = 0, limit: int = 262144, query: str = ""):
            return json.dumps(await self.connection.async_runtime.retrieve(_scope(scope, config), handle=handle, offset=offset, limit=limit, query=query), ensure_ascii=False, separators=(",", ":"))

        # Native tools expose mutable schemas and executor fields. Give the
        # framework its own schema, and attest the callable registration each
        # time before asking the runtime to issue lossy source grants.
        schema = copy.deepcopy(RECOVERY_SCHEMA)
        self.recovery_tool = StructuredTool.from_function(func=recover, coroutine=arecover, name="caveman_retrieve", description=RECOVERY_DESCRIPTION, args_schema=schema)
        self._recovery_schema = json.dumps(schema, sort_keys=True, allow_nan=False)
        self._recovery_methods = {name: getattr(self.recovery_tool, name) for name in
                                  ("func", "coroutine", "invoke", "ainvoke", "run", "arun")}

    def _recovery_intact(self):
        tool = self.recovery_tool
        try:
            return (type(tool) is StructuredTool and tool.name == "caveman_retrieve"

View on GitHub (pinned to 3ee70a1026)