360 lines
15 KiB
Python
360 lines
15 KiB
Python
"""Dedicated taskId + goPath conversation-history coverage."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import sqlalchemy as sa
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from langgraph.store.memory import InMemoryStore
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from deerflow.persistence.task_deeplink_threads.memory import (
|
|
MemoryTaskDeeplinkThreadStore,
|
|
)
|
|
from deerflow.persistence.task_deeplink_threads.model import TaskDeeplinkThreadRow
|
|
from deerflow.persistence.task_deeplink_threads.sql import TaskDeeplinkThreadRepository
|
|
from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore
|
|
from deerflow.persistence.thread_meta.model import ThreadMetaRow
|
|
from deerflow.persistence.thread_meta.sql import ThreadMetaRepository
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dedicated_history_is_partitioned_by_original_task_id_and_go_path():
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
try:
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(ThreadMetaRow.__table__.create)
|
|
await connection.run_sync(TaskDeeplinkThreadRow.__table__.create)
|
|
|
|
thread_repo = ThreadMetaRepository(session_factory)
|
|
history_repo = TaskDeeplinkThreadRepository(session_factory)
|
|
now = datetime.now(UTC)
|
|
|
|
await thread_repo.create(
|
|
"rwfx-oldest",
|
|
user_id="creator-a",
|
|
display_name="任务分析一",
|
|
metadata={
|
|
"taskId": "108",
|
|
"taskRecordId": "108",
|
|
"taskGoPath": "rwfx",
|
|
"thread_type": "task_deeplink",
|
|
"agent_id": "rwfx-agent",
|
|
},
|
|
)
|
|
await thread_repo.create(
|
|
"three-q",
|
|
user_id="creator-b",
|
|
metadata={
|
|
"taskId": "108",
|
|
"taskRecordId": "108",
|
|
"taskGoPath": "3qfx",
|
|
"thread_type": "task_deeplink",
|
|
"agent_id": "three-q-agent",
|
|
},
|
|
)
|
|
await thread_repo.create(
|
|
"xdfx-action",
|
|
user_id="creator-c",
|
|
metadata={
|
|
"taskId": "108",
|
|
"taskRecordId": "action-9",
|
|
"taskGoPath": "xdfx",
|
|
"thread_type": "task_deeplink",
|
|
"actionId": "action-9",
|
|
"agent_id": "xdfx-agent",
|
|
},
|
|
)
|
|
# A legacy task thread has no taskGoPath marker and must stay out of
|
|
# the new table: old data is not migrated implicitly.
|
|
await thread_repo.create(
|
|
"legacy-task-thread",
|
|
user_id="creator-d",
|
|
metadata={"taskId": "108", "agent_id": "rwfx-agent"},
|
|
)
|
|
|
|
# Listing reads the dedicated card, not ``threads_meta``. Touching
|
|
# only the generic row must not change order or title.
|
|
async with session_factory() as session:
|
|
row = await session.get(ThreadMetaRow, "rwfx-oldest")
|
|
assert row is not None
|
|
row.display_name = "通用表里的新标题"
|
|
row.updated_at = now + timedelta(minutes=1)
|
|
await session.commit()
|
|
|
|
rwfx = await history_repo.list_threads("108", "rwfx")
|
|
three_q = await history_repo.list_threads("108", "3qfx")
|
|
xdfx = await history_repo.list_threads("action-9", "xdfx")
|
|
|
|
assert [row["thread_id"] for row in rwfx] == ["rwfx-oldest"]
|
|
assert rwfx[0]["values"] == {"title": "任务分析一"}
|
|
assert [row["thread_id"] for row in three_q] == ["three-q"]
|
|
assert [row["thread_id"] for row in xdfx] == ["xdfx-action"]
|
|
|
|
await thread_repo.update_display_name("rwfx-oldest", "卡片同步后的标题", user_id=None)
|
|
await thread_repo.update_status("rwfx-oldest", "running", user_id=None)
|
|
synced = await history_repo.list_threads("108", "rwfx")
|
|
assert synced[0]["values"] == {"title": "卡片同步后的标题"}
|
|
assert synced[0]["status"] == "running"
|
|
|
|
async with session_factory() as session:
|
|
indexed_ids = set((await session.scalars(select(TaskDeeplinkThreadRow.thread_id))).all())
|
|
assert indexed_ids == {"rwfx-oldest", "three-q", "xdfx-action"}
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_metadata_registration_and_thread_delete_keep_index_in_sync():
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
try:
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(ThreadMetaRow.__table__.create)
|
|
await connection.run_sync(TaskDeeplinkThreadRow.__table__.create)
|
|
|
|
thread_repo = ThreadMetaRepository(session_factory)
|
|
history_repo = TaskDeeplinkThreadRepository(session_factory)
|
|
await thread_repo.create(
|
|
"late-binding",
|
|
user_id="creator",
|
|
metadata={"taskId": "2055", "agent_id": "three-q-agent"},
|
|
)
|
|
assert await history_repo.list_threads("2055", "3qfx") == []
|
|
|
|
await thread_repo.update_metadata(
|
|
"late-binding",
|
|
{
|
|
"taskRecordId": "2055",
|
|
"taskGoPath": "3qfx",
|
|
"thread_type": "task_deeplink",
|
|
},
|
|
user_id=None,
|
|
)
|
|
assert [row["thread_id"] for row in await history_repo.list_threads("2055", "3qfx")] == ["late-binding"]
|
|
|
|
await thread_repo.delete("late-binding", user_id=None)
|
|
assert await history_repo.list_threads("2055", "3qfx") == []
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_memory_mode_uses_the_same_dedicated_partition_contract():
|
|
raw_store = InMemoryStore()
|
|
thread_store = MemoryThreadMetaStore(raw_store)
|
|
history_store = MemoryTaskDeeplinkThreadStore(raw_store, thread_store)
|
|
|
|
await thread_store.create(
|
|
"memory-rwfx",
|
|
user_id="creator-a",
|
|
display_name="内存任务会话",
|
|
metadata={
|
|
"taskId": "108",
|
|
"taskRecordId": "108",
|
|
"taskGoPath": "rwfx",
|
|
"thread_type": "task_deeplink",
|
|
},
|
|
)
|
|
await thread_store.create(
|
|
"memory-legacy",
|
|
user_id="creator-b",
|
|
metadata={"taskId": "108"},
|
|
)
|
|
await thread_store.create(
|
|
"memory-xdfx",
|
|
user_id="creator-c",
|
|
metadata={
|
|
"taskId": "108",
|
|
"taskRecordId": "action-9",
|
|
"taskGoPath": "xdfx",
|
|
"thread_type": "task_deeplink",
|
|
},
|
|
)
|
|
|
|
rwfx = await history_store.list_threads("108", "rwfx")
|
|
xdfx = await history_store.list_threads("action-9", "xdfx")
|
|
assert [row["thread_id"] for row in rwfx] == ["memory-rwfx"]
|
|
assert rwfx[0]["values"] == {"title": "内存任务会话"}
|
|
assert [row["thread_id"] for row in xdfx] == ["memory-xdfx"]
|
|
|
|
await thread_store.update_display_name("memory-rwfx", "内存同步标题", user_id=None)
|
|
await thread_store.update_status("memory-rwfx", "running", user_id=None)
|
|
refreshed = await history_store.list_threads("108", "rwfx")
|
|
assert refreshed[0]["values"] == {"title": "内存同步标题"}
|
|
assert refreshed[0]["status"] == "running"
|
|
|
|
await thread_store.delete("memory-rwfx", user_id=None)
|
|
assert await history_store.list_threads("108", "rwfx") == []
|
|
|
|
|
|
def _b51_metadata(task_id: str, task_type: str | None) -> dict:
|
|
metadata: dict = {
|
|
"taskId": task_id,
|
|
"taskRecordId": task_id,
|
|
"taskGoPath": "b5-1",
|
|
"thread_type": "task_deeplink",
|
|
}
|
|
if task_type is not None:
|
|
metadata["taskType"] = task_type
|
|
return metadata
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_b51_sub_type_partitions_one_task_history():
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
try:
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(ThreadMetaRow.__table__.create)
|
|
await connection.run_sync(TaskDeeplinkThreadRow.__table__.create)
|
|
|
|
thread_repo = ThreadMetaRepository(session_factory)
|
|
history_repo = TaskDeeplinkThreadRepository(session_factory)
|
|
await thread_repo.create("b51-yuqing", user_id="u1", metadata=_b51_metadata("300", "yuqing"))
|
|
await thread_repo.create("b51-other", user_id="u2", metadata=_b51_metadata("300", "other"))
|
|
await thread_repo.create("b51-untyped", user_id="u3", metadata=_b51_metadata("300", None))
|
|
|
|
# Omitted sub_type keeps the legacy all-sub-types listing.
|
|
assert sorted(row["thread_id"] for row in await history_repo.list_threads("300", "b5-1")) == [
|
|
"b51-other",
|
|
"b51-untyped",
|
|
"b51-yuqing",
|
|
]
|
|
assert [
|
|
row["thread_id"] for row in await history_repo.list_threads("300", "b5-1", sub_type="yuqing")
|
|
] == ["b51-yuqing"]
|
|
# Explicit "" selects the no-type partition, not "every sub-type".
|
|
assert [
|
|
row["thread_id"] for row in await history_repo.list_threads("300", "b5-1", sub_type="")
|
|
] == ["b51-untyped"]
|
|
assert await history_repo.list_threads("300", "b5-1", sub_type="missing") == []
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_memory_mode_b51_sub_type_partition_contract():
|
|
raw_store = InMemoryStore()
|
|
thread_store = MemoryThreadMetaStore(raw_store)
|
|
history_store = MemoryTaskDeeplinkThreadStore(raw_store, thread_store)
|
|
|
|
await thread_store.create("mem-yuqing", user_id="u1", metadata=_b51_metadata("300", "yuqing"))
|
|
await thread_store.create("mem-untyped", user_id="u2", metadata=_b51_metadata("300", None))
|
|
|
|
assert [
|
|
row["thread_id"] for row in await history_store.list_threads("300", "b5-1", sub_type="yuqing")
|
|
] == ["mem-yuqing"]
|
|
assert [
|
|
row["thread_id"] for row in await history_store.list_threads("300", "b5-1", sub_type="")
|
|
] == ["mem-untyped"]
|
|
assert sorted(row["thread_id"] for row in await history_store.list_threads("300", "b5-1")) == [
|
|
"mem-untyped",
|
|
"mem-yuqing",
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_does_not_join_generic_threads_meta():
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
try:
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(ThreadMetaRow.__table__.create)
|
|
await connection.run_sync(TaskDeeplinkThreadRow.__table__.create)
|
|
|
|
thread_repo = ThreadMetaRepository(session_factory)
|
|
history_repo = TaskDeeplinkThreadRepository(session_factory)
|
|
await thread_repo.create(
|
|
"card-only",
|
|
user_id="creator",
|
|
display_name="独立卡片",
|
|
metadata={
|
|
"taskId": "77",
|
|
"taskRecordId": "77",
|
|
"taskGoPath": "rwfx",
|
|
"thread_type": "task_deeplink",
|
|
},
|
|
)
|
|
async with session_factory() as session:
|
|
generic = await session.get(ThreadMetaRow, "card-only")
|
|
assert generic is not None
|
|
await session.delete(generic)
|
|
await session.commit()
|
|
|
|
listed = await history_repo.list_threads("77", "rwfx")
|
|
assert [row["thread_id"] for row in listed] == ["card-only"]
|
|
assert listed[0]["values"] == {"title": "独立卡片"}
|
|
assert listed[0]["metadata"]["taskGoPath"] == "rwfx"
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
def _migration_config(database: Path) -> Config:
|
|
migrations = Path(__file__).parents[3] / "packages" / "harness" / "deerflow" / "persistence" / "migrations"
|
|
config = Config()
|
|
config.set_main_option("script_location", str(migrations))
|
|
config.set_main_option("sqlalchemy.url", f"sqlite+aiosqlite:///{database.resolve().as_posix()}")
|
|
return config
|
|
|
|
|
|
def test_card_migration_backfills_title_status_and_metadata(tmp_path: Path):
|
|
database = tmp_path / "deeplink-cards.db"
|
|
engine = sa.create_engine(f"sqlite:///{database.resolve().as_posix()}")
|
|
try:
|
|
with engine.begin() as connection:
|
|
connection.exec_driver_sql(
|
|
"CREATE TABLE threads_meta ("
|
|
"thread_id VARCHAR(64) PRIMARY KEY, display_name VARCHAR(256), "
|
|
"status VARCHAR(20) NOT NULL, metadata_json JSON, "
|
|
"created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL)"
|
|
)
|
|
connection.exec_driver_sql(
|
|
"CREATE TABLE task_deeplink_threads ("
|
|
"thread_id VARCHAR(64) PRIMARY KEY, task_id VARCHAR(128) NOT NULL, "
|
|
"source_task_id VARCHAR(128) NOT NULL, go_path VARCHAR(16) NOT NULL, "
|
|
"sub_type VARCHAR(64) NOT NULL DEFAULT '', agent_id VARCHAR(128), "
|
|
"created_by VARCHAR(64), created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL)"
|
|
)
|
|
connection.exec_driver_sql(
|
|
"INSERT INTO threads_meta "
|
|
"(thread_id, display_name, status, metadata_json, created_at, updated_at) VALUES "
|
|
"('rwfx-1', '任务分析一', 'running', "
|
|
"'{\"taskId\":\"108\",\"taskGoPath\":\"rwfx\"}', "
|
|
"CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
|
|
)
|
|
connection.exec_driver_sql(
|
|
"INSERT INTO task_deeplink_threads "
|
|
"(thread_id, task_id, source_task_id, go_path, sub_type, created_at, updated_at) "
|
|
"VALUES ('rwfx-1', '108', '108', 'rwfx', '', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
|
|
)
|
|
connection.exec_driver_sql(
|
|
"CREATE TABLE alembic_version (version_num VARCHAR(32) NOT NULL PRIMARY KEY)"
|
|
)
|
|
connection.exec_driver_sql(
|
|
"INSERT INTO alembic_version (version_num) VALUES ('20261022_01')"
|
|
)
|
|
|
|
command.upgrade(_migration_config(database), "head")
|
|
|
|
with engine.connect() as connection:
|
|
row = connection.execute(
|
|
sa.text(
|
|
"SELECT display_name, status, metadata_json "
|
|
"FROM task_deeplink_threads WHERE thread_id = 'rwfx-1'"
|
|
)
|
|
).one()
|
|
version = connection.execute(sa.text("SELECT version_num FROM alembic_version")).scalar_one()
|
|
assert row.display_name == "任务分析一"
|
|
assert row.status == "running"
|
|
assert "rwfx" in str(row.metadata_json)
|
|
assert version == "20261023_03"
|
|
finally:
|
|
engine.dispose()
|