# src/pages/builder/templates/agent-python/memory.py
"""Conversation memory — ctx.store for message history CRUD.

提供基于 conversation_id 的消息追加、读取、清空、列举、删除、更新等操作。
底层直接使用 raw blob store（pages_blob.Store / LocalFileBlobStore），
自行管理 JSON 序列化，不复用 BlobBackedStore 的 envelope 格式。

典型用法::

    async def handler(ctx):
        # 追加 user 消息
        msg_id = await ctx.store.append_message(
            ctx.conversation_id, "user", "Hello!"
        )
        # 获取对话历史（默认按时间升序，方便拼 prompt）
        messages = await ctx.store.get_messages(ctx.conversation_id)
        # 转换为 OpenAI 格式
        openai_msgs = ctx.store.to_openai_input(messages)

数据布局（与 Node `agent/memory.ts` 完全一致）：

    conversations/{encoded_cid}/meta                        会话元数据（保留 messageCount=0 后仍存在）
    conversations/{encoded_cid}/messages/{ts}_{msg_id}      消息正文（按时间戳排序）
    message_index/{msg_id}                                  消息→key 的二级索引（O(1) 定位）
    conversation_index/{rev_ts}_{encoded_cid}               会话最近活跃时间倒排索引
    user_conversation_index/{encoded_uid}/{rev_ts}_{cid}    用户维度会话倒排索引
    langgraph_checkpoints/{thread_id}/checkpoints/{cid}     LangGraph checkpoint
    langgraph_checkpoints/{thread_id}/latest                LangGraph latest checkpoint id
    langgraph_checkpoints/{thread_id}/writes/{cid}/{tid}    LangGraph pending writes
"""
from __future__ import annotations

import base64
import json
import time
import uuid
import urllib.parse
from dataclasses import dataclass, field
from typing import Any, List, Optional


# ─── Error Classes ───


class MemoryError(Exception):
    """Base class for all memory errors."""
    pass


class MemoryValidationError(MemoryError):
    """Bad input: limit > 100, conversation_id > 256, content > 50MB, non-object metadata."""
    pass


class MemoryNotFoundError(MemoryError):
    """Conversation/message not found."""
    pass


class MemoryQuotaExceededError(MemoryError):
    """More than 10000 messages per conversation."""
    pass


class MemoryStorageError(MemoryError):
    """Blob storage failures."""
    pass


# ─── Data Model ───


@dataclass
class Message:
    """A single message in a conversation."""
    message_id: str
    role: str  # 'user' | 'assistant' | 'system' | 'tool'
    content: Any  # str or list (for multimodal)
    created_at: int  # ms timestamp
    metadata: Optional[dict] = None
    updated_at: Optional[int] = None

    def to_dict(self) -> dict:
        d: dict[str, Any] = {
            "messageId": self.message_id,
            "role": self.role,
            "content": self.content,
            "createdAt": self.created_at,
        }
        if self.metadata is not None:
            d["metadata"] = self.metadata
        if self.updated_at is not None:
            d["updatedAt"] = self.updated_at
        return d

    @classmethod
    def from_dict(cls, d: dict) -> "Message":
        return cls(
            message_id=d.get("message_id") or d.get("messageId", ""),
            role=d["role"],
            content=d["content"],
            created_at=d.get("created_at") or d.get("createdAt", 0),
            metadata=d.get("metadata"),
            updated_at=d.get("updated_at") or d.get("updatedAt"),
        )


@dataclass
class ConversationMeta:
    """Metadata for a conversation."""
    conversation_id: str
    created_at: int  # ms timestamp
    last_message_at: int  # ms timestamp
    message_count: int
    metadata: Optional[dict] = None

    def to_dict(self) -> dict:
        d: dict[str, Any] = {
            "conversationId": self.conversation_id,
            "createdAt": self.created_at,
            "lastMessageAt": self.last_message_at,
            "messageCount": self.message_count,
        }
        if self.metadata is not None:
            d["metadata"] = self.metadata
        return d

    @classmethod
    def from_dict(cls, d: dict) -> "ConversationMeta":
        cid = d.get("conversationId") or d.get("conversation_id") or ""
        if not cid:
            raise KeyError(f"conversationId missing in meta dict, keys={list(d.keys())}")
        return cls(
            conversation_id=cid,
            created_at=d.get("createdAt") or d.get("created_at") or 0,
            last_message_at=d.get("lastMessageAt") or d.get("last_message_at") or 0,
            message_count=d.get("messageCount") if "messageCount" in d else d.get("message_count", 0),
            metadata=d.get("metadata"),
        )


@dataclass
class ListConversationsResult:
    """Paginated result for list_conversations."""
    items: List[ConversationMeta] = field(default_factory=list)
    next_cursor: Optional[str] = None
    previous_cursor: Optional[str] = None


# ─── Constants ───

_MAX_LIMIT = 100
_DEFAULT_LIMIT = 20
_MAX_CONVERSATION_ID_LEN = 256
_MAX_MESSAGE_ID_LEN = 256
_MAX_CONTENT_SIZE = 50 * 1024 * 1024  # 50MB
_MAX_MESSAGES_PER_CONVERSATION = 10000
_VALID_ROLES = ("user", "assistant", "system", "tool")

_MESSAGE_PREFIX = "conversations"
_MESSAGE_INDEX_PREFIX = "message_index"
_CONVERSATION_INDEX_PREFIX = "conversation_index"
_USER_CONVERSATION_INDEX_PREFIX = "user_conversation_index"
_LANGGRAPH_CHECKPOINT_PREFIX = "langgraph_checkpoints"
_LANGGRAPH_STORE_PREFIX = "langgraph_store"
_LANGGRAPH_STORE_KEY_SEPARATOR = "__key__"
_MAX_SAFE_INTEGER = 9007199254740991  # JS Number.MAX_SAFE_INTEGER

# Sentinel for "argument not provided"，区别于显式传入的 None。
# update_message 用它来判断 content / metadata 是否被传，对齐 Node 的
# `input.content === undefined` 判断。
_UNSET: Any = object()


def _json_default(obj: Any) -> Any:
    """Fallback for json.dumps — handles LangGraph checkpoint objects
    containing non-serializable types (Runtime, ChannelProtocol, etc.)."""
    if hasattr(obj, "model_dump"):
        return obj.model_dump()
    # For non-serializable objects, return None to strip them rather than
    # corrupting checkpoint data with repr() strings.
    return None


def _encode_path_segment(value: str) -> str:
    """Base64url-encode a path segment (no padding). Matches Node encodePathSegment."""
    return base64.urlsafe_b64encode(value.encode("utf-8")).rstrip(b"=").decode("ascii")


# ─── Key Schema Helpers ───


def _encode_cid(conversation_id: str) -> str:
    """URL-encode conversation_id for use in blob keys."""
    return urllib.parse.quote(conversation_id, safe="")


def _encode_segment(value: str) -> str:
    """URL-encode an arbitrary path segment (thread_id / checkpoint_id / task_id)."""
    return urllib.parse.quote(value, safe="")


def _message_key(encoded_cid: str, created_at: int, message_id: str) -> str:
    """conversations/{encoded_cid}/messages/{created_at}_{message_id}"""
    return f"{_MESSAGE_PREFIX}/{encoded_cid}/messages/{created_at}_{message_id}"


def _messages_prefix(encoded_cid: str) -> str:
    """Prefix for listing all messages in a conversation."""
    return f"{_MESSAGE_PREFIX}/{encoded_cid}/messages/"


def _meta_key(encoded_cid: str) -> str:
    """conversations/{encoded_cid}/meta"""
    return f"{_MESSAGE_PREFIX}/{encoded_cid}/meta"


def _message_index_key(message_id: str) -> str:
    """message_index/{message_id}

    二级索引：通过 message_id 直达消息正文 key。和 Node 行为一致，
    支撑 update_message / delete_message 的 O(1) 定位。
    """
    return f"{_MESSAGE_INDEX_PREFIX}/{_encode_segment(message_id)}"


def _index_key(last_message_at: int, encoded_cid: str) -> str:
    """conversation_index/{reverseTimestamp}_{encoded_cid}

    Use reverse timestamp (MAX_SAFE_INTEGER - ts) padded to 16 digits,
    aligned with Node memory.ts. Lexicographic ascending = most recent first.
    """
    return f"{_CONVERSATION_INDEX_PREFIX}/{_reverse_timestamp(last_message_at)}_{encoded_cid}"


def _reverse_timestamp(ts: int) -> str:
    """MAX_SAFE_INTEGER - ts, zero-padded to 16 digits. Matches Node reverseTimestamp()."""
    return str(_MAX_SAFE_INTEGER - ts).zfill(16)


def _user_index_key(encoded_user_id: str, last_message_at: int, encoded_cid: str) -> str:
    """user_conversation_index/{encoded_user_id}/{reverseTimestamp}_{encoded_cid}"""
    return f"{_USER_CONVERSATION_INDEX_PREFIX}/{encoded_user_id}/{_reverse_timestamp(last_message_at)}_{encoded_cid}"


def _user_index_prefix(encoded_user_id: str) -> str:
    """Prefix for listing all conversations of a user."""
    return f"{_USER_CONVERSATION_INDEX_PREFIX}/{encoded_user_id}/"


def _index_prefix() -> str:
    return f"{_CONVERSATION_INDEX_PREFIX}/"


def _generate_message_id() -> str:
    """Generate a unique message ID: msg_{random_hex}"""
    return f"msg_{uuid.uuid4().hex[:16]}"


def _extract_message_id_from_key(key: str) -> str:
    """Extract messageId from a message key.

    Key format: conversations/{cid}/messages/{created_at}_{message_id}
    The message_id part starts with 'msg_' so we split on the first '_' after
    the timestamp to get everything after it.
    """
    basename = key.rsplit("/", 1)[-1]  # "{created_at}_{message_id}"
    parts = basename.split("_", 1)
    return parts[1] if len(parts) > 1 else ""


# ─── Cursor Helpers ───


def _encode_cursor(last_message_at: int, conversation_id: str) -> str:
    """Encode pagination cursor as base64url JSON. Opaque to callers."""
    payload = json.dumps(
        {"v": 1, "lastMessageAt": last_message_at, "conversationId": conversation_id},
        separators=(",", ":"),
    )
    return base64.urlsafe_b64encode(payload.encode()).decode().rstrip("=")


def _decode_cursor(cursor: str) -> dict:
    """Decode and validate an opaque pagination cursor.

    Raises MemoryValidationError on invalid/unsupported cursor.
    """
    padded = cursor + "=" * (-len(cursor) % 4)
    try:
        raw = base64.urlsafe_b64decode(padded)
        data = json.loads(raw)
    except Exception:
        raise MemoryValidationError("Invalid cursor.")
    if not isinstance(data, dict) or data.get("v") != 1:
        raise MemoryValidationError("Unsupported cursor version.")
    if not isinstance(data.get("lastMessageAt"), int) or not isinstance(data.get("conversationId"), str):
        raise MemoryValidationError("Invalid cursor fields.")
    return data


