188 lines
6.7 KiB
Python
188 lines
6.7 KiB
Python
"""Abstract interface for thread metadata storage.
|
|
|
|
Implementations:
|
|
- ThreadMetaRepository: SQL-backed (sqlite / postgres via SQLAlchemy)
|
|
- MemoryThreadMetaStore: wraps LangGraph BaseStore (memory mode)
|
|
|
|
All mutating and querying methods accept a ``user_id`` parameter with
|
|
three-state semantics (see :mod:`deerflow.runtime.user_context`):
|
|
|
|
- ``AUTO`` (default): resolve from the request-scoped contextvar.
|
|
- Explicit ``str``: use the provided value verbatim.
|
|
- Explicit ``None``: bypass owner filtering (migration/CLI only).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import abc
|
|
from typing import Any
|
|
|
|
from deerflow.runtime.user_context import AUTO, _AutoSentinel
|
|
|
|
# Configuration wizards / machine workers that must not appear in the default
|
|
# conversation list. Prefer ``metadata.system=true`` on create; this set also
|
|
# covers older rows that only set ``thread_type``.
|
|
EXCLUDED_SYSTEM_THREAD_TYPES = frozenset(
|
|
{
|
|
"scheduler",
|
|
"roundtable",
|
|
"notebook",
|
|
"deep_research",
|
|
"writing_setup",
|
|
"report_structure_setup",
|
|
"task-analysis",
|
|
"task-analysis-qa",
|
|
"task_deeplink",
|
|
}
|
|
)
|
|
|
|
# 这些系统线程 id 只存在于编排内部，不进「最近的对话」；loopback 派活必须能跨
|
|
# 「点击触发的 HTTP 用户」和「绑定 created_by 签发的 JWT」读写，否则会 404 not accessible。
|
|
SHARED_ACCESS_THREAD_TYPES = frozenset(
|
|
{
|
|
"roundtable",
|
|
"task-analysis",
|
|
"task-analysis-qa",
|
|
}
|
|
)
|
|
|
|
|
|
def is_excluded_system_thread(metadata: dict[str, Any] | None) -> bool:
|
|
"""Return True when a thread should be hidden from user conversation lists.
|
|
|
|
Task-analysis seat runs created before ``thread_type`` was stamped still
|
|
carry ``task_analysis_seat``. Those rows stay in the execution record and
|
|
out of recent chats.
|
|
"""
|
|
md = metadata or {}
|
|
if md.get("system"):
|
|
return True
|
|
# Task-analysis keeps its user-facing history in task_analysis_runs /
|
|
# task_analysis_run_messages. The underlying LangGraph threads are runtime
|
|
# implementation details; recognise their task markers as a defensive
|
|
# fallback in case a legacy caller omitted ``system`` or ``thread_type``.
|
|
if any(
|
|
isinstance(md.get(key), str) and md[key].strip()
|
|
for key in (
|
|
"task_analysis_task_id",
|
|
"task_analysis_source_task_id",
|
|
"task_analysis_run_id",
|
|
)
|
|
):
|
|
return True
|
|
seat = md.get("task_analysis_seat")
|
|
if isinstance(seat, str) and seat.strip():
|
|
return True
|
|
thread_type = md.get("thread_type")
|
|
return isinstance(thread_type, str) and thread_type in EXCLUDED_SYSTEM_THREAD_TYPES
|
|
|
|
|
|
def is_shared_access_thread(metadata: dict[str, Any] | None) -> bool:
|
|
"""Return True when any authenticated caller may read/write this thread."""
|
|
md = metadata or {}
|
|
if md.get("taskId"):
|
|
return True
|
|
thread_type = md.get("thread_type")
|
|
return isinstance(thread_type, str) and thread_type in SHARED_ACCESS_THREAD_TYPES
|
|
|
|
|
|
class ThreadMetaStore(abc.ABC):
|
|
@abc.abstractmethod
|
|
async def create(
|
|
self,
|
|
thread_id: str,
|
|
*,
|
|
assistant_id: str | None = None,
|
|
user_id: str | None | _AutoSentinel = AUTO,
|
|
display_name: str | None = None,
|
|
metadata: dict | None = None,
|
|
) -> dict:
|
|
"""Create one metadata row.
|
|
|
|
Implementations must assemble the returned dict from the values passed
|
|
in, without a post-commit read-back: on MySQL such a read can be routed
|
|
to a lagging replica behind a read/write-splitting endpoint and fail.
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
async def get(self, thread_id: str, *, user_id: str | None | _AutoSentinel = AUTO) -> dict | None:
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
async def search(
|
|
self,
|
|
*,
|
|
metadata: dict | None = None,
|
|
status: str | None = None,
|
|
limit: int = 100,
|
|
offset: int = 0,
|
|
user_id: str | None | _AutoSentinel = AUTO,
|
|
exclude_system: bool = False,
|
|
query: str | None = None,
|
|
) -> list[dict]:
|
|
"""Search threads ordered by ``updated_at`` desc.
|
|
|
|
``exclude_system=True`` drops machine / wizard threads (``metadata.system``
|
|
or known ``thread_type`` values such as scheduler / report_structure_setup)
|
|
so user-facing conversation lists are not crowded out.
|
|
``query`` is a case-insensitive substring match on the thread title
|
|
(``display_name``). ``limit``/``offset`` apply to the *filtered*
|
|
result set.
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
async def count(
|
|
self,
|
|
*,
|
|
metadata: dict | None = None,
|
|
status: str | None = None,
|
|
user_id: str | None | _AutoSentinel = AUTO,
|
|
exclude_system: bool = False,
|
|
query: str | None = None,
|
|
) -> int:
|
|
"""Count threads matching the same filters as :meth:`search`.
|
|
|
|
Backs pagination UIs that need a total alongside per-page results.
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
async def update_display_name(self, thread_id: str, display_name: str, *, user_id: str | None | _AutoSentinel = AUTO) -> None:
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
async def update_status(self, thread_id: str, status: str, *, user_id: str | None | _AutoSentinel = AUTO) -> None:
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
async def update_assistant_id(self, thread_id: str, assistant_id: str, *, user_id: str | None | _AutoSentinel = AUTO) -> None:
|
|
"""Backfill ``assistant_id`` for a thread row.
|
|
|
|
Used by ``start_run`` when the row was pre-created via ``POST /threads``
|
|
without an assistant_id (the SDK's ``client.threads.create()`` doesn't
|
|
pass it), and the first run on the thread reveals the assistant. No-op
|
|
if the thread does not exist or the owner check fails.
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
async def update_metadata(self, thread_id: str, metadata: dict, *, user_id: str | None | _AutoSentinel = AUTO) -> None:
|
|
"""Merge ``metadata`` into the thread's metadata field.
|
|
|
|
Existing keys are overwritten by the new values; keys absent from
|
|
``metadata`` are preserved. No-op if the thread does not exist
|
|
or the owner check fails.
|
|
"""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
async def check_access(self, thread_id: str, user_id: str, *, require_existing: bool = False) -> bool:
|
|
"""Check if ``user_id`` has access to ``thread_id``."""
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
async def delete(self, thread_id: str, *, user_id: str | None | _AutoSentinel = AUTO) -> None:
|
|
pass
|