from __future__ import annotations import asyncio import os from pathlib import Path from types import SimpleNamespace import pytest from fastapi import HTTPException from app.gateway.routers import artifacts from deerflow.persistence.artifact_library import MemoryArtifactLibraryStore class _FakeThreadStore: def __init__(self, rows: list[dict]) -> None: self.rows = rows self.calls: list[dict] = [] async def search(self, **kwargs) -> list[dict]: self.calls.append(kwargs) offset = int(kwargs.get("offset", 0)) limit = int(kwargs.get("limit", 100)) return self.rows[offset : offset + limit] class _FakeAgentStore: async def list_by_ids(self, agent_ids: list[str], user_id: str) -> list[dict]: assert user_id == "owner" agents = { "analysis-agent": { "id": "analysis-agent", "name": "舆情分析智能体", } } return [agents[agent_id] for agent_id in agent_ids if agent_id in agents] def _strategy_task( project_id: str, *, name: str, synthesis: str, status: str = "done", stale: bool = False, updated_at: str = "2026-09-22T08:00:00+00:00", ) -> dict: return { "id": project_id, "title": name, "created_at": updated_at, "updated_at": updated_at, "messages": [ { "kind": "strategy-selection-project", "version": 1, "snapshot": { "version": 1, "taskDraft": {"name": name}, "synthesis": synthesis, "synthesisStatus": status, "workflowStale": stale, }, } ], } class _FakeStrategyTaskStore: def __init__(self, rows_by_user: dict[str, list[dict]]) -> None: self.rows_by_user = rows_by_user self.list_calls: list[dict] = [] self.get_calls: list[tuple[str, str]] = [] async def list_tasks( self, user_id: str, *, limit: int = 200, offset: int = 0, id_prefix: str | None = None, include_messages: bool = False, ) -> list[dict]: self.list_calls.append( { "user_id": user_id, "limit": limit, "offset": offset, "id_prefix": id_prefix, "include_messages": include_messages, } ) rows = self.rows_by_user.get(user_id, []) if id_prefix: rows = [row for row in rows if str(row.get("id") or "").startswith(id_prefix)] rows = rows[offset : offset + limit] if include_messages: return rows return [{key: value for key, value in row.items() if key != "messages"} for row in rows] async def get_task(self, task_id: str, user_id: str) -> dict | None: self.get_calls.append((task_id, user_id)) return next( (row for row in self.rows_by_user.get(user_id, []) if row.get("id") == task_id), None, ) def _library_item(*, thread_id: str, path: str, name: str, modified_at: str) -> dict: return { "thread_id": thread_id, "thread_title": "资料来源", "agent_id": None, "agent_name": None, "thread_type": None, "source_type": "general_qa", "strategy_project_id": None, "strategy_name": None, "strategy_stale": False, "strategy_source_kind": None, "download_url": None, "path": path, "name": name, "size_bytes": 10, "mime_type": "text/markdown", "modified_at": modified_at, "file_kind": "text_data", } @pytest.mark.asyncio async def test_artifact_library_aggregates_generated_non_media_files(tmp_path: Path, monkeypatch): thread_one = tmp_path / "thread-one" / "outputs" thread_two = tmp_path / "thread-two" / "outputs" thread_one.mkdir(parents=True) thread_two.mkdir(parents=True) report = thread_one / "研判报告.docx" table = thread_one / "数据汇总.xlsx" image = thread_one / "配图.png" audio = thread_two / "录音.mp3" notes = thread_two / "结论.md" archive = thread_two / "证据.zip" for path, content in ( (report, b"doc"), (table, b"sheet"), (image, b"image"), (audio, b"audio"), (notes, b"# notes"), (archive, b"zip"), ): path.write_bytes(content) os.utime(report, (100, 100)) os.utime(table, (200, 200)) os.utime(notes, (300, 300)) os.utime(archive, (400, 400)) rows = [ { "thread_id": "thread-one", "display_name": "战场环境综合研判", "metadata": {"agent_id": "analysis-agent"}, }, { "thread_id": "thread-two", "display_name": "生成发展报告", "metadata": {}, }, ] store = _FakeThreadStore(rows) class _Paths: def sandbox_outputs_dir(self, thread_id: str, *, user_id: str) -> Path: assert user_id == "owner" return tmp_path / thread_id / "outputs" async def _resolve_owner(_thread_id: str) -> str: return "owner" monkeypatch.setattr(artifacts, "get_paths", lambda: _Paths()) monkeypatch.setattr(artifacts, "aresolve_path_user_id", _resolve_owner) items = await artifacts._scan_artifact_library(store) assert [item["name"] for item in items] == ["证据.zip", "结论.md", "数据汇总.xlsx", "研判报告.docx"] assert [item["file_kind"] for item in items] == ["archive", "text_data", "spreadsheet", "document"] assert [item["source_type"] for item in items] == ["general_qa", "general_qa", "agent", "agent"] assert items[-1]["thread_title"] == "战场环境综合研判" assert items[-1]["agent_id"] == "analysis-agent" assert all(item["name"] not in {"配图.png", "录音.mp3"} for item in items) assert store.calls[0]["exclude_system"] is True @pytest.mark.asyncio async def test_artifact_library_endpoint_filters_searches_and_pages(tmp_path: Path, monkeypatch): output_dir = tmp_path / "thread-one" / "outputs" agent_output_dir = tmp_path / "thread-agent" / "outputs" output_dir.mkdir(parents=True) agent_output_dir.mkdir(parents=True) (output_dir / "任务报告.md").write_text("report", encoding="utf-8") (output_dir / "任务数据.csv").write_text("a,b", encoding="utf-8") (agent_output_dir / "智能体结论.md").write_text("agent report", encoding="utf-8") store = _FakeThreadStore( [ { "thread_id": "thread-one", "display_name": "任务研判对话", "metadata": {}, }, { "thread_id": "thread-agent", "display_name": "智能体研判对话", "metadata": {"agent_id": "analysis-agent"}, }, ] ) class _Paths: def sandbox_outputs_dir(self, thread_id: str, *, user_id: str) -> Path: return tmp_path / thread_id / "outputs" async def _resolve_owner(_thread_id: str) -> str: return "owner" monkeypatch.setattr(artifacts, "get_paths", lambda: _Paths()) monkeypatch.setattr(artifacts, "aresolve_path_user_id", _resolve_owner) monkeypatch.setattr(artifacts, "get_effective_user_id", lambda: "owner") index = MemoryArtifactLibraryStore() generation = await index.claim_sync("owner", force=True) assert generation is not None indexed_items = await artifacts._scan_artifact_library(store, user_id="owner") indexed_items = await artifacts._resolve_artifact_library_agent_names( indexed_items, _FakeAgentStore(), user_id="owner", ) assert await index.finish_sync("owner", generation, indexed_items) request = SimpleNamespace( app=SimpleNamespace( state=SimpleNamespace( artifact_library_store=index, artifact_library_sync_tasks={}, artifact_library_sync_semaphore=asyncio.Semaphore(1), thread_store=store, agent_store=_FakeAgentStore(), parallel_agent_task_store=None, ) ) ) raw_endpoint = artifacts.list_artifact_library.__wrapped__ response = await raw_endpoint( request=request, query="研判对话", file_kind="text_data", source_type="general_qa", offset=0, limit=1, refresh=False, ) assert response.total == 1 assert response.offset == 0 assert response.limit == 1 assert [item.name for item in response.items] == ["任务报告.md"] spreadsheet_response = await raw_endpoint( request=request, query="任务数据", file_kind="spreadsheet", source_type="all", offset=0, limit=50, refresh=False, ) assert spreadsheet_response.total == 1 assert spreadsheet_response.items[0].name == "任务数据.csv" agent_response = await raw_endpoint( request=request, query=None, file_kind="all", source_type="agent", offset=0, limit=50, refresh=False, ) assert agent_response.total == 1 assert agent_response.items[0].source_type == "agent" assert agent_response.items[0].agent_name == "舆情分析智能体" assert agent_response.items[0].name == "智能体结论.md" first_page = await raw_endpoint( request=request, query=None, file_kind="all", source_type="all", offset=0, limit=1, refresh=False, ) second_page = await raw_endpoint( request=request, query=None, file_kind="all", source_type="all", offset=1, limit=1, refresh=False, ) assert first_page.total == 3 assert second_page.total == 3 assert first_page.items[0].path != second_page.items[0].path assert second_page.offset == 1 @pytest.mark.asyncio async def test_artifact_library_marks_analysis_project_strategy_files(tmp_path: Path, monkeypatch): output_dir = tmp_path / "strategy-thread" / "outputs" output_dir.mkdir(parents=True) (output_dir / "策略简报.md").write_text("# strategy", encoding="utf-8") store = _FakeThreadStore( [ { "thread_id": "strategy-thread", "display_name": "策略项目对话", "metadata": { "analysis_project_id": "analysis-strategy-1", "analysis_project_mode": "strategy", "analysis_project_name": "海外市场策略", "agent_id": "analysis-agent", }, } ] ) class _Paths: def sandbox_outputs_dir(self, thread_id: str, *, user_id: str) -> Path: assert thread_id == "strategy-thread" assert user_id == "owner" return output_dir async def _resolve_owner(_thread_id: str) -> str: return "owner" monkeypatch.setattr(artifacts, "get_paths", lambda: _Paths()) monkeypatch.setattr(artifacts, "aresolve_path_user_id", _resolve_owner) items = await artifacts._scan_artifact_library(store) assert len(items) == 1 assert items[0]["source_type"] == "strategy" assert items[0]["strategy_source_kind"] == "analysis_project" assert items[0]["strategy_project_id"] == "analysis-strategy-1" assert items[0]["strategy_name"] == "海外市场策略" assert items[0]["agent_id"] == "analysis-agent" @pytest.mark.asyncio async def test_artifact_library_projects_completed_strategy_snapshots_and_filters(monkeypatch): thread_store = _FakeThreadStore([]) strategy_store = _FakeStrategyTaskStore( { "owner": [ _strategy_task( "strategy-selection-ready", name="产业韧性策略", synthesis="# 产业韧性方案\n\n正文", stale=True, ), _strategy_task( "strategy-selection-running", name="尚未完成", synthesis="draft", status="running", ), _strategy_task( "strategy-selection-empty", name="空终稿", synthesis=" ", ), { **_strategy_task( "strategy-selection-future", name="未来版本", synthesis="future", ), "messages": [ { "kind": "strategy-selection-project", "version": 1, "snapshot": { "version": 2, "taskDraft": {"name": "未来版本"}, "synthesis": "future", "synthesisStatus": "done", "workflowStale": False, }, } ], }, { **_strategy_task( "parallel-task-lookalike", name="非策略任务", synthesis="must stay hidden", ), "id": "parallel-task-lookalike", }, ], "other": [ _strategy_task( "strategy-selection-other-user", name="他人策略", synthesis="private", ) ], } ) request = SimpleNamespace( app=SimpleNamespace( state=SimpleNamespace(parallel_agent_task_store=strategy_store) ) ) monkeypatch.setattr(artifacts, "get_effective_user_id", lambda: "owner") index = MemoryArtifactLibraryStore() generation = await index.claim_sync("owner", force=True) assert generation is not None indexed_items = await artifacts._scan_artifact_library( thread_store, strategy_store, user_id="owner", ) assert await index.finish_sync("owner", generation, indexed_items) request.app.state.artifact_library_store = index request.app.state.artifact_library_sync_tasks = {} request.app.state.artifact_library_sync_semaphore = asyncio.Semaphore(1) request.app.state.thread_store = thread_store request.app.state.agent_store = _FakeAgentStore() raw_endpoint = artifacts.list_artifact_library.__wrapped__ all_strategy_items = await raw_endpoint( request=request, query=None, file_kind="all", source_type="strategy", offset=0, limit=50, refresh=False, ) assert [item.strategy_project_id for item in all_strategy_items.items] == [ "strategy-selection-ready" ] response = await raw_endpoint( request=request, query="韧性", file_kind="text_data", source_type="strategy", offset=0, limit=50, refresh=False, ) assert response.total == 1 item = response.items[0] assert item.source_type == "strategy" assert item.strategy_source_kind == "strategy_selection" assert item.strategy_project_id == "strategy-selection-ready" assert item.strategy_name == "产业韧性策略" assert item.strategy_stale is True assert item.name == "产业韧性策略.md" assert item.size_bytes == len("# 产业韧性方案\n\n正文".encode()) assert item.download_url == "/api/strategy-project-artifacts/strategy-selection-ready/download" assert strategy_store.list_calls == [ { "user_id": "owner", "limit": 200, "offset": 0, "id_prefix": "strategy-selection-", "include_messages": True, } ] @pytest.mark.asyncio async def test_strategy_snapshot_scan_pages_the_strategy_namespace(monkeypatch): strategy_store = _FakeStrategyTaskStore( { "owner": [ _strategy_task( "strategy-selection-newer", name="较新策略", synthesis="newer", ), _strategy_task( "strategy-selection-older", name="较早策略", synthesis="older", ), _strategy_task( "parallel-unrelated", name="普通并行任务", synthesis="not a strategy project", ), ] } ) monkeypatch.setattr(artifacts, "get_effective_user_id", lambda: "owner") monkeypatch.setattr(artifacts, "_STRATEGY_PROJECT_BATCH_SIZE", 1) items = await artifacts._scan_strategy_project_artifacts(strategy_store) assert [item["strategy_project_id"] for item in items] == [ "strategy-selection-newer", "strategy-selection-older", ] assert [call["offset"] for call in strategy_store.list_calls] == [0, 1, 2] assert all( call["id_prefix"] == "strategy-selection-" and call["include_messages"] is True for call in strategy_store.list_calls ) @pytest.mark.asyncio async def test_strategy_snapshot_scan_strict_mode_rejects_partial_source(): class _FailingStrategyTaskStore: async def list_tasks(self, *_args, **_kwargs): raise OSError("temporary strategy store failure") assert await artifacts._scan_strategy_project_artifacts( _FailingStrategyTaskStore(), user_id="owner", ) == [] with pytest.raises(artifacts.ArtifactLibraryScanIncomplete): await artifacts._scan_strategy_project_artifacts( _FailingStrategyTaskStore(), user_id="owner", strict=True, ) @pytest.mark.asyncio async def test_strategy_project_artifact_download_is_owner_scoped(monkeypatch): content = "# 当前用户策略方案\n\n仅当前用户可下载。" strategy_store = _FakeStrategyTaskStore( { "owner": [ _strategy_task( "strategy-selection-中文", name='A/B:策略', synthesis=content, stale=True, ), _strategy_task( "strategy-selection-pending", name="未完成策略", synthesis="draft", status="running", ), ], "other": [ _strategy_task( "strategy-selection-private", name="他人策略", synthesis="private", ) ], } ) request = SimpleNamespace( state=SimpleNamespace(user=SimpleNamespace(id="owner")), app=SimpleNamespace( state=SimpleNamespace(parallel_agent_task_store=strategy_store) ), ) monkeypatch.setattr(artifacts, "get_effective_user_id", lambda: "fallback") raw_endpoint = artifacts.download_strategy_project_artifact.__wrapped__ response = await raw_endpoint("strategy-selection-中文", request) assert response.body.decode("utf-8") == content assert response.media_type == "text/markdown; charset=utf-8" assert "attachment" in response.headers["content-disposition"] assert "A_B%3A" not in response.headers["content-disposition"] assert "A_B_" in response.headers["content-disposition"] assert strategy_store.get_calls[-1] == ("strategy-selection-中文", "owner") with pytest.raises(HTTPException) as error: await raw_endpoint("strategy-selection-private", request) assert error.value.status_code == 404 assert strategy_store.get_calls[-1] == ("strategy-selection-private", "owner") with pytest.raises(HTTPException) as error: await raw_endpoint("strategy-selection-pending", request) assert error.value.status_code == 404 assert strategy_store.get_calls[-1] == ("strategy-selection-pending", "owner") @pytest.mark.asyncio async def test_artifact_library_refresh_serves_snapshot_while_sync_runs(monkeypatch): index = MemoryArtifactLibraryStore() initial_generation = await index.claim_sync("owner", force=True) assert initial_generation is not None assert await index.finish_sync( "owner", initial_generation, [ _library_item( thread_id="old-thread", path="mnt/user-data/outputs/old.md", name="旧报告.md", modified_at="2026-09-01T00:00:00+00:00", ) ], ) scan_started = asyncio.Event() release_scan = asyncio.Event() async def _delayed_scan(*_args, **_kwargs): scan_started.set() await release_scan.wait() return [ _library_item( thread_id="new-thread", path="mnt/user-data/outputs/new.md", name="新报告.md", modified_at="2026-09-02T00:00:00+00:00", ) ] monkeypatch.setattr(artifacts, "_scan_artifact_library", _delayed_scan) monkeypatch.setattr(artifacts, "get_effective_user_id", lambda: "owner") request = SimpleNamespace( app=SimpleNamespace( state=SimpleNamespace( artifact_library_store=index, artifact_library_sync_tasks={}, artifact_library_sync_semaphore=asyncio.Semaphore(1), thread_store=_FakeThreadStore([]), agent_store=_FakeAgentStore(), parallel_agent_task_store=None, ) ) ) raw_endpoint = artifacts.list_artifact_library.__wrapped__ response = await raw_endpoint( request=request, query=None, file_kind="all", source_type="all", offset=0, limit=50, refresh=True, ) assert response.syncing is True assert response.sync_status == "syncing" assert [item.name for item in response.items] == ["旧报告.md"] await asyncio.wait_for(scan_started.wait(), timeout=1) task = request.app.state.artifact_library_sync_tasks["owner"] release_scan.set() await asyncio.wait_for(task, timeout=1) refreshed = await raw_endpoint( request=request, query=None, file_kind="all", source_type="all", offset=0, limit=50, refresh=False, ) assert refreshed.syncing is False assert refreshed.sync_status == "ready" assert [item.name for item in refreshed.items] == ["新报告.md"] @pytest.mark.asyncio async def test_incomplete_background_scan_keeps_last_complete_snapshot(monkeypatch): index = MemoryArtifactLibraryStore() initial_generation = await index.claim_sync("owner", force=True) assert initial_generation is not None assert await index.finish_sync( "owner", initial_generation, [ _library_item( thread_id="stable-thread", path="mnt/user-data/outputs/stable.md", name="稳定报告.md", modified_at="2026-09-01T00:00:00+00:00", ) ], ) async def _failed_scan(*_args, **_kwargs): raise artifacts.ArtifactLibraryScanIncomplete( r"failed at D:\private\sandbox\outputs" ) monkeypatch.setattr(artifacts, "_scan_artifact_library", _failed_scan) monkeypatch.setattr(artifacts, "get_effective_user_id", lambda: "owner") request = SimpleNamespace( app=SimpleNamespace( state=SimpleNamespace( artifact_library_store=index, artifact_library_sync_tasks={}, artifact_library_sync_semaphore=asyncio.Semaphore(1), thread_store=_FakeThreadStore([]), agent_store=_FakeAgentStore(), parallel_agent_task_store=None, ) ) ) raw_endpoint = artifacts.list_artifact_library.__wrapped__ first = await raw_endpoint( request=request, query=None, file_kind="all", source_type="all", offset=0, limit=50, refresh=True, ) assert first.syncing is True task = request.app.state.artifact_library_sync_tasks["owner"] await asyncio.wait_for(task, timeout=1) after_failure = await raw_endpoint( request=request, query=None, file_kind="all", source_type="all", offset=0, limit=50, refresh=False, ) assert after_failure.sync_status == "error" assert after_failure.syncing is False assert after_failure.sync_error == "资料索引同步失败，请稍后重试" assert "private" not in (after_failure.sync_error or "").lower() assert [item.name for item in after_failure.items] == ["稳定报告.md"]