def _cursor_sort_key(last_message_at: int, conversation_id: str) -> str:
    """Build the sort key string that a cursor maps to in the index.

    Format: {revTs:016d}_{encoded_cid} — same format as index key basenames.
    """
    return f"{_reverse_timestamp(last_message_at)}_{_encode_cid(conversation_id)}"


# ─── ConversationMemory ───


class ConversationMemory:
    """Conversation message history CRUD, accessed as ctx.store.

    Operates on a dedicated blob store (separate from per-route ctx.kv).
    All core methods are async.
    """

    def __init__(self, blob_store: Any, run_id: str) -> None:
        """
        Args:
            blob_store: Raw blob store instance (pages_blob.Store or LocalFileBlobStore).
                        Must implement: get(key, type=), set(key, value), delete(key),
                        list(prefix=) returning object with .blobs list of objects with .key.
            run_id: Current run ID, auto-injected into message metadata.
        """
        self._blob = blob_store
        self._run_id = run_id

    # ─── Validation Helpers ───

    def _validate_conversation_id(self, conversation_id: str) -> None:
        if not conversation_id or not isinstance(conversation_id, str):
            raise MemoryValidationError("conversation_id must be a non-empty string")
        if len(conversation_id) > _MAX_CONVERSATION_ID_LEN:
            raise MemoryValidationError(
                f"conversation_id exceeds max length of {_MAX_CONVERSATION_ID_LEN} characters"
            )

    def _validate_message_id(self, message_id: str) -> None:
        if not message_id or not isinstance(message_id, str):
            raise MemoryValidationError("message_id must be a non-empty string")
        if len(message_id) > _MAX_MESSAGE_ID_LEN:
            raise MemoryValidationError(
                f"message_id exceeds max length of {_MAX_MESSAGE_ID_LEN} characters"
            )

    def _validate_user_id(self, user_id: str) -> None:
        if not user_id or not isinstance(user_id, str):
            raise MemoryValidationError("user_id must be a non-empty string")
        if len(user_id) > _MAX_CONVERSATION_ID_LEN:
            raise MemoryValidationError(
                f"user_id exceeds max length of {_MAX_CONVERSATION_ID_LEN} characters"
            )

    def _validate_limit(self, limit: int) -> None:
        if not isinstance(limit, int) or limit < 1:
            raise MemoryValidationError("limit must be a positive integer")
        if limit > _MAX_LIMIT:
            raise MemoryValidationError(f"limit exceeds maximum of {_MAX_LIMIT}")

    def _validate_role(self, role: str) -> None:
        if role not in _VALID_ROLES:
            raise MemoryValidationError(
                f"role must be one of {_VALID_ROLES}, got '{role}'"
            )

    def _validate_content(self, content: Any) -> None:
        if content is None:
            raise MemoryValidationError("content must not be None")
        serialized = json.dumps(content, ensure_ascii=False) if not isinstance(content, str) else content
        if len(serialized.encode("utf-8")) > _MAX_CONTENT_SIZE:
            raise MemoryValidationError(
                f"content exceeds maximum size of {_MAX_CONTENT_SIZE // (1024 * 1024)}MB"
            )

    def _validate_metadata(self, metadata: Any) -> None:
        if metadata is not None and not isinstance(metadata, dict):
            raise MemoryValidationError("metadata must be a dict or None")

    @staticmethod
    def _assert_single_cursor(after: Optional[str], before: Optional[str]) -> None:
        if after is not None and before is not None:
            raise MemoryValidationError("after and before are mutually exclusive")

    # ─── Blob I/O Helpers ───

    async def _blob_get_json(self, key: str, **kwargs) -> Optional[dict]:
        """Get and parse JSON from blob store. Returns None if not found."""
        try:
            raw = await self._blob.get(key, type="text", **kwargs)
            if raw is None:
                return None
            return json.loads(raw)
        except (json.JSONDecodeError, TypeError, ValueError):
            return None
        except MemoryStorageError:
            raise
        except Exception as e:
            raise MemoryStorageError(f"Failed to read key '{key}': {e}") from e

    async def _blob_set_json(self, key: str, data: Any) -> None:
        """Serialize value to JSON and write to blob store.

        ``data`` 可以是 dict / list / 字符串 / 数字等任意可 JSON 序列化的对象，
        以便支持 ``message_index`` 这种简单结构 / LangGraph latest checkpoint id
        这种字符串 payload。
        """
        try:
            payload = json.dumps(data, ensure_ascii=False, separators=(",", ":"), default=_json_default)
            await self._blob.set(key, payload)
        except MemoryStorageError:
            raise
        except Exception as e:
            raise MemoryStorageError(f"Failed to write key '{key}': {e}") from e

    async def _blob_get_text(self, key: str, **kwargs) -> Optional[str]:
        """Get raw text from blob store. Returns None if not found."""
        try:
            raw = await self._blob.get(key, type="text", **kwargs)
            if raw is None:
                return None
            return raw if isinstance(raw, str) else str(raw)
        except MemoryStorageError:
            raise
        except Exception as e:
            raise MemoryStorageError(f"Failed to read key '{key}': {e}") from e

    async def _blob_delete(self, key: str) -> None:
        """Delete a key from blob store (idempotent)."""
        try:
            await self._blob.delete(key)
        except MemoryStorageError:
            raise
        except Exception as e:
            raise MemoryStorageError(f"Failed to delete key '{key}': {e}") from e

    async def _blob_list_keys(self, prefix: str, *, limit: Optional[int] = None, **kwargs) -> List[str]:
        """List keys with given prefix, sorted lexicographically.

        Args:
            prefix: Key prefix to filter.
            limit: Max keys to return (None = all). Passed to blob SDK.
        """
        try:
            list_kwargs = {"prefix": prefix, **kwargs}
            if limit is not None:
                list_kwargs["limit"] = limit
                list_kwargs["paginate"] = False
            result = await self._blob.list(**list_kwargs)
            return [blob.key for blob in result.blobs]
        except MemoryStorageError:
            raise
        except Exception as e:
            raise MemoryStorageError(f"Failed to list keys with prefix '{prefix}': {e}") from e

    # ─── Core API Methods ───

    async def append_message(
        self,
        conversation_id: str,
        role: str,
        content: Any,
        metadata: Optional[dict] = None,
        user_id: Optional[str] = None,
    ) -> str:
        """Append a message to a conversation.

        Args:
            conversation_id: Conversation identifier (max 256 chars).
            role: One of 'user', 'assistant', 'system', 'tool'.
            content: Message content (str or list for multimodal).
            metadata: Optional metadata dict. ``run_id`` is auto-injected unless
                already present. Pass ``idempotency_key`` in metadata to enable
                idempotent writes.
            user_id: Optional user identifier. When provided, the conversation
                is indexed under this user for later retrieval via
                ``list_conversations(user_id=...)``.

        Returns:
            The generated message_id.

        Raises:
            MemoryValidationError: Invalid input.
            MemoryQuotaExceededError: Conversation has >= 10000 messages.
            MemoryStorageError: Blob store failure.
        """
        # Validate
        self._validate_conversation_id(conversation_id)
        self._validate_role(role)
        self._validate_content(content)
        self._validate_metadata(metadata)
        if user_id is not None:
            self._validate_user_id(user_id)

        encoded_cid = _encode_cid(conversation_id)
        now_ms = int(time.time() * 1000)

        # Prepare metadata with run_id injection
        if metadata is None:
            metadata = {}
        else:
            metadata = dict(metadata)  # shallow copy to avoid mutating caller's dict
        if "run_id" not in metadata:
            metadata["run_id"] = self._run_id

        # Idempotency check
        idempotency_key = metadata.get("idempotency_key")
        if idempotency_key is not None:
            existing_id = await self._check_idempotency(encoded_cid, idempotency_key)
            if existing_id is not None:
                return existing_id

        # Load conversation meta (or prepare to create new)
        meta_k = _meta_key(encoded_cid)
        meta_data = await self._blob_get_json(meta_k)

        if meta_data is not None:
            meta = ConversationMeta.from_dict(meta_data)
            if meta.message_count >= _MAX_MESSAGES_PER_CONVERSATION:
                raise MemoryQuotaExceededError(
                    f"Conversation '{conversation_id}' has reached the maximum of "
                    f"{_MAX_MESSAGES_PER_CONVERSATION} messages"
                )
            old_last_message_at = meta.last_message_at
        else:
            meta = ConversationMeta(
                conversation_id=conversation_id,
                created_at=now_ms,
                last_message_at=0,
                message_count=0,
                metadata=None,
            )
            old_last_message_at = None  # no old index to delete

        # Generate message
        message_id = _generate_message_id()
        message = Message(
            message_id=message_id,
            role=role,
            content=content,
            created_at=now_ms,
            metadata=metadata if metadata else None,
        )

        # --- Parallel writes (no data dependency between these) ---
        import asyncio

        msg_key = _message_key(encoded_cid, now_ms, message_id)
        new_index_key = _index_key(now_ms, encoded_cid)
        index_entry = {"conversationId": conversation_id, "lastMessageAt": now_ms}

        # Update conversation meta (prepare before write)
        meta.last_message_at = now_ms
        meta.message_count += 1

        # Batch write: message body + message index + meta + conversation index
        write_tasks = [
            self._blob_set_json(msg_key, message.to_dict()),
            self._blob_set_json(
                _message_index_key(message_id),
                {
                    "conversationId": conversation_id,
                    "key": msg_key,
                    "messageId": message_id,
                    "createdAt": now_ms,
                },
            ),
            self._blob_set_json(meta_k, meta.to_dict()),
            self._blob_set_json(new_index_key, index_entry),
        ]

        # User conversation index (if user_id provided)
        if user_id is not None:
            encoded_user_id = _encode_segment(user_id)
            new_user_key = _user_index_key(encoded_user_id, now_ms, encoded_cid)
            index_entry_user = {"conversationId": conversation_id, "lastMessageAt": now_ms}
            write_tasks.append(self._blob_set_json(new_user_key, index_entry_user))

        await asyncio.gather(*write_tasks)

        # --- Cleanup old index keys (can also be parallel) ---
        delete_tasks = []
        if old_last_message_at is not None and old_last_message_at != now_ms:
            old_index_key = _index_key(old_last_message_at, encoded_cid)
            delete_tasks.append(self._blob_delete(old_index_key))

        if user_id is not None:
            encoded_user_id = _encode_segment(user_id)
            if old_last_message_at is not None and old_last_message_at != now_ms:
                old_user_key = _user_index_key(encoded_user_id, old_last_message_at, encoded_cid)
                delete_tasks.append(self._blob_delete(old_user_key))
            elif old_last_message_at is None:
                # First message in this conversation — scan for stale user index
                # (edge case: conversation was deleted + recreated under same id)
                prefix = _user_index_prefix(encoded_user_id)
                existing_keys = await self._blob_list_keys(prefix)
                suffix = f"_{encoded_cid}"
                for k in existing_keys:
                    if k.endswith(suffix) and k != _user_index_key(encoded_user_id, now_ms, encoded_cid):
                        delete_tasks.append(self._blob_delete(k))

        if delete_tasks:
            await asyncio.gather(*delete_tasks)

        return message_id

    async def _bulk_append_messages(
        self,
        conversation_id: str,
        messages_data: List[dict],
    ) -> List[str]:
        """Internal bulk append: write N messages with only 1 meta/index update.

        Args:
            conversation_id: Conversation identifier.
            messages_data: List of dicts with keys: role, content, metadata.

        Returns:
            List of generated message_ids.
        """
        import asyncio

        if not messages_data:
            return []

        encoded_cid = _encode_cid(conversation_id)
        now_ms = int(time.time() * 1000)

        # Load meta once
        meta_k = _meta_key(encoded_cid)
        meta_data = await self._blob_get_json(meta_k)

        if meta_data is not None:
            meta = ConversationMeta.from_dict(meta_data)
            if meta.message_count + len(messages_data) > _MAX_MESSAGES_PER_CONVERSATION:
                raise MemoryQuotaExceededError(
                    f"Conversation '{conversation_id}' would exceed {_MAX_MESSAGES_PER_CONVERSATION} messages"
                )
            old_last_message_at = meta.last_message_at
        else:
            meta = ConversationMeta(
                conversation_id=conversation_id,
                created_at=now_ms,
                last_message_at=0,
                message_count=0,
                metadata=None,
            )
            old_last_message_at = None

        # Build all messages and collect write tasks
        write_tasks = []
        message_ids = []
        last_ts = now_ms

        for i, msg_data in enumerate(messages_data):
            # Increment timestamp by 1ms per message to ensure unique keys + ordering
            ts = now_ms + i
            last_ts = ts
            message_id = _generate_message_id()
            message_ids.append(message_id)

            msg_metadata = dict(msg_data.get("metadata") or {})
            if "run_id" not in msg_metadata:
                msg_metadata["run_id"] = self._run_id

            message = Message(
                message_id=message_id,
                role=msg_data["role"],
                content=msg_data["content"],
                created_at=ts,
                metadata=msg_metadata if msg_metadata else None,
            )

            msg_key = _message_key(encoded_cid, ts, message_id)
            write_tasks.append(self._blob_set_json(msg_key, message.to_dict()))
            write_tasks.append(self._blob_set_json(
                _message_index_key(message_id),
                {"conversationId": conversation_id, "key": msg_key, "messageId": message_id, "createdAt": ts},
            ))

        # Update meta once (for all messages)
        meta.last_message_at = last_ts
        meta.message_count += len(messages_data)
        write_tasks.append(self._blob_set_json(meta_k, meta.to_dict()))

        # Update conversation index once
        new_index_key = _index_key(last_ts, encoded_cid)
        write_tasks.append(self._blob_set_json(
            new_index_key, {"conversationId": conversation_id, "lastMessageAt": last_ts}
        ))

        # Parallel write all
        await asyncio.gather(*write_tasks)

        # Cleanup old index
        if old_last_message_at is not None and old_last_message_at != last_ts:
            await self._blob_delete(_index_key(old_last_message_at, encoded_cid))

        return message_ids

    async def _check_idempotency(self, encoded_cid: str, idempotency_key: str) -> Optional[str]:
        """Scan recent messages for matching idempotency_key. Returns message_id if found."""
        import asyncio

        prefix = _messages_prefix(encoded_cid)
        keys = await self._blob_list_keys(prefix)
        # Scan at most last 50 messages — parallel fetch for performance
        check_keys = keys[-min(len(keys), 50):]
        if not check_keys:
            return None

        results = await asyncio.gather(*[self._blob_get_json(k) for k in check_keys])
        for data in results:
            if data and isinstance(data.get("metadata"), dict):
                if data["metadata"].get("idempotency_key") == idempotency_key:
                    return data.get("messageId") or data.get("message_id", "")
        return None

    async def get_messages(
        self,
        conversation_id: str,
        limit: int = _DEFAULT_LIMIT,
        order: str = "asc",
        after: Optional[str] = None,
        before: Optional[str] = None,
    ) -> List[Message]:
        """Get messages from a conversation.

        Args:
            conversation_id: Conversation identifier.
            limit: Max messages to return (default 20, max 100).
            order: 'asc' (oldest first, default) or 'desc' (newest first).
            after: Cursor — message_id to start after (exclusive).
            before: Cursor — message_id to end before (exclusive).

        Returns:
            List of Message objects in requested order.
        """
        import asyncio

        self._validate_conversation_id(conversation_id)
        self._validate_limit(limit)
        if order not in ("asc", "desc"):
            raise MemoryValidationError("order must be 'asc' or 'desc'")
        self._assert_single_cursor(after, before)

        encoded_cid = _encode_cid(conversation_id)
        prefix = _messages_prefix(encoded_cid)
        keys = await self._blob_list_keys(prefix, consistency="strong")
        # keys are sorted lexicographically by {created_at}_{message_id}

        # Apply cursor filtering
        if after is not None:
            after_idx = self._find_message_key_index(keys, after)
            if after_idx is not None:
                keys = keys[after_idx + 1:]

        if before is not None:
            before_idx = self._find_message_key_index(keys, before)
            if before_idx is not None:
                keys = keys[:before_idx]

        # Apply order and limit
        if order == "desc":
            keys = list(reversed(keys))

        keys = keys[:limit]

        # Fetch message bodies concurrently for much lower latency.
        # Instead of N sequential HTTP round-trips, all reads happen in parallel
        # (wall-clock = 1 RTT instead of N × RTT).
        async def _fetch_one(key: str) -> Optional[Message]:
            data = await self._blob_get_json(key, consistency="strong")
            return Message.from_dict(data) if data else None

        results = await asyncio.gather(*[_fetch_one(k) for k in keys])
        return [m for m in results if m is not None]

    @staticmethod
    def _find_message_key_index(keys: List[str], message_id: str) -> Optional[int]:
        """Find index of key containing given message_id."""
        suffix = f"_{message_id}"
        for i, key in enumerate(keys):
            if key.endswith(suffix):
                return i
        return None

    async def update_message(
        self,
        conversation_id: str,
        message_id: str,
        content: Any = _UNSET,
        metadata: Any = _UNSET,
    ) -> Message:
        """Update an existing message's content / metadata.

        语义对齐 Node ``memory.updateMessage``：
          - ``content``：传入即整体替换。未传则保持不变。
          - ``metadata``：传入即与原 metadata **浅合并**（同 key 覆盖、新 key 追加）；
            传入 ``None`` 视为「未传」（不变）。
          - 自动写入 ``updated_at`` 时间戳。

        Args:
            conversation_id: 必须和 message 当前所属 conversation 一致；
                对不上抛 ``MemoryNotFoundError``。
            message_id: 待更新消息 id。
            content: 新 content（str / list / dict 等）；省略表示保留原值。
            metadata: 用于浅合并的 metadata；省略 / None 表示保留原值。

        Returns:
            更新后的 ``Message`` 对象。

        Raises:
            MemoryValidationError: content / metadata / id 不合法或两个字段都没传。
            MemoryNotFoundError: 消息不存在或 conversation 对不上。
            MemoryStorageError: blob 读写失败。
        """
        self._validate_conversation_id(conversation_id)
        self._validate_message_id(message_id)

        content_provided = content is not _UNSET
        metadata_provided = metadata is not _UNSET

        if not content_provided and not metadata_provided:
            raise MemoryValidationError("content or metadata is required for update_message")

        if content_provided:
            self._validate_content(content)
        if metadata_provided and metadata is not None and not isinstance(metadata, dict):
            raise MemoryValidationError("metadata must be a dict or None")

        # 通过 message_index 二级索引拿到原始 key
        index_data = await self._blob_get_json(_message_index_key(message_id))
        if not index_data or (index_data.get("conversationId") or index_data.get("conversation_id")) != conversation_id:
            raise MemoryNotFoundError(
                f"Message '{message_id}' not found in conversation '{conversation_id}'"
            )

        msg_key = index_data.get("key")
        if not msg_key:
            raise MemoryNotFoundError(f"Message '{message_id}' has no stored key")

        existed = await self._blob_get_json(msg_key)
        if not existed:
            raise MemoryNotFoundError(f"Message '{message_id}' not found")

        existed_msg = Message.from_dict(existed)
        if content_provided:
            existed_msg.content = content
        if metadata_provided and metadata is not None:
            base = dict(existed_msg.metadata or {})
            base.update(metadata)
            existed_msg.metadata = base if base else None
        existed_msg.updated_at = int(time.time() * 1000)

        await self._blob_set_json(msg_key, existed_msg.to_dict())
        return existed_msg

    async def delete_message(self, conversation_id: str, message_id: str) -> None:
        """Delete a single message by message_id.

        语义对齐 Node ``memory.deleteMessage``：
          - 通过 message_index 定位 → 删消息本体 → 删 message_index → 重算 meta。
          - 找不到时抛 ``MemoryNotFoundError``。
          - 删完最后一条消息后，meta 仍**保留**（``message_count=0``，``last_message_at``
            回退到 ``created_at``），与 ``clear_messages`` 行为一致。

        Args:
            conversation_id: 消息所属 conversation。
            message_id: 待删除消息 id。

        Raises:
            MemoryValidationError: id 不合法。
            MemoryNotFoundError: 消息或 conversation 不存在。
            MemoryStorageError: blob 读写失败。
        """
        self._validate_conversation_id(conversation_id)
        self._validate_message_id(message_id)

        index_data = await self._blob_get_json(_message_index_key(message_id))
        if not index_data or (index_data.get("conversationId") or index_data.get("conversation_id")) != conversation_id:
            raise MemoryNotFoundError(
                f"Message '{message_id}' not found in conversation '{conversation_id}'"
            )

        msg_key = index_data.get("key")
        if not msg_key:
            raise MemoryNotFoundError(f"Message '{message_id}' has no stored key")

        encoded_cid = _encode_cid(conversation_id)
        meta_k = _meta_key(encoded_cid)
        meta_data = await self._blob_get_json(meta_k)
        if meta_data is None:
            raise MemoryNotFoundError(f"Conversation '{conversation_id}' not found")
        previous_meta = ConversationMeta.from_dict(meta_data)

        import asyncio

        # 删消息本体 + 二级索引 (parallel)
        await asyncio.gather(
            self._blob_delete(msg_key),
            self._blob_delete(_message_index_key(message_id)),
        )

        # 重算 meta：扫剩余消息得到最新 last_message_at；空则回退到 created_at
        next_meta = await self._recalculate_conversation_meta(encoded_cid, previous_meta)

        # 更新 meta + conversation_index (parallel)
        new_index_key = _index_key(next_meta.last_message_at, encoded_cid)
        old_index_key = _index_key(previous_meta.last_message_at, encoded_cid)
        write_ops: list = [self._blob_set_json(meta_k, next_meta.to_dict())]
        if new_index_key != old_index_key:
            write_ops.append(self._blob_set_json(
                new_index_key,
                {
                    "conversationId": conversation_id,
                    "lastMessageAt": next_meta.last_message_at,
                },
            ))
            write_ops.append(self._blob_delete(old_index_key))
        await asyncio.gather(*write_ops)

    async def _recalculate_conversation_meta(
        self,
        encoded_cid: str,
        previous_meta: ConversationMeta,
    ) -> ConversationMeta:
        """重新计算 conversation 的 message_count / last_message_at。

        删除单条消息后调用 —— 扫剩余 message keys，从 key 中解析 created_at
        （key 格式: conversations/{cid}/messages/{created_at}_{msg_id}）；
        没消息时 last_message_at 回退到 created_at。
        """
        prefix = _messages_prefix(encoded_cid)
        keys = await self._blob_list_keys(prefix)

        latest_created_at = 0
        for key in keys:
            # key 末段是 "{created_at}_{message_id}"，直接解析 timestamp
            segment = key.rsplit("/", 1)[-1]
            try:
                ts = int(segment.split("_", 1)[0])
            except (ValueError, IndexError):
                ts = 0
            if ts > latest_created_at:
                latest_created_at = ts

        return ConversationMeta(
            conversation_id=previous_meta.conversation_id,
            created_at=previous_meta.created_at,
            last_message_at=latest_created_at if keys else previous_meta.created_at,
            message_count=len(keys),
            metadata=previous_meta.metadata,
        )

    async def clear_messages(self, conversation_id: str) -> None:
        """Delete all messages in a conversation, **保留 meta**。

        语义对齐 Node ``memory.clearMessages``：清空消息后 ``get_conversation``
        仍然返回 meta（``message_count=0``、``last_message_at=created_at``）。
        若要彻底删除请用 ``delete_conversation``。

        Raises:
            MemoryNotFoundError: conversation 不存在。
        """
        import asyncio

        self._validate_conversation_id(conversation_id)

        encoded_cid = _encode_cid(conversation_id)
        meta_k = _meta_key(encoded_cid)
        meta_data = await self._blob_get_json(meta_k)

        if meta_data is None:
            raise MemoryNotFoundError(f"Conversation '{conversation_id}' not found")

        previous_meta = ConversationMeta.from_dict(meta_data)

        # Delete all message bodies + message_index entries in parallel.
        # Extract messageId from key name (no need to read message body).
        prefix = _messages_prefix(encoded_cid)
        keys = await self._blob_list_keys(prefix)

        delete_ops: list = []
        for key in keys:
            delete_ops.append(self._blob_delete(key))
            msg_id = _extract_message_id_from_key(key)
            if msg_id:
                delete_ops.append(self._blob_delete(_message_index_key(msg_id)))
        if delete_ops:
            await asyncio.gather(*delete_ops)

        # 重置 meta + 更新 conversation_index in parallel
        next_meta = ConversationMeta(
            conversation_id=previous_meta.conversation_id,
            created_at=previous_meta.created_at,
            last_message_at=previous_meta.created_at,
            message_count=0,
            metadata=previous_meta.metadata,
        )
        new_index_key = _index_key(next_meta.last_message_at, encoded_cid)
        old_index_key = _index_key(previous_meta.last_message_at, encoded_cid)

        write_ops: list = [self._blob_set_json(meta_k, next_meta.to_dict())]
        if new_index_key != old_index_key:
            write_ops.append(self._blob_set_json(
                new_index_key,
                {
                    "conversationId": conversation_id,
                    "lastMessageAt": next_meta.last_message_at,
                },
            ))
            write_ops.append(self._blob_delete(old_index_key))
        await asyncio.gather(*write_ops)

    async def list_conversations(
        self,
        limit: int = _DEFAULT_LIMIT,
        order: str = "desc",
        after: Optional[str] = None,
        before: Optional[str] = None,
        user_id: Optional[str] = None,
    ) -> ListConversationsResult:
        """List conversations, ordered by last_message_at.

        Args:
            limit: Max conversations to return (default 20, max 100).
            order: 'desc' (most recent first, default) or 'asc'.
            after: Opaque cursor from a previous ``next_cursor``.
            before: Opaque cursor from a previous ``previous_cursor``.
            user_id: If provided, only list conversations indexed under this user.

        Returns:
            ListConversationsResult with items, next_cursor, previous_cursor.
        """
        self._validate_limit(limit)
        if order not in ("asc", "desc"):
            raise MemoryValidationError("order must be 'asc' or 'desc'")
        self._assert_single_cursor(after, before)
        if user_id is not None:
            self._validate_user_id(user_id)

        # Determine prefix (global vs user-scoped)
        if user_id is not None:
            encoded_user_id = _encode_segment(user_id)
            prefix = _user_index_prefix(encoded_user_id)
        else:
            prefix = _index_prefix()

        # Optimization: in the default case (desc order, no before cursor, using after or fresh),
        # blob list's natural lexicographic order = most-recent-first (reversed timestamp),
        # so we can pass limit directly to avoid full scan.
        can_use_sdk_limit = (order == "desc" and before is None)

        if can_use_sdk_limit:
            # Request limit+1 extra to detect hasMore; if using after cursor,
            # we can't easily skip via SDK cursor, so fetch more and filter client-side.
            # For the common first-page case (no after), this is optimal.
            fetch_limit = limit + 1 if after is None else None
            keys = await self._blob_list_keys(prefix, consistency="strong", limit=fetch_limit)
        else:
            keys = await self._blob_list_keys(prefix, consistency="strong")

        # Default order is desc (most recent first). With revTs, ascending lex order
        # already means most-recent-first, so desc = natural order, asc = reversed.
        if order == "asc":
            keys = list(reversed(keys))

        # Apply cursor-based boundary filtering
        prefix_len = len(prefix)
        if after is not None:
            cursor_data = _decode_cursor(after)
            cursor_sk = _cursor_sort_key(cursor_data["lastMessageAt"], cursor_data["conversationId"])
            keys = self._apply_after_cursor(keys, prefix_len, cursor_sk, order)

        if before is not None:
            cursor_data = _decode_cursor(before)
            cursor_sk = _cursor_sort_key(cursor_data["lastMessageAt"], cursor_data["conversationId"])
            keys = self._apply_before_cursor(keys, prefix_len, cursor_sk, order)

        # Parse encoded_cids from key basenames, take limit+1 to detect next page
        import asyncio

        candidate_cids: List[str] = []
        for key in keys:
            if len(candidate_cids) > limit:
                break
            remainder = key[prefix_len:]
            sep_idx = remainder.find("_")
            if sep_idx < 0:
                continue
            candidate_cids.append(remainder[sep_idx + 1:])

        # Fetch metas in parallel (aligned with Node Promise.all approach)
        meta_results = await asyncio.gather(
            *[self._blob_get_json(_meta_key(cid), consistency="strong") for cid in candidate_cids]
        )

        items: List[ConversationMeta] = []
        has_more = False
        for meta_data in meta_results:
            if len(items) >= limit:
                has_more = True
                break
            if meta_data:
                try:
                    items.append(ConversationMeta.from_dict(meta_data))
                except (KeyError, TypeError):
                    continue

        # Build cursors
        next_cursor: Optional[str] = None
        previous_cursor: Optional[str] = None
        if items:
            if has_more:
                last = items[-1]
                next_cursor = _encode_cursor(last.last_message_at, last.conversation_id)
            if after is not None:
                first = items[0]
                previous_cursor = _encode_cursor(first.last_message_at, first.conversation_id)

        return ListConversationsResult(
            items=items,
            next_cursor=next_cursor,
            previous_cursor=previous_cursor,
        )

    @staticmethod
    def _apply_after_cursor(
        keys: List[str], prefix_len: int, cursor_sk: str, order: str
    ) -> List[str]:
        """Skip keys up to and including the cursor position."""
        # In desc order (natural revTs ascending): cursor_sk should be compared
        # with the basename of each key. We skip all items <= cursor for desc,
        # and all items >= cursor for asc.
        result = []
        found = False
        for key in keys:
            basename = key[prefix_len:]
            if not found:
                if order == "desc":
                    # desc: natural lex asc. "after" means skip items with
                    # sort_key <= cursor_sk (i.e. more recent or equal)
                    if basename > cursor_sk:
                        found = True
                        result.append(key)
                else:
                    # asc: reversed lex. "after" means skip items with
                    # sort_key >= cursor_sk (i.e. older or equal, since list is reversed)
                    if basename < cursor_sk:
                        found = True
                        result.append(key)
            else:
                result.append(key)
        return result

    @staticmethod
    def _apply_before_cursor(
        keys: List[str], prefix_len: int, cursor_sk: str, order: str
    ) -> List[str]:
        """Take keys only before the cursor position."""
        result = []
        for key in keys:
            basename = key[prefix_len:]
            if order == "desc":
                # desc: natural lex asc. "before" means stop at sort_key >= cursor_sk
                if basename >= cursor_sk:
                    break
            else:
                # asc: reversed lex. "before" means stop at sort_key <= cursor_sk
                if basename <= cursor_sk:
                    break
            result.append(key)
        return result

    async def delete_conversation(self, conversation_id: str) -> None:
        """Delete a conversation and all its messages.

        Raises:
            MemoryNotFoundError: Conversation doesn't exist.
        """
        import asyncio

        self._validate_conversation_id(conversation_id)

        encoded_cid = _encode_cid(conversation_id)
        meta_k = _meta_key(encoded_cid)

        # Read meta + list messages in parallel
        prefix = _messages_prefix(encoded_cid)
        meta_data, keys = await asyncio.gather(
            self._blob_get_json(meta_k),
            self._blob_list_keys(prefix),
        )

        if meta_data is None:
            raise MemoryNotFoundError(f"Conversation '{conversation_id}' not found")

        meta = ConversationMeta.from_dict(meta_data)

        # Delete all message bodies + message_index entries in parallel.
        # Extract messageId from key name (no need to read message body).
        delete_ops: list = []
        for key in keys:
            delete_ops.append(self._blob_delete(key))
            msg_id = _extract_message_id_from_key(key)
            if msg_id:
                delete_ops.append(self._blob_delete(_message_index_key(msg_id)))

        # Delete meta + conversation index entry
        delete_ops.append(self._blob_delete(meta_k))
        delete_ops.append(self._blob_delete(_index_key(meta.last_message_at, encoded_cid)))

        await asyncio.gather(*delete_ops)

    async def update_conversation(
        self, conversation_id: str, metadata: dict
    ) -> ConversationMeta:
        """Update conversation metadata (only metadata field is writable).

        语义对齐 Node ``memory.updateConversation``：传入的 ``metadata`` 与原
        metadata 做**浅合并**（同 key 覆盖、新 key 追加），而非整体替换。

        Args:
            conversation_id: Conversation identifier.
            metadata: 用于浅合并的 metadata dict。

        Returns:
            Updated ConversationMeta.

        Raises:
            MemoryNotFoundError: Conversation doesn't exist.
            MemoryValidationError: metadata is not a dict.
        """
        self._validate_conversation_id(conversation_id)
        if not isinstance(metadata, dict):
            raise MemoryValidationError("metadata must be a dict for update_conversation")

        encoded_cid = _encode_cid(conversation_id)
        meta_k = _meta_key(encoded_cid)
        meta_data = await self._blob_get_json(meta_k)

        if meta_data is None:
            raise MemoryNotFoundError(f"Conversation '{conversation_id}' not found")

        meta = ConversationMeta.from_dict(meta_data)
        merged = dict(meta.metadata or {})
        merged.update(metadata)
        meta.metadata = merged if merged else None
        await self._blob_set_json(meta_k, meta.to_dict())
        return meta

    async def get_conversation(self, conversation_id: str) -> ConversationMeta:
        """Get conversation metadata.

        Raises:
            MemoryNotFoundError: Conversation doesn't exist.
        """
        self._validate_conversation_id(conversation_id)

        encoded_cid = _encode_cid(conversation_id)
        meta_k = _meta_key(encoded_cid)
        meta_data = await self._blob_get_json(meta_k)

        if meta_data is None:
            raise MemoryNotFoundError(f"Conversation '{conversation_id}' not found")

        return ConversationMeta.from_dict(meta_data)

    # ─── Framework Helpers ───

    @staticmethod
    def to_openai_input(messages: List[Message]) -> List[dict]:
        """Convert Messages to OpenAI-compatible format.

        Strips platform fields (message_id, created_at, metadata).
        Returns list of ``{"role": ..., "content": ...}`` dicts.
        """
        return [{"role": m.role, "content": m.content} for m in messages]

    @staticmethod
    def to_anthropic_messages(messages: List[Message]) -> List[dict]:
        """Convert Messages to Anthropic-compatible format.

        Strips platform fields (message_id, created_at, metadata).
        Returns list of ``{"role": ..., "content": ...}`` dicts.
        """
        return [{"role": m.role, "content": m.content} for m in messages]

    @property
    def langgraph_checkpointer(self) -> Any:
        """Lazy-constructed adapter implementing LangGraph BaseCheckpointSaver.

        Maps ``conversation_id`` → ``thread_id``.

        独立前缀 ``langgraph_checkpoints/`` 存储，与消息历史完全隔离，
        不影响 ``message_count`` / quota。
        """
        cached = getattr(self, "_langgraph_checkpointer_cached", None)
        if cached is not None:
            return cached
        adapter = _LangGraphCheckpointerAdapter(self)
        self._langgraph_checkpointer_cached = adapter
        return adapter

    @property
    def langgraph_store(self) -> Any:
        """LangGraph BaseStore adapter — provides get/put/search/batch/listNamespaces.

        独立前缀 ``langgraph_store/items/`` 存储，与消息历史和 checkpoint 完全隔离。
        存储布局与 Node ``context.memory.langgraphStore`` 完全一致，支持跨语言互操作。

        Usage::

            store = ctx.store.langgraph_store
            await store.put(["agent", "memories"], "facts", {"content": "..."})
            item = await store.get(["agent", "memories"], "facts")
            items = await store.search(["agent"])
        """
        cached = getattr(self, "_langgraph_store_cached", None)
        if cached is not None:
            return cached
        adapter = _LangGraphStoreAdapter(self)
        self._langgraph_store_cached = adapter
        return adapter

    def openai_session(self, session_id: str, *, max_items: int = 100) -> "_EdgeOneMemorySession":
        """Create an OpenAI Agents SDK Session backed by ctx.store.

        Usage with OpenAI Agents SDK::

            from agents import Agent, Runner

            async def handler(ctx):
                agent = Agent(name="Assistant", instructions="Reply concisely.")
                session = ctx.store.openai_session(ctx.conversation_id)

                result = await Runner.run(agent, user_input, session=session)
                return {"reply": result.final_output}

        The session implements the Agents SDK Session protocol:
          - get_items(): reads history from memory
          - add_items(): persists new items to memory
          - pop_item(): removes and returns the last item
          - clear_session(): clears all items

        Args:
            session_id: Conversation/session identifier (typically ctx.conversation_id).
            max_items: Maximum items to retrieve per get_items() call (default 100).
        """
        return _EdgeOneMemorySession(self, session_id, max_items=max_items)

    def session(self, session_id: str, *, max_items: int = 100) -> "_EdgeOneMemorySession":
        """Alias for :meth:`openai_session` (backward compatibility)."""
        return self.openai_session(session_id, max_items=max_items)

    def claude_session_store(self) -> "EdgeOneSessionStore":
        """Create a Claude Agent SDK SessionStore backed by EdgeOne blob storage.

        Returns a lazily-constructed singleton. The store implements the
        ``SessionStore`` protocol from ``claude_agent_sdk`` (append/load/
        list_sessions/delete/list_subkeys).

        Usage with Claude Agent SDK::

            from claude_agent_sdk import query, ClaudeAgentOptions

            async def handler(ctx):
                store = ctx.store.claude_session_store()
                async for msg in query(
                    prompt="Fix the bug in auth.py",
                    options=ClaudeAgentOptions(session_store=store),
                ):
                    ...

        Returns:
            EdgeOneSessionStore instance (singleton per ConversationMemory).
        """
        cached = getattr(self, "_claude_session_store_cached", None)
        if cached is not None:
            return cached
        store = EdgeOneSessionStore(self._blob)
        self._claude_session_store_cached = store
        return store


