rayhpeng
439c10d6f2
Following the rename/delete bug fix in PR1, migrate the remaining direct LangGraph Store reads/writes in the threads router and services to the ThreadMetaStore abstraction so that the sqlite and memory backends behave identically and the legacy dual-write paths can be removed. Migrated endpoints (threads.py): - create_thread: idempotency check + write now use thread_meta_repo.get/create instead of dual-writing the LangGraph Store and the SQL row. - get_thread: reads from thread_meta_repo.get; the checkpoint-only fallback for legacy threads is preserved. - patch_thread: replaced _store_get/_store_put with thread_meta_repo.update_metadata. - delete_thread_data: dropped the legacy store.adelete; thread_meta_repo.delete already covers it. Removed dead code (services.py): - _upsert_thread_in_store — redundant with the immediately following thread_meta_repo.create() call. - _sync_thread_title_after_run — worker.py's finally block already syncs the title via thread_meta_repo.update_display_name() after each run. Removed dead code (threads.py): - _store_get / _store_put / _store_upsert helpers (no remaining callers). - THREADS_NS constant. - get_store import (router no longer touches the LangGraph Store directly). New abstract method: - ThreadMetaStore.update_metadata(thread_id, metadata) merges metadata into the thread's metadata field. Implemented in both ThreadMetaRepository (SQL, read-modify-write inside one session) and MemoryThreadMetaStore. Three new unit tests cover merge / empty / nonexistent behaviour. Net change: -134 lines. Full test suite: 1693 passed, 14 skipped. Verified end-to-end with curl in gateway mode against sqlite backend (create / patch / get / rename / search / delete). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
121 lines
3.9 KiB
Python
121 lines
3.9 KiB
Python
"""In-memory ThreadMetaStore backed by LangGraph BaseStore.
|
|
|
|
Used when database.backend=memory. Delegates to the LangGraph Store's
|
|
``("threads",)`` namespace — the same namespace used by the Gateway
|
|
router for thread records.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import Any
|
|
|
|
from langgraph.store.base import BaseStore
|
|
|
|
from deerflow.persistence.thread_meta.base import ThreadMetaStore
|
|
|
|
THREADS_NS: tuple[str, ...] = ("threads",)
|
|
|
|
|
|
class MemoryThreadMetaStore(ThreadMetaStore):
|
|
def __init__(self, store: BaseStore) -> None:
|
|
self._store = store
|
|
|
|
async def create(
|
|
self,
|
|
thread_id: str,
|
|
*,
|
|
assistant_id: str | None = None,
|
|
owner_id: str | None = None,
|
|
display_name: str | None = None,
|
|
metadata: dict | None = None,
|
|
) -> dict:
|
|
now = time.time()
|
|
record: dict[str, Any] = {
|
|
"thread_id": thread_id,
|
|
"assistant_id": assistant_id,
|
|
"owner_id": owner_id,
|
|
"display_name": display_name,
|
|
"status": "idle",
|
|
"metadata": metadata or {},
|
|
"values": {},
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
}
|
|
await self._store.aput(THREADS_NS, thread_id, record)
|
|
return record
|
|
|
|
async def get(self, thread_id: str) -> dict | None:
|
|
item = await self._store.aget(THREADS_NS, thread_id)
|
|
return item.value if item is not None else None
|
|
|
|
async def search(
|
|
self,
|
|
*,
|
|
metadata: dict | None = None,
|
|
status: str | None = None,
|
|
limit: int = 100,
|
|
offset: int = 0,
|
|
) -> list[dict]:
|
|
filter_dict: dict[str, Any] = {}
|
|
if metadata:
|
|
filter_dict.update(metadata)
|
|
if status:
|
|
filter_dict["status"] = status
|
|
|
|
items = await self._store.asearch(
|
|
THREADS_NS,
|
|
filter=filter_dict or None,
|
|
limit=limit,
|
|
offset=offset,
|
|
)
|
|
return [self._item_to_dict(item) for item in items]
|
|
|
|
async def update_display_name(self, thread_id: str, display_name: str) -> None:
|
|
item = await self._store.aget(THREADS_NS, thread_id)
|
|
if item is None:
|
|
return
|
|
record = dict(item.value)
|
|
record["display_name"] = display_name
|
|
record["updated_at"] = time.time()
|
|
await self._store.aput(THREADS_NS, thread_id, record)
|
|
|
|
async def update_status(self, thread_id: str, status: str) -> None:
|
|
item = await self._store.aget(THREADS_NS, thread_id)
|
|
if item is None:
|
|
return
|
|
record = dict(item.value)
|
|
record["status"] = status
|
|
record["updated_at"] = time.time()
|
|
await self._store.aput(THREADS_NS, thread_id, record)
|
|
|
|
async def update_metadata(self, thread_id: str, metadata: dict) -> None:
|
|
"""Merge ``metadata`` into the in-memory record. No-op if absent."""
|
|
item = await self._store.aget(THREADS_NS, thread_id)
|
|
if item is None:
|
|
return
|
|
record = dict(item.value)
|
|
merged = dict(record.get("metadata") or {})
|
|
merged.update(metadata)
|
|
record["metadata"] = merged
|
|
record["updated_at"] = time.time()
|
|
await self._store.aput(THREADS_NS, thread_id, record)
|
|
|
|
async def delete(self, thread_id: str) -> None:
|
|
await self._store.adelete(THREADS_NS, thread_id)
|
|
|
|
@staticmethod
|
|
def _item_to_dict(item) -> dict[str, Any]:
|
|
"""Convert a Store SearchItem to the dict format expected by callers."""
|
|
val = item.value
|
|
return {
|
|
"thread_id": item.key,
|
|
"assistant_id": val.get("assistant_id"),
|
|
"owner_id": val.get("owner_id"),
|
|
"display_name": val.get("display_name"),
|
|
"status": val.get("status", "idle"),
|
|
"metadata": val.get("metadata", {}),
|
|
"created_at": str(val.get("created_at", "")),
|
|
"updated_at": str(val.get("updated_at", "")),
|
|
}
|