| from collections.abc import Mapping |
| from typing import Any |
|
|
|
|
| def _mapping_get(value: Any, key: str) -> Any: |
| if isinstance(value, Mapping): |
| return value.get(key) |
|
|
| getter = getattr(value, "get", None) |
| if callable(getter): |
| try: |
| return getter(key, None) |
| except TypeError: |
| return getter(key) |
| return None |
|
|
|
|
| def get_runtime_context(runtime: Any) -> dict[str, Any]: |
| """Return runtime context as dict; tolerate missing/None context.""" |
| context = getattr(runtime, "context", None) |
| if isinstance(context, Mapping): |
| return dict(context) |
| if context is not None and hasattr(context, "items"): |
| return dict(context.items()) |
| return {} |
|
|
|
|
| def get_runtime_thread_id(runtime: Any) -> str | None: |
| """Resolve thread_id from runtime context first, then runtime config fallbacks.""" |
| context = get_runtime_context(runtime) |
| thread_id = context.get("thread_id") |
| if thread_id: |
| return str(thread_id) |
|
|
| config = getattr(runtime, "config", None) |
| metadata = _mapping_get(config, "metadata") |
| if metadata is not None: |
| thread_id = _mapping_get(metadata, "thread_id") |
| if thread_id: |
| return str(thread_id) |
|
|
| configurable = _mapping_get(config, "configurable") |
| if configurable is not None: |
| thread_id = _mapping_get(configurable, "thread_id") |
| if thread_id: |
| return str(thread_id) |
|
|
| return None |
|
|