# ─── LangGraph Store Adapter ───

try:
    from langgraph.store.base import BaseStore as _LangGraphBaseStore
except ImportError:
    _LangGraphBaseStore = object  # type: ignore[misc,assignment]


class _LangGraphStoreAdapter(_LangGraphBaseStore):
    """LangGraph BaseStore 适配器 — 对齐 Node ``createLangGraphStore``。

    存储布局：
        langgraph_store/items/{base64url(ns1)}/{base64url(ns2)}/{__key__}/{base64url(key)}

    只需实现 batch / abatch，BaseStore 的 get/put/search/delete/list_namespaces
    都有默认实现（内部调 self.batch）。
    """

    def __init__(self, memory: "ConversationMemory") -> None:
        if _LangGraphBaseStore is not object:
            super().__init__()
        self._memory = memory

    # ─── Key helpers ───

    @staticmethod
    def _namespace_to_path(namespace) -> str:
        return "/".join(_encode_path_segment(str(s)) for s in namespace)

    @classmethod
    def _item_key(cls, namespace, key: str) -> str:
        prefix = f"{_LANGGRAPH_STORE_PREFIX}/items/"
        ns_path = cls._namespace_to_path(namespace)
        encoded_key = _encode_path_segment(str(key))
        if ns_path:
            return f"{prefix}{ns_path}/{_LANGGRAPH_STORE_KEY_SEPARATOR}/{encoded_key}"
        return f"{prefix}{_LANGGRAPH_STORE_KEY_SEPARATOR}/{encoded_key}"

    @classmethod
    def _search_prefix(cls, namespace_prefix) -> str:
        prefix = f"{_LANGGRAPH_STORE_PREFIX}/items/"
        ns_path = cls._namespace_to_path(namespace_prefix)
        return f"{prefix}{ns_path}/" if ns_path else prefix

    # ─── Internal helpers ───

    async def _parse_stored_item(self, blob_key: str) -> Optional[dict]:
        try:
            raw = await self._memory._blob.get(blob_key, type="text", consistency="strong")
            if raw is None:
                return None
            item = json.loads(raw)
        except Exception:
            return None
        if (
            not item
            or not isinstance(item, dict)
            or not isinstance(item.get("value"), dict)
            or not isinstance(item.get("key"), str)
            or not isinstance(item.get("namespace"), list)
            or not isinstance(item.get("createdAt"), (int, float))
            or not isinstance(item.get("updatedAt"), (int, float))
        ):
            return None
        return item

    @staticmethod
    def _to_public_item(item: dict) -> Any:
        """Convert internal item to LangGraph Item (if available) or plain dict."""
        from datetime import datetime, timezone
        try:
            from langgraph.store.base import Item
            return Item(
                value=item["value"],
                key=item["key"],
                namespace=tuple(str(s) for s in item["namespace"]),
                created_at=datetime.fromtimestamp(item["createdAt"] / 1000, tz=timezone.utc),
                updated_at=datetime.fromtimestamp(item["updatedAt"] / 1000, tz=timezone.utc),
            )
        except ImportError:
            return {
                "value": item["value"],
                "key": item["key"],
                "namespace": tuple(str(s) for s in item["namespace"]),
                "created_at": datetime.fromtimestamp(item["createdAt"] / 1000, tz=timezone.utc),
                "updated_at": datetime.fromtimestamp(item["updatedAt"] / 1000, tz=timezone.utc),
            }

    @staticmethod
    def _namespace_starts_with(namespace, prefix) -> bool:
        ns = list(namespace)
        pf = list(prefix)
        if len(pf) > len(ns):
            return False
        return all(ns[i] == pf[i] for i in range(len(pf)))

    @staticmethod
    def _namespace_matches_path(namespace, path, direction: str) -> bool:
        ns = list(namespace)
        p = list(path)
        if len(p) > len(ns):
            return False
        offset = len(ns) - len(p) if direction == "suffix" else 0
        return all(
            seg == "*" or ns[offset + i] == seg
            for i, seg in enumerate(p)
        )

    @staticmethod
    def _compare_filter_value(item_value: Any, filter_value: Any) -> bool:
        if isinstance(filter_value, dict):
            operators = list(filter_value.keys())
            valid_ops = {"$eq", "$ne", "$gt", "$gte", "$lt", "$lte", "$in", "$nin"}
            if operators and all(op in valid_ops for op in operators):
                for op in operators:
                    v = filter_value[op]
                    if op == "$eq" and item_value != v:
                        return False
                    elif op == "$ne" and item_value == v:
                        return False
                    elif op == "$gt" and not (float(item_value or 0) > float(v or 0)):
                        return False
                    elif op == "$gte" and not (float(item_value or 0) >= float(v or 0)):
                        return False
                    elif op == "$lt" and not (float(item_value or 0) < float(v or 0)):
                        return False
                    elif op == "$lte" and not (float(item_value or 0) <= float(v or 0)):
                        return False
                    elif op == "$in" and (not isinstance(v, list) or item_value not in v):
                        return False
                    elif op == "$nin" and isinstance(v, list) and item_value in v:
                        return False
                return True
        return item_value == filter_value

    @classmethod
    def _matches_filter(cls, item: dict, filter_: Optional[dict]) -> bool:
        if not filter_:
            return True
        value = item.get("value", {})
        return all(
            cls._compare_filter_value(value.get(k), v)
            for k, v in filter_.items()
        )

    # ─── Core: abatch (async) ───

    async def abatch(self, ops) -> list:
        """Execute operations asynchronously. This is the core implementation."""
        results: list = []
        for op in ops:
            op_type = type(op).__name__
            if op_type == "GetOp":
                ns = list(op.namespace)
                blob_key = self._item_key(ns, str(op.key))
                item = await self._parse_stored_item(blob_key)
                result = self._to_public_item(item) if item else None
                results.append(result)
            elif op_type == "PutOp":
                ns = list(op.namespace)
                k = str(op.key)
                blob_key = self._item_key(ns, k)
                if op.value is None:
                    await self._memory._blob_delete(blob_key)
                else:
                    previous = await self._parse_stored_item(blob_key)
                    now = int(time.time() * 1000)
                    stored = {
                        "value": op.value,
                        "key": k,
                        "namespace": ns,
                        "createdAt": previous["createdAt"] if previous else now,
                        "updatedAt": now,
                    }
                    await self._memory._blob_set_json(blob_key, stored)
                results.append(None)
            elif op_type == "SearchOp":
                ns_prefix = list(op.namespace_prefix)
                prefix = self._search_prefix(ns_prefix)
                listed = await self._memory._blob.list(prefix=prefix, consistency="strong")
                keys = [blob.key for blob in listed.blobs]
                # Filter keys that contain the separator, then parallel fetch
                valid_keys = [bk for bk in keys if f"/{_LANGGRAPH_STORE_KEY_SEPARATOR}/" in bk]
                items_raw = await asyncio.gather(*[self._parse_stored_item(bk) for bk in valid_keys])
                items: list[dict] = [
                    it for it in items_raw
                    if it and self._namespace_starts_with(it["namespace"], ns_prefix)
                ]
                filtered = [it for it in items if self._matches_filter(it, getattr(op, "filter", None))]
                filtered.sort(key=lambda it: ("/".join(it["namespace"]), it["key"]))
                offset = getattr(op, "offset", 0) or 0
                limit = getattr(op, "limit", 10) or 10
                paged = filtered[offset: offset + limit]
                try:
                    from langgraph.store.base import SearchItem
                    from datetime import datetime, timezone
                    results.append([
                        SearchItem(
                            value=it["value"],
                            key=it["key"],
                            namespace=tuple(str(s) for s in it["namespace"]),
                            created_at=datetime.fromtimestamp(it["createdAt"] / 1000, tz=timezone.utc),
                            updated_at=datetime.fromtimestamp(it["updatedAt"] / 1000, tz=timezone.utc),
                        )
                        for it in paged
                    ])
                except ImportError:
                    results.append([self._to_public_item(it) for it in paged])
            elif op_type == "ListNamespacesOp":
                all_prefix = f"{_LANGGRAPH_STORE_PREFIX}/items/"
                listed = await self._memory._blob.list(prefix=all_prefix, consistency="strong")
                keys = [blob.key for blob in listed.blobs]
                # Filter keys with separator, then parallel fetch
                valid_keys = [bk for bk in keys if f"/{_LANGGRAPH_STORE_KEY_SEPARATOR}/" in bk]
                items_raw = await asyncio.gather(*[self._parse_stored_item(bk) for bk in valid_keys])

                namespaces: set[str] = set()
                for it in items_raw:
                    if not it:
                        continue
                    ns = [str(s) for s in it["namespace"]]
                    conditions = getattr(op, "match_conditions", None) or []
                    matches = all(
                        self._namespace_matches_path(
                            ns,
                            [str(s) for s in (cond.path if hasattr(cond, 'path') else [])],
                            getattr(cond, "match_type", "prefix"),
                        )
                        for cond in conditions
                    )
                    if not matches:
                        continue
                    max_depth = getattr(op, "max_depth", None)
                    if max_depth is not None and max_depth > 0:
                        ns = ns[:max_depth]
                    namespaces.add(json.dumps(ns))
                result_ns = [tuple(json.loads(s)) for s in namespaces]
                result_ns.sort(key=lambda ns: "/".join(ns))
                offset = getattr(op, "offset", 0) or 0
                limit = getattr(op, "limit", 100) or 100
                results.append(result_ns[offset: offset + limit])
            else:
                results.append(None)
        return results

    # ─── Core: batch (sync) — bridges to abatch ───

    # Shared worker thread + event loop for sync→async bridge.
    # Avoids creating/destroying ThreadPoolExecutor on every batch call.
    _worker_loop: Any = None
    _worker_thread: Any = None

    @classmethod
    def _get_worker_loop(cls):
        """Get or create a shared background event loop for sync batch calls."""
        import asyncio
        import threading
        if cls._worker_loop is None or cls._worker_loop.is_closed():
            cls._worker_loop = asyncio.new_event_loop()
            cls._worker_thread = threading.Thread(
                target=cls._worker_loop.run_forever, daemon=True, name="langgraph-store-worker"
            )
            cls._worker_thread.start()
        return cls._worker_loop

    def batch(self, ops) -> list:
        """Execute operations synchronously. Bridges to abatch."""
        import asyncio
        import threading

        ops_list = list(ops)

        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = None

        if loop and loop.is_running():
            loop_thread_id = getattr(loop, '_thread_id', None)
            if loop_thread_id and loop_thread_id != threading.current_thread().ident:
                # Called from worker thread — safe to submit to main loop
                future = asyncio.run_coroutine_threadsafe(self.abatch(ops_list), loop)
                return future.result(timeout=60)
            else:
                # Called from main loop thread — use shared background loop
                worker_loop = self._get_worker_loop()
                future = asyncio.run_coroutine_threadsafe(self.abatch(ops_list), worker_loop)
                return future.result(timeout=60)
        return asyncio.run(self.abatch(ops_list))

    def start(self) -> None:
        pass

    def stop(self) -> None:
        pass


