{"record":{"id":"d941f52f68373e83","repo":"JuliusBrussee/caveman","slug":"agno-middleware-requires-a-nonempty-native-session-id","errorCode":null,"errorMessage":"Agno middleware requires a nonempty native session_id","messagePattern":"Agno middleware requires a nonempty native session_id","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"packages/middleware/python/caveman_middleware/agno.py","lineNumber":37,"sourceCode":"    from agno.tools.function import Function\nexcept ModuleNotFoundError as error:\n    raise ImportError(\"Install caveman-middleware[agno] to use the Agno adapter\") from error\n\nfrom caveman_cloud.middleware import Adapter, Candidate, MiddlewareError, MiddlewareRuntime, Scope\nfrom caveman_cloud.middleware.runtime import RECOVERY_DESCRIPTION, RECOVERY_SCHEMA\nfrom ._native import Attempt, manifest, owner, plain\nfrom ._versions import matches_framework\n\nADAPTER = Adapter(\"agno\", \"0.1.0\", \"3.0.9\", \"agno-message-v1\")\n_SIGNATURES = {name: inspect.signature(getattr(Model, name)) for name in\n               (\"response\", \"aresponse\", \"response_stream\", \"aresponse_stream\")}\n\n\ndef scope_from_run(run, *, namespace: str) -> Scope:\n    \"\"\"Resolve a native Agent or Team run's session, with explicit branch metadata.\"\"\"\n    session = getattr(run, \"session_id\", None)\n    if not isinstance(session, str) or not session:\n        raise ValueError(\"Agno middleware requires a nonempty native session_id\")\n    metadata = getattr(run, \"metadata\", None) or {}\n    return Scope(namespace, session, metadata.get(\"caveman_branch_id\", \"main\"), metadata.get(\"caveman_cache_epoch\", \"0\"))\n\n\ndef _scope(source, run):\n    result = source if isinstance(source, Scope) else source(run)\n    if not isinstance(result, Scope):\n        raise TypeError(\"Agno scope resolver must return a Caveman Scope\")\n    return result\n\n\ndef _message_view(messages):\n    if type(messages) is not list or any(type(message) is not Message for message in messages):\n        return None\n    # Metrics/timers and checkpoint bookkeeping are not model input. All other\n    # native values participate in lineage; opaque values bypass the whole view.\n    excluded = {\"metrics\", \"created_at\", \"from_history\", \"checkpoint_status\", \"checkpoint_created_at\"}\n    try:","sourceCodeStart":19,"sourceCodeEnd":55,"githubUrl":"https://github.com/JuliusBrussee/caveman/blob/3ee70a102609e550bd2e68004bf5990a9341c851/packages/middleware/python/caveman_middleware/agno.py#L19-L55","documentation":"scope_from_run derives a Caveman Scope from a native Agno Agent/Team run, and the session_id is the mandatory scope key for branching and recovery. If the run object has no session_id, or it is empty/non-string, scoping is impossible so it raises ValueError.","triggerScenarios":"Calling scope_from_run (or the adapter's state/_frame path that uses it) with a run object created without session_id, with session_id=\"\", or with a mock/stub lacking the attribute.","commonSituations":"Running an Agno agent in ephemeral/no-session mode; constructing runs manually in tests without session_id; upgrading agno where session assignment became lazy; passing the wrong object (e.g. a Run result instead of the run config).","solutions":["Ensure the Agent/Team run has a nonempty session_id: pass session_id=\"<id>\" when creating/running it","Generate a session id if your workflow has none (e.g. uuid4().hex) before invoking the adapter","Check you are passing the run object that actually carries session_id","Loosen the check only if you can supply an alternative scope key via a custom scope resolver returning a Scope"],"exampleFix":"# before\nagent.run(input)  # run.session_id is None -> ValueError\n# after\nagent.run(input, session_id=str(uuid4()))","handlingStrategy":"validation","validationCode":"def ensure_run_scopable(run):\n    sid = getattr(run, \"session_id\", None)\n    if not isinstance(sid, str) or not sid:\n        raise ValueError(\"run must carry a nonempty session_id before adapter use\")\n    return sid","typeGuard":"def has_session(run) -> bool:\n    sid = getattr(run, \"session_id\", None)\n    return isinstance(sid, str) and bool(sid)","tryCatchPattern":"try:\n    scope = scope_from_run(run, namespace=\"my-ns\")\nexcept ValueError as e:\n    if \"session_id\" in str(e):\n        run.session_id = str(uuid4())  # or recreate the run with one\n        scope = scope_from_run(run, namespace=\"my-ns\")\n    else:\n        raise","preventionTips":["Always create Agno Agent/Team runs with an explicit session_id","Assign uuid4-based session ids in ephemeral workflows","Validate run fixtures in tests carry session_id","After agno upgrades, re-check that session_id is populated at run construction"],"tags":["python","agno","session","scope"],"backgroundTag":"missing-required-argument","analyzedSha":"3ee70a102609e550bd2e68004bf5990a9341c851","analyzedAt":"2026-09-20T15:53:39.229Z","contentChangedAt":"2026-09-20T15:53:39.229Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}