# ─── LangGraph Checkpointer Adapter ───

try:
    from langgraph.checkpoint.base import BaseCheckpointSaver as _BaseCheckpointSaver
except ImportError:
    _BaseCheckpointSaver = object  # type: ignore[misc,assignment]

try:
    from langgraph.checkpoint.base import CheckpointTuple as _CheckpointTuple
except ImportError:
    _CheckpointTuple = None  # type: ignore[misc,assignment]


def _to_checkpoint_tuple(data: dict) -> Any:
    """Convert a stored dict to a CheckpointTuple (if available) or a namespace object."""
    import pickle

    # Restore channel_values from pickle if stored separately
    checkpoint = dict(data.get("checkpoint") or {})
    channel_values_b64 = data.get("channel_values_pickle")
    if channel_values_b64 and "channel_values" not in checkpoint:
        try:
            cv = pickle.loads(base64.b64decode(channel_values_b64))
            checkpoint["channel_values"] = cv
        except Exception:
            checkpoint["channel_values"] = {}

    # Clean config: remove LangGraph internal keys that were serialized as None.
    # When these keys exist with None value, LangGraph uses None instead of
    # DEFAULT_RUNTIME, causing 'NoneType' has no attribute 'override' errors.
    config = data.get("config") or {}
    configurable = config.get("configurable") or {}
    _INTERNAL_KEYS_TO_STRIP = (
        "__pregel_runtime",
        "__pregel_store",
        "__pregel_checkpointer",
        "__pregel_cache",
    )
    for key in _INTERNAL_KEYS_TO_STRIP:
        configurable.pop(key, None)
    if configurable:
        config["configurable"] = configurable

    parent_config = data.get("parent_config")
    if isinstance(parent_config, dict):
        parent_configurable = (parent_config.get("configurable") or {})
        for key in _INTERNAL_KEYS_TO_STRIP:
            parent_configurable.pop(key, None)

    # Clean config: remove LangGraph internal keys that were serialized as None.
    # When these keys exist with None value, LangGraph uses None instead of
    # DEFAULT_RUNTIME, causing 'NoneType' has no attribute 'override' errors.
    config = data.get("config") or {}
    configurable = config.get("configurable") or {}
    _INTERNAL_KEYS_TO_STRIP = (
        "__pregel_runtime",
        "__pregel_store",
        "__pregel_checkpointer",
        "__pregel_cache",
    )
    for key in _INTERNAL_KEYS_TO_STRIP:
        configurable.pop(key, None)
    if configurable:
        config["configurable"] = configurable

    parent_config = data.get("parent_config")
    if isinstance(parent_config, dict):
        parent_configurable = (parent_config.get("configurable") or {})
        for key in _INTERNAL_KEYS_TO_STRIP:
            parent_configurable.pop(key, None)

    if _CheckpointTuple is not None:
        return _CheckpointTuple(
            config=config,
            checkpoint=checkpoint,
            metadata=data.get("metadata"),
            parent_config=parent_config,
            pending_writes=data.get("pending_writes"),
        )
    # Fallback: return a simple namespace so .checkpoint / .config attribute access works
    class _Ns:
        pass
    ns = _Ns()
    ns.config = config
    ns.checkpoint = checkpoint
    ns.metadata = data.get("metadata")
    ns.parent_config = parent_config
    ns.pending_writes = data.get("pending_writes")
    return ns


class _LangGraphCheckpointerAdapter(_BaseCheckpointSaver):
    """LangGraph checkpoint saver — 独立前缀存储，对齐 Node ``createLangGraphCheckpointer``。

    存储布局：

        langgraph_checkpoints/{thread_id}/checkpoints/{checkpoint_id}     单个 checkpoint
        langgraph_checkpoints/{thread_id}/latest                          最新 checkpoint id
        langgraph_checkpoints/{thread_id}/writes/{checkpoint_id}/{task_id}   pending writes

    所有 thread_id / checkpoint_id / task_id 都做 URL-encode，避免出现 ``/`` 等
    特殊字符破坏 key 结构。

    LangGraph 自身有 sync 和 async 两套 API，这里同时实现两套：
      - ``aget_tuple`` / ``aput`` / ``aput_writes`` / ``alist`` / ``adelete_thread``：
        async 版本（LangGraph BaseCheckpointSaver async）。
      - ``get_tuple`` / ``put`` / ``put_writes`` / ``list`` / ``delete_thread``：
        sync 别名（LangGraph 在 sync 路径下会调；底层仍调 async，使用方需在
        async 上下文里）。

    与 Node 行为差异：Node 一份代码同时挂 camelCase / snake_case 别名；这里
    Python 全部 snake_case，并保持 a*前缀的 async 名（Python 生态约定）。
    """

    def __init__(self, memory: ConversationMemory) -> None:
        if _BaseCheckpointSaver is not object:
            super().__init__()
        self._memory = memory

    # ─── Key helpers ───

    @staticmethod
    def _thread_base(thread_id: str, checkpoint_ns: str = "") -> str:
        base = f"{_LANGGRAPH_CHECKPOINT_PREFIX}/{_encode_segment(thread_id)}"
        if checkpoint_ns:
            return f"{base}/namespaces/{_encode_segment(checkpoint_ns)}"
        return base

    @classmethod
    def _checkpoint_key(cls, thread_id: str, checkpoint_id: str, checkpoint_ns: str = "") -> str:
        return f"{cls._thread_base(thread_id, checkpoint_ns)}/checkpoints/{_encode_segment(checkpoint_id)}"

    @classmethod
    def _latest_key(cls, thread_id: str, checkpoint_ns: str = "") -> str:
        return f"{cls._thread_base(thread_id, checkpoint_ns)}/latest"

    @classmethod
    def _writes_key(cls, thread_id: str, checkpoint_id: str, task_id: str, checkpoint_ns: str = "") -> str:
        return f"{cls._thread_base(thread_id, checkpoint_ns)}/writes/{_encode_segment(checkpoint_id)}/{_encode_segment(task_id)}"

    @classmethod
    def _thread_prefix(cls, thread_id: str, checkpoint_ns: str = "") -> str:
        return f"{cls._thread_base(thread_id, checkpoint_ns)}/"

    @classmethod
    def _checkpoints_prefix(cls, thread_id: str, checkpoint_ns: str = "") -> str:
        return f"{cls._thread_base(thread_id, checkpoint_ns)}/checkpoints/"

    @staticmethod
    def _resolve_thread_id(config: Any) -> str:
        configurable = (config or {}).get("configurable") or {}
        thread_id = configurable.get("thread_id") or configurable.get("threadId")
        if not thread_id:
            raise MemoryValidationError(
                "LangGraph checkpoint config requires configurable.thread_id"
            )
        return str(thread_id)

    @staticmethod
    def _resolve_checkpoint_ns(config: Any) -> str:
        configurable = (config or {}).get("configurable") or {}
        return str(configurable.get("checkpoint_ns") or configurable.get("checkpointNs") or "")

    @staticmethod
    def _resolve_checkpoint_id(config: Any, checkpoint: Any = None) -> str:
        if isinstance(checkpoint, dict) and checkpoint.get("id"):
            return str(checkpoint["id"])
        configurable = (config or {}).get("configurable") or {}
        cid = configurable.get("checkpoint_id") or configurable.get("checkpointId")
        if cid:
            return str(cid)
        return uuid.uuid4().hex

    # ─── Async API（LangGraph BaseCheckpointSaver async 接口） ───

    async def aget_tuple(self, config: Any) -> Optional[Any]:
        """Get a checkpoint tuple by thread_id (+ optional checkpoint_id).

        - 没传 ``checkpoint_id``：返回 thread 的 latest。
        - 传了 ``checkpoint_id``：返回指定 checkpoint。
        - 找不到：返回 ``None``。

        返回 CheckpointTuple。pending_writes 从独立存储的 writes 中重建，
        确保 LangGraph 的 aget_delta_channel_history 能正确遍历 parent chain
        收集 DeltaChannel 数据。
        """
        thread_id = self._resolve_thread_id(config)
        checkpoint_ns = self._resolve_checkpoint_ns(config)
        configurable = (config or {}).get("configurable") or {}
        checkpoint_id = configurable.get("checkpoint_id") or configurable.get("checkpointId")
        if not checkpoint_id:
            checkpoint_id = await self._read_latest_id(thread_id, checkpoint_ns)
        if not checkpoint_id:
            return None

        data = await self._memory._blob_get_json(
            self._checkpoint_key(thread_id, str(checkpoint_id), checkpoint_ns),
            consistency="strong",
        )
        if data is None:
            return None

        # Prevent infinite parent chain loop: if parent_config points to
        # the same checkpoint_id, nullify it to terminate the chain.
        parent_config = data.get("parent_config")
        if isinstance(parent_config, dict):
            parent_configurable = parent_config.get("configurable") or {}
            parent_ckpt_id = parent_configurable.get("checkpoint_id") or parent_configurable.get("checkpointId")
            if not parent_ckpt_id or parent_ckpt_id == checkpoint_id:
                data["parent_config"] = None

        # Rebuild pending_writes from stored writes for this checkpoint.
        # LangGraph's aget_delta_channel_history expects pending_writes as
        # [(task_id, channel_name, value), ...] tuples.
        pending_writes = await self._load_pending_writes(thread_id, str(checkpoint_id), checkpoint_ns)
        data["pending_writes"] = pending_writes

        return _to_checkpoint_tuple(data)

    async def _load_pending_writes(self, thread_id: str, checkpoint_id: str, checkpoint_ns: str = "") -> list:
        """Load all writes for this thread and convert to pending_writes format.

        LangGraph's DeltaChannel writes are stored under various checkpoint_ids
        (the parent checkpoint id at the time of writing). We load ALL writes
        for the thread so aget_delta_channel_history can find them.
        Uses concurrent reads for performance.
        """
        import asyncio
        import pickle
        # Only load writes for this specific checkpoint_id, not all writes in the thread.
        # This matches Node behavior where pending_writes are scoped to a checkpoint.
        prefix = f"{self._thread_base(thread_id, checkpoint_ns)}/writes/{_encode_segment(checkpoint_id)}/"
        try:
            listed = await self._memory._blob.list(prefix=prefix, consistency="strong")
            keys = [blob.key for blob in listed.blobs]
        except Exception:
            return []

        if not keys:
            return []

        # Concurrent read all writes
        async def _read_write(key: str):
            try:
                raw = await self._memory._blob.get(key, type="text", consistency="strong")
                if raw is None:
                    return None
                return json.loads(raw)
            except Exception:
                return None

        results = await asyncio.gather(*[_read_write(k) for k in keys])

        pending_writes = []
        for wd in results:
            if not isinstance(wd, dict):
                continue
            task_id = wd.get("task_id", "")

            # New format: writes serialized via pickle+base64
            writes_pickle = wd.get("writes_pickle")
            if writes_pickle:
                try:
                    writes = pickle.loads(base64.b64decode(writes_pickle))
                    for channel_name, channel_value in writes:
                        pending_writes.append((task_id, channel_name, channel_value))
                    continue
                except Exception:
                    pass

            # Legacy format: writes stored as plain JSON
            writes_legacy = wd.get("writes")
            if writes_legacy:
                for channel_name, channel_value in writes_legacy:
                    pending_writes.append((task_id, channel_name, channel_value))

        return pending_writes

    async def aput(
        self,
        config: Any,
        checkpoint: Any,
        metadata: Any = None,
        new_versions: Any = None,
    ) -> dict:
        """Persist a checkpoint and update ``latest`` pointer.

        返回更新后的 config（``configurable.checkpoint_id`` 注入实际 id），
        与 LangGraph 约定一致。
        """
        thread_id = self._resolve_thread_id(config)
        checkpoint_ns = self._resolve_checkpoint_ns(config)
        checkpoint_id = self._resolve_checkpoint_id(config, checkpoint)

        next_configurable = dict((config or {}).get("configurable") or {})
        next_configurable["thread_id"] = thread_id
        next_configurable["checkpoint_ns"] = checkpoint_ns
        next_configurable["checkpoint_id"] = checkpoint_id
        next_config = dict(config or {})
        next_config["configurable"] = next_configurable

        ckpt_payload = dict(checkpoint or {})
        ckpt_payload["id"] = checkpoint_id

        # channel_values may contain non-JSON-serializable objects (Runtime, etc.)
        # Serialize them separately using pickle+base64.
        import pickle
        channel_values = ckpt_payload.pop("channel_values", None)
        channel_values_b64 = None
        if channel_values is not None:
            try:
                channel_values_b64 = base64.b64encode(pickle.dumps(channel_values)).decode("ascii")
            except Exception:
                channel_values_b64 = None

        await self._memory._blob_set_json(
            self._checkpoint_key(thread_id, checkpoint_id, checkpoint_ns),
            {
                "config": next_config,
                "checkpoint": ckpt_payload,
                "channel_values_pickle": channel_values_b64,
                "metadata": metadata,
                "parent_config": config,
                "pending_writes": [],
                "new_versions": new_versions,
            },
        )
        await self._memory._blob_set_json(self._latest_key(thread_id, checkpoint_ns), checkpoint_id)
        return next_config

    async def aput_writes(self, config: Any, writes: Any, task_id: Any) -> None:
        """Persist pending writes for a checkpoint task.

        等价于 LangGraph ``BaseCheckpointSaver.aput_writes``。如果 config 没
        给 checkpoint_id，则落到 latest；都没有就用 ``"pending"`` 兜底。

        writes 中包含 LangChain Message 对象等不可 JSON 安全 round-trip 的类型，
        用 pickle+base64 序列化以保留完整 Python 类型信息。
        """
        import pickle
        thread_id = self._resolve_thread_id(config)
        checkpoint_ns = self._resolve_checkpoint_ns(config)
        configurable = (config or {}).get("configurable") or {}
        checkpoint_id = (
            configurable.get("checkpoint_id")
            or configurable.get("checkpointId")
            or await self._read_latest_id(thread_id, checkpoint_ns)
            or "pending"
        )
        task_id_str = str(task_id) if task_id else uuid.uuid4().hex

        # Serialize writes using pickle to preserve Python object types
        writes_b64 = None
        try:
            writes_b64 = base64.b64encode(pickle.dumps(writes)).decode("ascii")
        except Exception:
            pass

        await self._memory._blob_set_json(
            self._writes_key(thread_id, str(checkpoint_id), task_id_str, checkpoint_ns),
            {
                "config": config,
                "writes_pickle": writes_b64,
                "task_id": task_id_str,
            },
        )

    async def alist(self, config: Any, *, limit: Optional[int] = None) -> list:
        """List all checkpoints for a thread, newest first."""
        thread_id = self._resolve_thread_id(config)
        checkpoint_ns = self._resolve_checkpoint_ns(config)
        keys = await self._memory._blob_list_keys(self._checkpoints_prefix(thread_id, checkpoint_ns))
        # keys 是字典序升序，倒序就是按 checkpoint_id 字典序的最新优先；
        # 真实业务里 checkpoint_id 一般是单调递增 UUID/timestamp，足够近似新→旧。
        keys = list(reversed(keys))
        if limit is not None and limit > 0:
            keys = keys[:limit]
        tuples = []
        for key in keys:
            data = await self._memory._blob_get_json(key)
            if data is not None:
                tuples.append(_to_checkpoint_tuple(data))
        return tuples

    async def adelete_thread(self, thread_id: str) -> None:
        """Delete every key under a thread (checkpoints + writes + latest)."""
        if not thread_id:
            raise MemoryValidationError("thread_id is required for delete_thread")
        keys = await self._memory._blob_list_keys(self._thread_prefix(thread_id))
        for key in keys:
            await self._memory._blob_delete(key)

    @staticmethod
    def get_next_version(current: Any, _channel: Any = None) -> int:
        """Compute next channel version. 与 Node ``getNextVersion`` 行为一致。

        LangGraph Python 的签名是 ``get_next_version(current, channel)``；
        我们忽略 channel，纯按 current+1 处理（Node 实现也只看 current）。
        """
        try:
            value = float(current) if current is not None else 0.0
        except (TypeError, ValueError):
            return 1
        if value != value or value in (float("inf"), float("-inf")):  # NaN/inf
            return 1
        return int(value) + 1

    # ─── Sync 别名（部分 LangGraph 集成会调 sync 名） ───
    #
    # LangGraph 的 sync API 实际是阻塞调用，用在非 async 上下文里。我们这里
    # 没有真正的 sync 后端（blob_store 全是 async），所以这些 sync 名只能在
    # 已有 event loop 的场景下被「await」。其作用主要是：
    #   1. 满足某些工具链按 ``hasattr(checkpointer, "get_tuple")`` 做 duck-typing；
    #   2. 业务侧可以拿到一个 awaitable，自己 ``await`` 即可。
    # 真要在纯同步环境里用，需要业务自己包 ``asyncio.run`` 之类。

    def get_tuple(self, config: Any) -> Any:
        return self._run_sync(self.aget_tuple(config))

    def put(self, config: Any, checkpoint: Any, metadata: Any = None, new_versions: Any = None) -> Any:
        return self._run_sync(self.aput(config, checkpoint, metadata, new_versions))

    def put_writes(self, config: Any, writes: Any, task_id: Any) -> Any:
        return self._run_sync(self.aput_writes(config, writes, task_id))

    def list(self, config: Any, *, limit: Optional[int] = None) -> Any:
        return self._run_sync(self.alist(config, limit=limit))

    def delete_thread(self, thread_id: str) -> Any:
        return self._run_sync(self.adelete_thread(thread_id))

    @staticmethod
    def _run_sync(coro) -> Any:
        """Run an async coroutine from a sync context."""
        import asyncio
        import threading
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = None

        if loop is None:
            return asyncio.run(coro)

        loop_thread_id = getattr(loop, '_thread_id', None)
        if loop_thread_id and loop_thread_id == threading.current_thread().ident:
            import concurrent.futures
            with concurrent.futures.ThreadPoolExecutor(1) as pool:
                return pool.submit(asyncio.run, coro).result(timeout=60)

        future = asyncio.run_coroutine_threadsafe(coro, loop)
        return future.result(timeout=60)

    # ─── Internal helpers ───

    async def _read_latest_id(self, thread_id: str, checkpoint_ns: str = "") -> Optional[str]:
        raw = await self._memory._blob_get_text(
            self._latest_key(thread_id, checkpoint_ns), consistency="strong"
        )
        if raw is None:
            return None
        # latest 写入时是 json.dumps("xxx")，读出来要解一次
        try:
            value = json.loads(raw)
        except (json.JSONDecodeError, ValueError):
            value = raw
        return str(value) if value else None


# ─── OpenAI Agents SDK Session Adapter ───


_SESSION_VALID_ROLES = {"user", "assistant", "system", "tool"}
_SESSION_METADATA_MARKER = {"agent_sdk_session": True}


class _EdgeOneMemorySession:
    """OpenAI Agents SDK Session protocol adapter backed by ctx.store.

    Runner.run(..., session=session) will automatically:
      1. Call get_items() to read history and prepend to current input
      2. Call add_items() to persist this turn's user / assistant / tool items

    This adapter stores each Agents SDK item as a memory message with
    metadata ``{"agent_sdk_session": True}`` to distinguish from manually
    written messages. The full item dict is stored as ``content``.
    """

    session_settings = None

    def __init__(self, memory: ConversationMemory, session_id: str, *, max_items: int = 100):
        self.memory = memory
        self.session_id = session_id
        self.max_items = max_items

    async def get_items(self, limit: int | None = None) -> list:
        """Retrieve conversation history as Agents SDK input items."""
        effective_limit = min(limit or self.max_items, 100)
        messages = await self.memory.get_messages(
            self.session_id,
            limit=effective_limit,
            order="asc",
        )

        items = []
        for message in messages:
            item = self._message_to_item(message)
            if item is not None:
                items.append(item)
        return items[-limit:] if limit is not None else items

    async def add_items(self, items: list) -> None:
        """Persist new Agents SDK items to memory."""
        if not items:
            return

        # Use bulk append for efficiency (1 meta update instead of N)
        messages_data = []
        for item in items:
            normalized = self._jsonable(item)
            role = self._role_for_item(normalized)
            messages_data.append({
                "role": role,
                "content": normalized,
                "metadata": {
                    **_SESSION_METADATA_MARKER,
                    "item_type": normalized.get("type") if isinstance(normalized, dict) else None,
                },
            })

        await self.memory._bulk_append_messages(self.session_id, messages_data)

    async def pop_item(self) -> Optional[dict]:
        """Remove and return the most recent item from the session."""
        messages = await self.memory.get_messages(self.session_id, limit=100, order="desc")
        for message in messages:
            item = self._message_to_item(message)
            if item is None:
                continue
            await self.memory.delete_message(self.session_id, message.message_id)
            return item
        return None

    async def clear_session(self) -> None:
        """Clear all items for this session."""
        await self.memory.clear_messages(self.session_id)

    def _message_to_item(self, message: Any) -> Optional[dict]:
        content = getattr(message, "content", None)
        role = getattr(message, "role", None)
        metadata = getattr(message, "metadata", None) or {}

        # New format: content holds the full Agents SDK input item.
        if metadata.get("agent_sdk_session") and isinstance(content, dict):
            return content

        # Legacy compat: plain text messages written by hand.
        if role in _SESSION_VALID_ROLES and content is not None:
            return {"role": role, "content": content}

        return None

    @staticmethod
    def _role_for_item(item: Any) -> str:
        """Determine the best role string for storage from an Agents SDK item."""
        if isinstance(item, dict):
            role = item.get("role")
            if role in _SESSION_VALID_ROLES:
                return role

            item_type = item.get("type")
            if item_type == "message":
                msg_role = item.get("role")
                return msg_role if msg_role in _SESSION_VALID_ROLES else "assistant"
            if item_type in ("function_call_output", "computer_call_output"):
                return "tool"
            if item_type in ("function_call", "computer_call", "reasoning"):
                return "assistant"

        return "tool"

    @staticmethod
    def _jsonable(item: Any) -> dict:
        """Convert an Agents SDK item to a JSON-serializable dict."""
        if isinstance(item, dict):
            return item
        if hasattr(item, "model_dump"):
            return item.model_dump(exclude_unset=True)
        if hasattr(item, "dict"):
            return item.dict(exclude_unset=True)
        return {"role": "tool", "content": str(item)}


# ─── Claude Agent SDK SessionStore Adapter ───


_CLAUDE_SESSION_PREFIX = "claude_sessions"


class EdgeOneSessionStore:
    """Claude Agent SDK SessionStore backed by EdgeOne blob storage.

    Implements the ``SessionStore`` protocol from ``claude_agent_sdk``:
      - append(key, entries): persist transcript entries
      - load(key): retrieve all entries in append order
      - list_sessions(project_key): enumerate sessions with mtime
      - delete(key): remove session (cascade) or single subpath
      - list_subkeys(key): discover subagent subpaths

    Storage layout (multi-part-file, aligned with S3 reference impl)::

        claude_sessions/{project_key}/{session_id}/parts/{ts13}_{rand6}
        claude_sessions/{project_key}/{session_id}/subpaths/{subpath}/parts/{ts13}_{rand6}

    Each ``append()`` creates one new part file containing JSONL (one entry
    per line). ``load()`` lists all part files in lexicographic order (which
    equals chronological order due to the 13-digit timestamp prefix), reads
    each, and concatenates entries.

    Usage::

        from claude_agent_sdk import query, ClaudeAgentOptions

        async def handler(ctx):
            store = ctx.store.claude_session_store()
            async for msg in query(
                prompt="Fix the bug",
                options=ClaudeAgentOptions(session_store=store),
            ):
                ...
    """

    def __init__(self, blob_store: Any) -> None:
        """
        Args:
            blob_store: Raw blob store instance. Must implement get(key, type=),
                        set(key, value), delete(key), list(prefix=).
        """
        self._blob = blob_store

    # ─── Key Helpers ───

    @staticmethod
    def _parts_prefix(key: dict) -> str:
        """Build the parts/ prefix for a given SessionKey."""
        base = (
            f"{_CLAUDE_SESSION_PREFIX}"
            f"/{_encode_segment(key['project_key'])}"
            f"/{_encode_segment(key['session_id'])}"
        )
        subpath = key.get("subpath")
        if subpath:
            base = f"{base}/subpaths/{_encode_segment(subpath)}"
        return f"{base}/parts/"

    @staticmethod
    def _session_base(project_key: str, session_id: str) -> str:
        """Build the base prefix for an entire session (for cascade delete)."""
        return (
            f"{_CLAUDE_SESSION_PREFIX}"
            f"/{_encode_segment(project_key)}"
            f"/{_encode_segment(session_id)}/"
        )

    @staticmethod
    def _project_prefix(project_key: str) -> str:
        """Build the prefix for all sessions in a project."""
        return f"{_CLAUDE_SESSION_PREFIX}/{_encode_segment(project_key)}/"

    # ─── Required Methods ───

    async def append(self, key: dict, entries: list) -> None:
        """Persist a batch of transcript entries as a new part file.

        Args:
            key: SessionKey dict with project_key, session_id, optional subpath.
            entries: List of JSON-serializable SessionStoreEntry objects.
        """
        if not entries:
            return
        prefix = self._parts_prefix(key)
        ts = str(int(time.time() * 1000)).zfill(13)
        rand = uuid.uuid4().hex[:6]
        part_key = f"{prefix}{ts}_{rand}"
        payload = "\n".join(
            json.dumps(e, ensure_ascii=False, separators=(",", ":"))
            for e in entries
        )
        await self._blob.set(part_key, payload)

    async def load(self, key: dict) -> Optional[list]:
        """Load all entries for a session/subpath in append order.

        Returns:
            List of entries in chronological order, or None if session unknown.
        """
        prefix = self._parts_prefix(key)
        keys = await self._list_keys(prefix)
        if not keys:
            return None
        keys.sort()  # 正序 = 时间顺序 = append 顺序
        entries: list = []
        for k in keys:
            raw = await self._blob.get(k, type="text")
            if raw:
                for line in raw.split("\n"):
                    line = line.strip()
                    if line:
                        try:
                            entries.append(json.loads(line))
                        except (json.JSONDecodeError, ValueError):
                            continue
        return entries if entries else None

    # ─── Optional Methods ───

    async def list_sessions(self, project_key: str) -> list:
        """Enumerate all sessions under a project with their last-modified time.

        Returns:
            List of {"session_id": str, "mtime": int} dicts.
        """
        prefix = self._project_prefix(project_key)
        keys = await self._list_keys(prefix)
        sessions: dict = {}  # session_id → max mtime
        for k in keys:
            remainder = k[len(prefix):]
            # remainder: {session_id}/parts/{ts}_{rand}
            # or:        {session_id}/subpaths/{sp}/parts/{ts}_{rand}
            sid_encoded = remainder.split("/")[0]
            sid = urllib.parse.unquote(sid_encoded)
            # Extract timestamp from part filename
            filename = k.rsplit("/", 1)[-1]
            ts_str = filename.split("_")[0]
            try:
                ts = int(ts_str)
            except (ValueError, IndexError):
                continue
            if sid not in sessions or ts > sessions[sid]:
                sessions[sid] = ts
        return [{"session_id": sid, "mtime": mtime} for sid, mtime in sessions.items()]

    async def delete(self, key: dict) -> None:
        """Delete session data.

        Without subpath: cascade-delete everything under the session.
        With subpath: delete only that subpath's parts.
        """
        subpath = key.get("subpath")
        if subpath:
            prefix = self._parts_prefix(key)
        else:
            prefix = self._session_base(key["project_key"], key["session_id"])
        keys = await self._list_keys(prefix)
        for k in keys:
            await self._blob.delete(k)

    async def list_subkeys(self, key: dict) -> list:
        """Discover subagent subpaths for a session.

        Returns:
            List of subpath strings (e.g. ["subagents/agent-abc", ...]).
        """
        base = (
            f"{_CLAUDE_SESSION_PREFIX}"
            f"/{_encode_segment(key['project_key'])}"
            f"/{_encode_segment(key['session_id'])}"
            f"/subpaths/"
        )
        keys = await self._list_keys(base)
        subpaths: set = set()
        for k in keys:
            remainder = k[len(base):]
            # remainder: {encoded_subpath}/parts/{ts}_{rand}
            segments = remainder.split("/")
            if segments:
                decoded = urllib.parse.unquote(segments[0])
                if decoded and decoded != "." and ".." not in decoded:
                    subpaths.add(decoded)
        return sorted(subpaths)

    # ─── Internal Helpers ───

    async def _list_keys(self, prefix: str) -> List[str]:
        """List all keys with given prefix."""
        result = await self._blob.list(prefix=prefix)
        return [blob.key for blob in result.blobs]