feat: 新增独立企业深度研究工作台，并接入工作流报告结构与 8BF 深色锁定

Co-authored-by: Cursor <cursoragent@cursor.com>

offline-backend-20260512/backend/CLAUDE.md

@@ -996,6 +996,8 @@ The embedded detail proxy hides WeKnora's outer `.main > .aside_box` application
 
 **Embedded native knowledge-base switching:** The original WeKnora breadcrumb picker remains available inside the detail iframe. A signed 15-minute embed cookie carries only the current DeerFlow user's visible mappings (their own plus published bases); the proxied `/api/v1/knowledge-bases` response is filtered to that allowlist. Choosing another entry performs a full iframe navigation, rechecks the DeerFlow mapping and read/write permission, then reissues the cookie for the selected base. Never expose the upstream service-admin list directly or permit an arbitrary remote knowledge-base ID.
 
+The independent enterprise-research workbench (`/api/enterprise-research/tasks`) is deliberately separate from `/api/deep-research`: it stores an owner-scoped task plus a bounded, deduplicated evidence snapshot in `enterprise_research_tasks`, then calls WeKnora through the existing DeerFlow LLMWiki mapping and visibility checks. Never expose remote WeKnora IDs or credentials to its browser API, and do not make this workbench mutate a knowledge base or an existing Deep Research session/report. After collection, its plan endpoints can save an editable Markdown outline and an explicitly selected source subset; validate every selected chunk ID against the saved snapshot. The AI-plan variant may call a configured DeerFlow model but must not retrieve new evidence. `enterprise_research_report_jobs` freezes that plan/source subset plus action/model/style/length and before-image before calling the shared `DeerFlowCompletionBackend`, so the writer cannot run a fresh uncontrolled retrieval or accidentally use later task edits. Actions are `initial`, `rewrite`, and report-only `followup`; only successful initial/rewrite jobs may replace the task report. Its SSE route sends the persisted draft first and then only live `report_delta`/`reasoning_delta` events; reasoning must never be appended to report Markdown. Active rows carry a nullable owner/task dedupe key with a database unique constraint; terminal transitions clear it so history remains unlimited while simultaneous workers cannot create two active jobs for one task. Startup recovery must requeue the exact previously-read running snapshot and then use an atomic queued→running claim. Cancellation, completion, and failure are conditional state transitions so cross-process races cannot turn cancelled work into a committed report. Restore only accepts a completed rewrite before-image, rejects active jobs, and writes a new reversible history row before changing the task. The persistence contract is `deerflow.persistence.enterprise_research`; large evidence/report payloads use `PortableLongText` because MySQL `TEXT` cannot safely hold them. Regression coverage: `tests/test_enterprise_research_tasks.py`.
+
 ### Embedded Client (`packages/harness/deerflow/client.py`)
 
 `DeerFlowClient` provides direct in-process access to all DeerFlow capabilities without HTTP services. All return types align with the Gateway API response schemas, so consumer code works identically in HTTP and embedded modes.

offline-backend-20260512/backend/README.md

@@ -81,6 +81,7 @@ Per-thread isolated execution with virtual path translation:
 - **Skills path**: `/mnt/skills` → `deer-flow/skills/` directory
 - **Skills loading**: Recursively discovers nested `SKILL.md` files under `skills/{public,custom}` and preserves nested container paths
 - **Conditional WeKnora LLMWiki**: a non-empty `llmwiki.weknora.api_base_url` enables the DeerFlow-native WeKnora BFF and UI; an empty address preserves the existing LLMWiki page. WeKnora remains independently deployed and owns its database, Redis, storage, parsers and models. DeerFlow keeps only the API/admin addresses, user/publication mappings, agent knowledge-base bindings and per-conversation selections; it stores and sends no WeKnora credential, so the endpoint must be restricted to a trusted network and allow DeerFlow traffic. Owners publish directly to the public LLMWiki without an approval queue for ordinary bases; the system-owned conversation-deposit base is always published, shared by all users, excluded from My Knowledge, immutable through user/admin detail operations, and recognized by both its reserved name and `system` owner. An existing same-name remote mapping is adopted in place as the system deposit so its mapping id and document links stay valid. Its user-facing knowledge-base description and generated document metadata use the `cmzs` brand; list/detail reads synchronize stale remote descriptions, while conversation-deposit search results, source previews and embedded-detail JSON rewrite legacy brand text to `cmzs`. Deposit requests must reference a thread accessible to the current account. Personal knowledge-base lists are scoped to the current DeerFlow account even for admins, while frontend LLMWiki query caches also include the active account id so switching accounts cannot reuse another account's list/detail data. Admins remain the only role that sees the WeKnora connection-config tab. Knowledge cards open a dedicated WeKnora-style detail route: Wiki-enabled bases default to Wiki, then documents and graph; standard bases hide Wiki and default to documents. Writable ordinary knowledge-base details expose a native multi-file upload action that submits files sequentially and reports aggregate success/failure; read-only public bases and the system deposit base do not. It supports file/URL/manual-Markdown imports, Wiki page browsing, document preview, real parsed chunks and Wiki graph nodes/edges. New chats use a compact WeKnora-style searchable/grouped multi-select inside the composer, while agent create/edit pages bind spaces through an inline panel. A single proxy service identity means DeerFlow users are isolated logically while their WeKnora entities share that identity's workspace. The read-only `llmwiki_search` tool reauthorizes the current DeerFlow user on every call and renders credential-free source previews; source previews prefer cleaned Wiki/LLMWiki text when a retrieved chunk maps to a Wiki page, with raw chunks as the fallback. The detail page first creates a signed iframe session through the normal DeerFlow API auth path, then points the iframe directly at the proxied WeKnora `/platform/...` route using the signed `deerflow_weknora_embed` cookie; global auth/CSRF bypass is limited to those signed iframe proxy requests, and HTTPS cookie attributes honor `X-Forwarded-Proto` behind reverse proxies. The iframe proxy keeps WeKnora's root-relative `/platform`, `/assets`, `/locales`, and `/api/v1/*` paths, but is registered last through `llmwiki.proxy_router` so it cannot intercept first-party Gateway APIs such as `POST /api/v1/auth/login/username`. See [docs/LLMWIKI_WEKNORA_INTEGRATION_ZH.md](docs/LLMWIKI_WEKNORA_INTEGRATION_ZH.md).
+- **Enterprise Research workbench**: `/api/enterprise-research/tasks` is an independent, user-scoped evidence workbench; it never enters or changes `/api/deep-research`. DeerFlow persists each task's topic, generated queries, selected DeerFlow knowledge-base mapping IDs, collection state, and a bounded/deduplicated WeKnora evidence snapshot. Collection reuses the LLMWiki BFF's mapping-based authorization, so browsers never send a remote WeKnora ID or credential and can retrieve only sources in their own or published knowledge bases. Once collection finishes, the task can persist an editable Markdown research plan and a validated subset of its snapshot sources; the optional AI-plan endpoint improves that outline with a selected DeerFlow model without retrieving new material. Persistent `report-jobs` support initial writing, full-report rewrite, and report-only follow-up. Each job freezes its plan, source subset, model/style/length controls and before-image, streams Markdown/reasoning through DeerFlow's standard model adapter, persists the latest draft for reconnect, and commits only a successful non-follow-up report back to the owning task. Startup recovery uses compare-and-set requeue/claim, while cancellation and completion are conditional transitions so another process cannot overwrite a cancelled job. Markdown/Word delivery, clickable evidence citations, before/after comparison and reversible rewrite versions are served from the same task history without modifying an existing Deep Research report.
 - **Embedded WeKnora detail chrome**: the proxy hides WeKnora's application sidebar (`.main > .aside_box`) and expands the detail outlet to the full iframe width, while preserving the Wiki page's own index/navigation panel.
 - **Embedded WeKnora knowledge-base switching**: the original breadcrumb picker works inside the iframe, but lists only the signed-in user's own and published knowledge bases. Every switch is reauthorized through DeerFlow and receives a fresh short-lived iframe session, so it never exposes the upstream service-admin knowledge-base list.
 - **Prefixed WeKnora deployment**: the frontend may be published under a separate prefix such as `/magentweb`, while embedded WeKnora detail traffic is served by the Gateway under `/deerflow`. The Gateway rewrites HTML root assets, WeKnora API URLs, redirects, and Vite dynamic dependency maps so page assets stay under `/deerflow`; the prefixed proxy routes must be registered before the unprefixed fallback. The signed embed cookie covers `/deerflow`, and auth/CSRF recognition includes all supported static directories.
@@ -153,6 +154,7 @@ FastAPI application providing REST endpoints for frontend integration:
 | `DELETE /api/threads/{id}` | Delete DeerFlow-managed local thread data after LangGraph thread deletion; unexpected failures are logged server-side and return a generic 500 detail |
 | `GET /api/threads/{id}/artifacts/{path}` | Serve generated artifacts |
 | `GET/POST /api/deep-research/*` | Durable Deep Research sessions and jobs. `quick`/`basic` run the compact report flow; `detailed` plans subtopics and writes independent sections; `deep` recursively investigates bounded follow-up branches. All modes reuse DeerFlow's configured material provider, LLM factory, hidden-thread artifacts, and SSE event log. Deep Research session/job/event/source/message writes serialise their rows in the writer transaction, rather than doing a post-commit ORM refresh or readback; this prevents MySQL read/write-splitting replica lag from failing “start writing”, dispatcher claim, report persistence, or follow-up history. A report is written to the session and confirmed before its Job becomes `completed`; an unconfirmed write is retried and never exposed as a false completion. Retrying an already-created queued job re-nudges the dispatcher, recovering an interrupted HTTP response without creating a second report. If progress-event persistence, durable replay, or live fan-out is briefly unavailable, the worker/stream logs the degradation and continues or reconnects rather than failing the whole job. Optional report illustrations use a separately configured OpenAI-compatible image endpoint and are served only through the owning session's authenticated artifact route. |
+| `GET/POST/PUT/DELETE /api/enterprise-research/tasks*` | Independent enterprise-research evidence tasks. The authenticated DeerFlow account owns the task and all saved evidence snapshots; collection calls the configured WeKnora service only after resolving selected mapping IDs through normal visibility and permission checks. `POST /{id}/plan/draft` creates an editable deterministic plan, `POST /{id}/plan/ai` optionally improves it with the configured model, and `PUT /{id}/plan` persists its Markdown plus a source subset revalidated against that task's snapshot. `POST /{id}/report-jobs` freezes those inputs and accepts `initial` / `rewrite` / `followup` plus model, style and length controls; the job list is durable history, its SSE route replays the latest persisted draft before live Markdown/reasoning deltas, `/cancel` safely stops an active job, and `/restore` creates another reversible version while restoring a completed rewrite's before-image. It deliberately cannot change existing Deep Research reports or knowledge bases. |
 | `POST /api/ai-writing/sample/extract` | Extract text from Word/PDF/Markdown/TXT samples for AI-writing imitation mode |
 | `POST /cop/saveSuperiorTask` | TaskCOP compatibility API (DeerFlow Bearer/session authentication required): create a situation-overview task with a sequential four-digit id (`0001`, `0002`, ...) and return the legacy `{state, msg, data}` envelope |
 | `POST /api/taskcop/import/tasks` | TaskCOP compatibility API (authenticated): import tasks from a JSON array or `tasks`/`records` envelope; every row's supplied `id`/`taskId` is authoritative and a duplicate id completely replaces the stored task |

offline-backend-20260512/backend/app/gateway/app.py

@@ -41,6 +41,7 @@ from app.gateway.routers import (
     feedback,
     dashboard_sessions,
     deep_research,
+    enterprise_research,
     html_page_favorites,
     intent,
     knowledge,
@@ -1560,6 +1561,9 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
     # Deep Research (深度研究) API at /api/deep-research
     app.include_router(deep_research.router)
 
+    # Independent enterprise-research evidence workbench at /api/enterprise-research.
+    app.include_router(enterprise_research.router)
+
     # Suggestions API is mounted at /api/threads/{thread_id}/suggestions
     app.include_router(suggestions.router)
 

offline-backend-20260512/backend/app/gateway/deps.py

@@ -305,6 +305,31 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
 
         app.state.report_structure_store = make_report_structure_store(sf)
 
+        from deerflow.persistence.enterprise_research import (
+            make_enterprise_research_report_job_store,
+            make_enterprise_research_task_store,
+        )
+
+        app.state.enterprise_research_task_store = make_enterprise_research_task_store(sf)
+        app.state.enterprise_research_report_job_store = make_enterprise_research_report_job_store(sf)
+
+        # 企业研究报告写作是独立的、冻结计划/证据快照的后台任务；它不复用或改变
+        # 已有 deep-research 的 Job / dispatcher。
+        from app.gateway.enterprise_research_report_executor import EnterpriseResearchReportExecutor
+
+        app.state.enterprise_research_report_executor = EnterpriseResearchReportExecutor(
+            app.state.enterprise_research_task_store,
+            app.state.enterprise_research_report_job_store,
+        )
+
+        async def _recover_enterprise_research_reports() -> None:
+            try:
+                await app.state.enterprise_research_report_executor.recover_pending()
+            except Exception:  # noqa: BLE001
+                logger.warning("[enterprise-research] startup report recovery failed", exc_info=True)
+
+        asyncio.create_task(_recover_enterprise_research_reports())
+
         from deerflow.integrations.weknora.runtime import register_llmwiki_store
         from deerflow.persistence.llmwiki import make_llmwiki_store
 

offline-backend-20260512/backend/app/gateway/enterprise_research_report_executor.py

@@ -0,0 +1,347 @@
+"""Independent, snapshot-based report writer for the enterprise workbench.
+
+This deliberately does *not* use the existing deep-research job executor.  It
+only consumes a task's persisted plan and selected WeKnora evidence snapshot,
+then streams Markdown through the shared DeerFlow model adapter.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from collections import defaultdict
+from collections.abc import AsyncIterator, Awaitable, Callable
+from difflib import SequenceMatcher
+from typing import Any
+
+from deerflow.agents.deep_research.adapters.llm import DeerFlowCompletionBackend
+from deerflow.agents.deep_research.config import DeepResearchConfig
+from deerflow.persistence.enterprise_research import EnterpriseResearchReportJobStore, EnterpriseResearchTaskStore
+
+logger = logging.getLogger(__name__)
+_TERMINAL = {"completed", "failed", "cancelled"}
+
+
+class EnterpriseResearchLiveHub:
+    """Small in-process fan-out for live deltas; completed output stays in SQL."""
+
+    def __init__(self) -> None:
+        self._queues: dict[str, set[asyncio.Queue[dict[str, Any]]]] = defaultdict(set)
+        self._lock = asyncio.Lock()
+
+    async def subscribe(
+        self, job_id: str
+    ) -> tuple[asyncio.Queue[dict[str, Any]], Callable[[], Awaitable[None]]]:
+        queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=256)
+        async with self._lock:
+            self._queues[job_id].add(queue)
+
+        async def unsubscribe() -> None:
+            async with self._lock:
+                queues = self._queues.get(job_id)
+                if queues is not None:
+                    queues.discard(queue)
+                    if not queues:
+                        self._queues.pop(job_id, None)
+
+        return queue, unsubscribe
+
+    async def publish(self, job_id: str, event: dict[str, Any]) -> None:
+        async with self._lock:
+            queues = list(self._queues.get(job_id, ()))
+        for queue in queues:
+            try:
+                queue.put_nowait(event)
+            except asyncio.QueueFull:
+                # A slow browser can reconnect and receive the persisted draft;
+                # never let it stall model streaming for every other client.
+                try:
+                    queue.get_nowait()
+                    queue.put_nowait(event)
+                except (asyncio.QueueEmpty, asyncio.QueueFull):
+                    pass
+
+
+class EnterpriseResearchReportExecutor:
+    def __init__(self, task_store: EnterpriseResearchTaskStore, job_store: EnterpriseResearchReportJobStore) -> None:
+        self._tasks = task_store
+        self._jobs = job_store
+        self._hub = EnterpriseResearchLiveHub()
+        self._running: dict[str, asyncio.Task[None]] = {}
+        self._guard = asyncio.Lock()
+
+    async def ensure_started(self, job_id: str, user_id: str) -> None:
+        """Start exactly one in-process runner for an owner-scoped persisted job."""
+
+        async with self._guard:
+            existing = self._running.get(job_id)
+            if existing is not None and not existing.done():
+                return
+            task = asyncio.create_task(self._run(job_id, user_id), name=f"enterprise-research-report:{job_id}")
+            self._running[job_id] = task
+            task.add_done_callback(lambda _: self._running.pop(job_id, None))
+
+    async def recover_pending(self) -> None:
+        """Requeue process-orphaned jobs and let the normal atomic claim decide the runner."""
+
+        for job in await self._jobs.list_recoverable(limit=100):
+            if job["status"] == "running":
+                requeued = await self._jobs.requeue_job(
+                    job["id"], job["user_id"], expected_updated_at=job.get("updated_at")
+                )
+                if not requeued:
+                    # Another process already recovered or resumed this exact
+                    # snapshot; its normal claim now owns the execution.
+                    continue
+            await self.ensure_started(job["id"], job["user_id"])
+
+    async def cancel(self, job_id: str, user_id: str) -> dict[str, Any] | None:
+        job = await self._jobs.get_job(job_id, user_id)
+        if job is None or job["status"] in _TERMINAL:
+            return job
+        running = self._running.get(job_id)
+        if running is not None and not running.done():
+            running.cancel()
+            try:
+                await running
+            except asyncio.CancelledError:
+                pass
+            return await self._jobs.get_job(job_id, user_id)
+        cancelled = await self._jobs.transition_job(
+            job_id, user_id, expected_statuses={"queued", "running"},
+            data={"status": "cancelled", "phase": "cancelled"},
+        )
+        if cancelled is not None:
+            await self._tasks.update_task(job["task_id"], user_id, {"active_report_job_id": None})
+            await self._hub.publish(job_id, {"type": "cancelled", "job": cancelled})
+            return cancelled
+        return await self._jobs.get_job(job_id, user_id)
+
+    async def stream_events(self, job_id: str, user_id: str) -> AsyncIterator[dict[str, Any]]:
+        """Replay the latest durable draft, then attach to live token events."""
+
+        queue, unsubscribe = await self._hub.subscribe(job_id)
+        try:
+            current = await self._jobs.get_job(job_id, user_id)
+            if current is None:
+                return
+            yield {"type": "snapshot", "job": current}
+            if current["status"] in _TERMINAL:
+                return
+            while True:
+                try:
+                    event = await asyncio.wait_for(queue.get(), timeout=15)
+                except TimeoutError:
+                    current = await self._jobs.get_job(job_id, user_id)
+                    if current is None:
+                        return
+                    yield {"type": "heartbeat", "status": current["status"], "phase": current["phase"]}
+                    if current["status"] in _TERMINAL:
+                        return
+                    continue
+                yield event
+                if event.get("type") in {"completed", "failed", "cancelled"}:
+                    return
+        finally:
+            await unsubscribe()
+
+    async def _set_phase(self, job_id: str, user_id: str, phase: str, **fields: Any) -> dict[str, Any] | None:
+        row = await self._jobs.transition_job(
+            job_id, user_id, expected_statuses={"running"}, data={"phase": phase, **fields}
+        )
+        if row is not None:
+            await self._hub.publish(job_id, {"type": "phase", "phase": phase, "status": row["status"]})
+        return row
+
+    @staticmethod
+    def _messages(job: dict[str, Any]) -> list[dict[str, str]]:
+        sources = job.get("sources") or []
+        evidence: list[str] = []
+        for index, source in enumerate(sources[:24], start=1):
+            if not isinstance(source, dict):
+                continue
+            chunk_id = str(source.get("chunk_id") or f"source-{index}")
+            title = str(source.get("title") or source.get("filename") or "未命名资料")
+            content = str(source.get("content") or "").strip()[:2_500]
+            if content:
+                evidence.append(f"[资料 {index} | id={chunk_id} | {title}]\n{content}")
+        action = str(job.get("action") or "initial")
+        style = str(job.get("style") or "formal")
+        instruction = str(job.get("instruction") or "").strip()
+        target_length = int(job.get("target_length") or 0)
+        source_context = f"# 已确认资料（共 {len(evidence)} 条）\n" + "\n\n".join(evidence)
+        common_system = (
+            "你是一名严谨的企业研究报告撰写人。只可使用用户确认的研究计划、现有报告和资料快照；"
+            "不可联网、不可补充未在资料中出现的具体事实或数据。不要输出 <think> 标签。每个事实性判断"
+            "尽量以 [资料:id] 标识其对应资料；证据不足时明确写为待验证判断。"
+        )
+        if action == "followup":
+            return [
+                {"role": "system", "content": common_system + "请直接回答报告追问，不要重写整篇报告。"},
+                {
+                    "role": "user",
+                    "content": (
+                        f"# 当前报告\n{job.get('original_report') or ''}\n\n{source_context}\n\n"
+                        f"# 用户问题\n{instruction}"
+                    ),
+                },
+            ]
+        if action == "rewrite":
+            length_hint = f"目标篇幅约 {target_length} 字。" if target_length else "篇幅与原文大体相当。"
+            return [
+                {
+                    "role": "system",
+                    "content": common_system + "输出仅为改写后的完整 Markdown 报告，不要解释改写过程。",
+                },
+                {
+                    "role": "user",
+                    "content": (
+                        f"请全文改写下列企业研究报告。写作风格：{style}；{length_hint}\n"
+                        f"改写要求：{instruction or '优化结构、逻辑与表达，并保留事实和引用。'}\n\n"
+                        f"# 研究计划\n{job['plan_snapshot']}\n\n"
+                        f"# 原报告\n{job.get('original_report') or ''}\n\n{source_context}"
+                    ),
+                },
+            ]
+        length_hint = f"报告目标篇幅约 {target_length} 字。" if target_length else ""
+        return [
+            {
+                "role": "system",
+                "content": common_system + "输出仅为完整 Markdown 报告，不要解释写作过程。",
+            },
+            {
+                "role": "user",
+                "content": (
+                    "请按下列研究计划撰写企业研究报告。保留计划中有意义的章节层级，"
+                    f"内容应连贯、可核验，结论与建议要说明证据边界。写作风格：{style}。{length_hint}\n\n"
+                    f"# 研究计划\n{job['plan_snapshot']}\n\n"
+                    + source_context
+                ),
+            },
+        ]
+
+    @staticmethod
+    def _summary(
+        report: str,
+        source_count: int,
+        action: str,
+        *,
+        original_report: str = "",
+        instruction: str = "",
+    ) -> str:
+        headings = sum(1 for line in report.splitlines() if line.lstrip().startswith("#"))
+        if action == "followup":
+            return f"已基于当前报告和 {source_count} 条已确认资料完成追问回答，共 {len(report)} 字。"
+        if action == "rewrite":
+            before_blocks = [block.strip() for block in original_report.split("\n\n") if block.strip()]
+            after_blocks = [block.strip() for block in report.split("\n\n") if block.strip()]
+            matcher = SequenceMatcher(a=before_blocks, b=after_blocks, autojunk=False)
+            changed_blocks = sum(
+                max(before_end - before_start, after_end - after_start)
+                for tag, before_start, before_end, after_start, after_end in matcher.get_opcodes()
+                if tag != "equal"
+            )
+            before_headings = sum(1 for line in original_report.splitlines() if line.lstrip().startswith("#"))
+            request = instruction.strip()[:120] or "优化结构、逻辑与表达，并保留事实和引用"
+            return (
+                f"已依据 {source_count} 条已确认资料完成全文改写：原文 {len(original_report)} 字，"
+                f"改写后 {len(report)} 字，约调整 {changed_blocks} 个内容块，标题节点 "
+                f"{before_headings}→{headings}。改写要求：{request}。"
+            )
+        return f"已依据 {source_count} 条已确认资料完成报告，共 {len(report)} 字，包含 {headings} 个标题层级节点。"
+
+    async def _run(self, job_id: str, user_id: str) -> None:
+        job = await self._jobs.get_job(job_id, user_id)
+        if job is None or job["status"] in _TERMINAL:
+            return
+        if not await self._jobs.claim_job(job_id, user_id):
+            return
+        if await self._jobs.transition_job(
+            job_id, user_id, expected_statuses={"running"},
+            data={"phase": "understanding", "error_message": None, "report_markdown": ""},
+        ) is None:
+            return
+        await self._hub.publish(job_id, {"type": "phase", "phase": "understanding", "status": "running"})
+        report_parts: list[str] = []
+        last_persist = 0.0
+        try:
+            await asyncio.sleep(0)  # give the client an observable understanding phase
+            if await self._set_phase(job_id, user_id, "writing") is None:
+                return
+            config = DeepResearchConfig(
+                mode="basic", language="zh-CN", smart_model=job.get("model_name") or None,
+                custom_outline=str(job["plan_snapshot"])[:4_000], max_context_words=24_000,
+            ).clamp()
+            backend = DeerFlowCompletionBackend(config)
+
+            async def on_delta(delta: str) -> None:
+                nonlocal last_persist
+                report_parts.append(delta)
+                await self._hub.publish(job_id, {"type": "report_delta", "delta": delta})
+                now = time.monotonic()
+                if now - last_persist >= 1.25:
+                    last_persist = now
+                    persisted = await self._jobs.transition_job(
+                        job_id, user_id, expected_statuses={"running"},
+                        data={"report_markdown": "".join(report_parts)},
+                    )
+                    if persisted is None:
+                        raise asyncio.CancelledError
+
+            async def on_reasoning(delta: str) -> None:
+                # Thought traces are deliberately separate from Markdown. The
+                # client may choose to show them in a collapsed activity item.
+                await self._hub.publish(job_id, {"type": "reasoning_delta", "delta": delta})
+
+            result = await backend.stream_complete(
+                model_role="smart", messages=self._messages(job), operation="enterprise_research_write_report",
+                on_delta=on_delta, on_reasoning=on_reasoning,
+            )
+            report = result.text.strip()
+            if not report:
+                raise RuntimeError("模型没有返回可写入的报告内容")
+            if await self._set_phase(job_id, user_id, "reviewing") is None:
+                return
+            action = str(job.get("action") or "initial")
+            summary = self._summary(
+                report,
+                len(job.get("sources") or []),
+                action,
+                original_report=str(job.get("original_report") or ""),
+                instruction=str(job.get("instruction") or ""),
+            )
+            completed = await self._jobs.transition_job(
+                job_id, user_id, expected_statuses={"running"},
+                data={"status": "completed", "phase": "completed", "report_markdown": report, "report_summary": summary, "model_name": result.model},
+            )
+            if completed is None:
+                # A different process accepted cancellation while this worker
+                # was between its last token and commit. Never overwrite it.
+                return
+            task_update: dict[str, Any] = {"active_report_job_id": None}
+            if action != "followup":
+                task_update.update(
+                    {"report_markdown": report, "report_summary": summary, "report_model_name": result.model}
+                )
+            await self._tasks.update_task(job["task_id"], user_id, task_update)
+            await self._hub.publish(job_id, {"type": "completed", "job": completed})
+        except asyncio.CancelledError:
+            cancelled = await self._jobs.transition_job(
+                job_id, user_id, expected_statuses={"queued", "running"},
+                data={"status": "cancelled", "phase": "cancelled"},
+            )
+            if cancelled is not None:
+                await self._tasks.update_task(job["task_id"], user_id, {"active_report_job_id": None})
+                await self._hub.publish(job_id, {"type": "cancelled", "job": cancelled})
+            raise
+        except Exception as exc:  # noqa: BLE001 - turn model/provider failures into a visible job state
+            logger.exception("enterprise research report job %s failed", job_id)
+            message = str(exc)[:2_000] or "报告生成失败，请稍后重试。"
+            failed = await self._jobs.transition_job(
+                job_id, user_id, expected_statuses={"running"},
+                data={"status": "failed", "phase": "failed", "report_markdown": "".join(report_parts), "error_message": message},
+            )
+            if failed is not None:
+                await self._tasks.update_task(job["task_id"], user_id, {"active_report_job_id": None})
+                await self._hub.publish(job_id, {"type": "failed", "job": failed})

offline-backend-20260512/backend/app/gateway/routers/enterprise_research.py

@@ -0,0 +1,710 @@
+"""Independent enterprise-research workbench API.
+
+This router intentionally does not alter the established ``/api/deep-research``
+workflow.  It stores a small owner-scoped research task and calls WeKnora through
+the same DeerFlow-owned authorization mapping used by the knowledge-base page.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from typing import Any, Literal
+from urllib.parse import urlencode
+from uuid import uuid4
+
+from fastapi import APIRouter, HTTPException, Query, Request, status
+from pydantic import BaseModel, Field
+from starlette.responses import StreamingResponse
+
+from app.gateway.deps import get_optional_user_from_request
+from deerflow.agents.deep_research.adapters.llm import DeerFlowCompletionBackend
+from deerflow.agents.deep_research.config import DeepResearchConfig
+from deerflow.integrations.weknora.client import WeKnoraError
+from deerflow.integrations.weknora.runtime import build_weknora_client, get_resolved_llmwiki_runtime
+from deerflow.persistence.enterprise_research import EnterpriseResearchReportJobStore, EnterpriseResearchTaskStore
+from deerflow.persistence.llmwiki import LlmWikiStore
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/api/enterprise-research", tags=["enterprise-research"])
+
+TaskStatus = Literal["draft", "collecting", "ready", "partial", "failed"]
+
+
+class EnterpriseResearchEvidence(BaseModel):
+    chunk_id: str = Field(default="", max_length=256)
+    content: str = Field(default="", max_length=4_000)
+    knowledge_id: str = Field(default="", max_length=256)
+    knowledge_base_id: str = Field(default="", max_length=256)
+    knowledge_base_name: str = Field(default="", max_length=512)
+    title: str = Field(default="", max_length=1_024)
+    filename: str = Field(default="", max_length=1_024)
+    score: float = 0.0
+    chunk_index: int = 0
+    url: str = Field(default="", max_length=4_096)
+
+
+class EnterpriseResearchTaskResponse(BaseModel):
+    id: str
+    subject: str
+    focus: str
+    template_id: str
+    knowledge_base_ids: list[str]
+    queries: list[str]
+    selected_source_ids: list[str] = Field(default_factory=list)
+    sources: list[EnterpriseResearchEvidence]
+    plan_markdown: str | None = None
+    report_markdown: str | None = None
+    report_summary: str | None = None
+    report_model_name: str | None = None
+    active_report_job_id: str | None = None
+    status: TaskStatus
+    warning: str | None = None
+    created_at: str | None = None
+    updated_at: str | None = None
+
+
+class EnterpriseResearchTaskListResponse(BaseModel):
+    tasks: list[EnterpriseResearchTaskResponse]
+
+
+class EnterpriseResearchTaskCreate(BaseModel):
+    subject: str = Field(..., min_length=1, max_length=180)
+    focus: str = Field(default="", max_length=800)
+    template_id: str = Field(default="topic", min_length=1, max_length=64)
+    knowledge_base_ids: list[str] = Field(default_factory=list, max_length=100)
+    queries: list[str] = Field(..., min_length=1, max_length=8)
+
+
+class EnterpriseResearchPlanUpdate(BaseModel):
+    plan_markdown: str = Field(..., min_length=1, max_length=16_000)
+    selected_source_ids: list[str] = Field(default_factory=list, max_length=48)
+
+
+class EnterpriseResearchAiPlanCreate(BaseModel):
+    model_name: str | None = Field(default=None, max_length=256)
+    instruction: str = Field(default="", max_length=2_000)
+
+
+class EnterpriseResearchReportJobCreate(BaseModel):
+    model_name: str | None = Field(default=None, max_length=256)
+    action: Literal["initial", "rewrite", "followup"] = "initial"
+    instruction: str = Field(default="", max_length=4_000)
+    style: Literal["formal", "analytical", "concise"] = "formal"
+    target_length: int | None = Field(default=None, ge=800, le=30_000)
+
+
+class EnterpriseResearchReportJobResponse(BaseModel):
+    id: str
+    task_id: str
+    status: Literal["queued", "running", "completed", "failed", "cancelled"]
+    phase: str
+    model_name: str | None = None
+    action: Literal["initial", "rewrite", "followup"] = "initial"
+    instruction: str | None = None
+    style: str | None = None
+    target_length: int | None = None
+    original_report: str | None = None
+    report_markdown: str = ""
+    report_summary: str | None = None
+    error_message: str | None = None
+    created_at: str | None = None
+    updated_at: str | None = None
+
+
+class EnterpriseResearchReportJobListResponse(BaseModel):
+    jobs: list[EnterpriseResearchReportJobResponse]
+
+
+def _task_store(request: Request) -> EnterpriseResearchTaskStore:
+    store = getattr(request.app.state, "enterprise_research_task_store", None)
+    if store is None:
+        raise HTTPException(status_code=503, detail="企业深度研究任务存储暂不可用")
+    return store
+
+
+def _report_job_store(request: Request) -> EnterpriseResearchReportJobStore:
+    store = getattr(request.app.state, "enterprise_research_report_job_store", None)
+    if store is None:
+        raise HTTPException(status_code=503, detail="企业研究报告任务存储暂不可用")
+    return store
+
+
+def _report_executor(request: Request):
+    executor = getattr(request.app.state, "enterprise_research_report_executor", None)
+    if executor is None:
+        raise HTTPException(status_code=503, detail="企业研究报告执行器暂不可用")
+    return executor
+
+
+def _report_start_lock(request: Request) -> asyncio.Lock:
+    lock = getattr(request.app.state, "enterprise_research_report_start_lock", None)
+    if lock is None:
+        lock = asyncio.Lock()
+        request.app.state.enterprise_research_report_start_lock = lock
+    return lock
+
+
+def _llmwiki_store(request: Request) -> LlmWikiStore:
+    store = getattr(request.app.state, "llmwiki_store", None)
+    if store is None:
+        raise HTTPException(status_code=503, detail="知识库权限存储暂不可用")
+    return store
+
+
+async def _actor(request: Request) -> tuple[str, bool]:
+    user = await get_optional_user_from_request(request)
+    if user is None:
+        return "default", True
+    return str(user.id), getattr(user, "system_role", None) == "admin"
+
+
+def _response(row: dict[str, Any]) -> EnterpriseResearchTaskResponse:
+    return EnterpriseResearchTaskResponse(
+        id=str(row["id"]),
+        subject=str(row.get("subject") or ""),
+        focus=str(row.get("focus") or ""),
+        template_id=str(row.get("template_id") or "topic"),
+        knowledge_base_ids=[str(item) for item in row.get("knowledge_base_ids") or []],
+        queries=[str(item) for item in row.get("queries") or []],
+        selected_source_ids=[str(item) for item in row.get("selected_source_ids") or []],
+        sources=[EnterpriseResearchEvidence.model_validate(item) for item in row.get("sources") or [] if isinstance(item, dict)],
+        plan_markdown=str(row["plan_markdown"]) if row.get("plan_markdown") else None,
+        report_markdown=str(row["report_markdown"]) if row.get("report_markdown") else None,
+        report_summary=str(row["report_summary"]) if row.get("report_summary") else None,
+        report_model_name=str(row["report_model_name"]) if row.get("report_model_name") else None,
+        active_report_job_id=str(row["active_report_job_id"]) if row.get("active_report_job_id") else None,
+        status=row.get("status") if row.get("status") in {"draft", "collecting", "ready", "partial", "failed"} else "failed",
+        warning=str(row["warning"]) if row.get("warning") else None,
+        created_at=row.get("created_at"),
+        updated_at=row.get("updated_at"),
+    )
+
+
+def _report_job_response(row: dict[str, Any]) -> EnterpriseResearchReportJobResponse:
+    job_status = str(row.get("status") or "failed")
+    if job_status not in {"queued", "running", "completed", "failed", "cancelled"}:
+        job_status = "failed"
+    action = str(row.get("action") or "initial")
+    if action not in {"initial", "rewrite", "followup"}:
+        action = "initial"
+    return EnterpriseResearchReportJobResponse(
+        id=str(row["id"]), task_id=str(row["task_id"]), status=job_status, phase=str(row.get("phase") or job_status),
+        model_name=str(row["model_name"]) if row.get("model_name") else None,
+        action=action,
+        instruction=str(row["instruction"]) if row.get("instruction") else None,
+        style=str(row["style"]) if row.get("style") else None,
+        target_length=int(row["target_length"]) if row.get("target_length") else None,
+        original_report=str(row["original_report"]) if row.get("original_report") else None,
+        report_markdown=str(row.get("report_markdown") or ""),
+        report_summary=str(row["report_summary"]) if row.get("report_summary") else None,
+        error_message=str(row["error_message"]) if row.get("error_message") else None,
+        created_at=row.get("created_at"), updated_at=row.get("updated_at"),
+    )
+
+
+def _normalise_ids(values: list[str]) -> list[str]:
+    seen: set[str] = set()
+    result: list[str] = []
+    for value in values:
+        item = str(value).strip()
+        if item and item not in seen:
+            seen.add(item)
+            result.append(item)
+    return result[:100]
+
+
+def _normalise_queries(values: list[str]) -> list[str]:
+    seen: set[str] = set()
+    result: list[str] = []
+    for value in values:
+        item = str(value).strip()[:512]
+        if item and item not in seen:
+            seen.add(item)
+            result.append(item)
+    return result[:8]
+
+
+def _default_plan(task: dict[str, Any], selected_source_ids: list[str]) -> str:
+    """Build a transparent editable plan; report prose is a later, separate job."""
+
+    subject = str(task.get("subject") or "研究主题")
+    focus = str(task.get("focus") or "")
+    template_id = str(task.get("template_id") or "topic")
+    if template_id == "event":
+        chapters = [
+            "一、事件边界与时间线",
+            "二、相关主体与立场",
+            "三、影响传导与风险研判",
+            "四、后续观察与应对建议",
+        ]
+    elif template_id == "policy":
+        chapters = [
+            "一、政策与行业背景",
+            "二、现状、数据与变化趋势",
+            "三、传导机制与关键影响",
+            "四、风险、机会与行动建议",
+        ]
+    else:
+        chapters = [
+            "一、研究对象与事实背景",
+            "二、关键问题、观点与分歧",
+            "三、影响、风险与机会",
+            "四、结论、待验证问题与建议",
+        ]
+    sources_by_id = {str(item.get("chunk_id") or ""): item for item in task.get("sources") or [] if isinstance(item, dict)}
+    selected_titles = [
+        str(sources_by_id[item].get("title") or sources_by_id[item].get("filename") or "未命名资料")
+        for item in selected_source_ids
+        if item in sources_by_id
+    ]
+    source_lines = "\n".join(f"- {title}" for title in selected_titles[:12]) or "- 暂无已选资料"
+    scope = focus or "围绕主题的事实、影响与可验证判断"
+    return (
+        f"# {subject}研究计划\n\n"
+        "## 研究目标\n"
+        f"围绕“{subject}”形成可追溯、可核验的研究结论；重点范围：{scope}。\n\n"
+        "## 核心问题\n"
+        "1. 当前已知事实、背景和时间范围是什么？\n"
+        f"2. 与“{subject}”相关的主要主体、观点或变化是什么？\n"
+        "3. 证据能够支持哪些影响、风险、机会或待验证判断？\n\n"
+        "## 报告结构\n"
+        + "\n".join(f"{chapter}\n- 说明本章需回答的核心问题，并在正文中标记资料依据。" for chapter in chapters)
+        + "\n\n## 已选证据\n"
+        + source_lines
+        + "\n\n## 写作约束\n"
+        "- 仅使用已选资料作为事实依据；不补充未经验证的外部事实。\n"
+        "- 对资料不足或观点相互矛盾之处明确标记，避免将推测写成结论。\n"
+        "- 生成报告前可继续编辑本计划与已选证据。\n"
+    )
+
+
+def _validate_selected_source_ids(task: dict[str, Any], values: list[str]) -> list[str]:
+    available_ids = {str(item.get("chunk_id") or "") for item in task.get("sources") or [] if isinstance(item, dict)}
+    selected_ids = _normalise_ids(values)[:48]
+    if any(item not in available_ids for item in selected_ids):
+        raise HTTPException(status_code=422, detail="选中的资料已变化，请刷新后重新选择")
+    return selected_ids
+
+
+async def _visible_knowledge_bases(request: Request, task: dict[str, Any], user_id: str, is_admin: bool) -> list[dict[str, Any]]:
+    store = _llmwiki_store(request)
+    selected_ids = _normalise_ids(task.get("knowledge_base_ids") or [])
+    if not selected_ids:
+        return await store.list_visible(user_id, scope="all", is_admin=False)
+    rows: list[dict[str, Any]] = []
+    for mapping_id in selected_ids:
+        row = await store.get_authorized(mapping_id, user_id, write=False, is_admin=is_admin)
+        if row is None:
+            # Never reveal whether this belongs to a different account.
+            raise HTTPException(status_code=404, detail="知识库不存在或无访问权限")
+        rows.append(row)
+    return rows
+
+
+def _client(request: Request):
+    runtime = get_resolved_llmwiki_runtime(getattr(request.app.state, "config", None))
+    if not runtime.weknora_enabled:
+        raise HTTPException(status_code=409, detail="WeKnora 知识服务尚未启用")
+    try:
+        return build_weknora_client(runtime)
+    except RuntimeError as exc:
+        raise HTTPException(status_code=503, detail=str(exc)) from None
+
+
+def _evidence_view(result: dict[str, Any], mappings: dict[str, dict[str, Any]]) -> dict[str, Any] | None:
+    remote_base_id = str(result.get("knowledge_base_id") or "")
+    mapping = mappings.get(remote_base_id)
+    if mapping is None:
+        return None
+    params = urlencode({"knowledge_base_id": mapping["id"], "chunk_id": result.get("chunk_id") or ""})
+    return {
+        "chunk_id": str(result.get("chunk_id") or ""),
+        "content": str(result.get("content") or "")[:4_000],
+        "knowledge_id": str(result.get("knowledge_id") or ""),
+        "knowledge_base_id": str(mapping["id"]),
+        "knowledge_base_name": str(mapping.get("name") or ""),
+        "title": str(result.get("title") or ""),
+        "filename": str(result.get("filename") or ""),
+        "score": float(result.get("score") or 0),
+        "chunk_index": int(result.get("chunk_index") or 0),
+        "url": f"/weknora-source-preview?{params}",
+    }
+
+
+def _dedupe_evidence(results: list[dict[str, Any]], mappings: dict[str, dict[str, Any]]) -> list[dict[str, Any]]:
+    seen: set[str] = set()
+    evidence: list[dict[str, Any]] = []
+    for result in results:
+        item = _evidence_view(result, mappings)
+        if item is None:
+            continue
+        key = item["chunk_id"] or f"{item['knowledge_id']}:{item['chunk_index']}:{item['title']}"
+        if key in seen:
+            continue
+        seen.add(key)
+        evidence.append(item)
+        if len(evidence) >= 48:
+            break
+    return evidence
+
+
+@router.get("/tasks", response_model=EnterpriseResearchTaskListResponse)
+async def list_tasks(request: Request, limit: int = Query(default=20, ge=1, le=50)) -> EnterpriseResearchTaskListResponse:
+    user_id, _ = await _actor(request)
+    rows = await _task_store(request).list_tasks(user_id, limit=limit)
+    return EnterpriseResearchTaskListResponse(tasks=[_response(row) for row in rows])
+
+
+@router.post("/tasks", response_model=EnterpriseResearchTaskResponse, status_code=status.HTTP_201_CREATED)
+async def create_task(request: Request, body: EnterpriseResearchTaskCreate) -> EnterpriseResearchTaskResponse:
+    user_id, _ = await _actor(request)
+    queries = _normalise_queries(body.queries)
+    if not queries:
+        raise HTTPException(status_code=422, detail="至少需要一个有效检索词")
+    row = await _task_store(request).create_task(
+        {
+            "id": f"ert_{uuid4().hex}",
+            "user_id": user_id,
+            "subject": body.subject.strip(),
+            "focus": body.focus.strip(),
+            "template_id": body.template_id.strip(),
+            "knowledge_base_ids": _normalise_ids(body.knowledge_base_ids),
+            "queries": queries,
+            "status": "draft",
+        }
+    )
+    return _response(row)
+
+
+@router.post("/tasks/{task_id}/collect", response_model=EnterpriseResearchTaskResponse)
+async def collect_task(request: Request, task_id: str) -> EnterpriseResearchTaskResponse:
+    """Collect a bounded, user-authorized WeKnora evidence snapshot for a task."""
+
+    user_id, is_admin = await _actor(request)
+    task_store = _task_store(request)
+    task = await task_store.get_task(task_id, user_id)
+    if task is None:
+        raise HTTPException(status_code=404, detail="研究任务不存在")
+    task = await task_store.update_task(task_id, user_id, {"status": "collecting", "warning": None})
+    assert task is not None
+
+    try:
+        bases = await _visible_knowledge_bases(request, task, user_id, is_admin)
+        if not bases:
+            row = await task_store.update_task(task_id, user_id, {"status": "failed", "warning": "当前账号没有可用知识库。"})
+            return _response(row or task)
+        client = _client(request)
+        remote_ids = [str(row["weknora_id"]) for row in bases]
+        outcomes = await asyncio.gather(
+            *(client.search(query, remote_ids) for query in task["queries"]),
+            return_exceptions=True,
+        )
+        raw_results: list[dict[str, Any]] = []
+        failed_queries = 0
+        for outcome in outcomes:
+            if isinstance(outcome, Exception):
+                failed_queries += 1
+                logger.warning("Enterprise research query failed: %s", outcome)
+            else:
+                raw_results.extend(outcome)
+        sources = _dedupe_evidence(raw_results, {str(row["weknora_id"]): row for row in bases})
+        if not sources:
+            task_status: TaskStatus = "failed"
+            warning = "未检索到可用证据，请调整研究主题或检查知识库解析状态。"
+        elif failed_queries:
+            task_status = "partial"
+            warning = f"{failed_queries} 个检索方向暂时不可用，已保留其余结果。"
+        else:
+            task_status = "ready"
+            warning = None
+        row = await task_store.update_task(task_id, user_id, {"status": task_status, "sources": sources, "warning": warning})
+        return _response(row or task)
+    except HTTPException:
+        raise
+    except WeKnoraError as exc:
+        await task_store.update_task(task_id, user_id, {"status": "failed", "warning": str(exc)})
+        raise HTTPException(status_code=exc.status_code, detail=str(exc)) from None
+    except Exception:
+        logger.exception("Enterprise research collection failed for %s", task_id)
+        await task_store.update_task(task_id, user_id, {"status": "failed", "warning": "证据收集出现异常，请稍后重试。"})
+        raise HTTPException(status_code=502, detail="证据收集失败，请稍后重试") from None
+
+
+@router.post("/tasks/{task_id}/plan/draft", response_model=EnterpriseResearchTaskResponse)
+async def create_plan_draft(request: Request, task_id: str) -> EnterpriseResearchTaskResponse:
+    """Create a deterministic, editable research-plan draft from persisted evidence."""
+
+    user_id, _ = await _actor(request)
+    task_store = _task_store(request)
+    task = await task_store.get_task(task_id, user_id)
+    if task is None:
+        raise HTTPException(status_code=404, detail="研究任务不存在")
+    if not task.get("sources"):
+        raise HTTPException(status_code=409, detail="请先完成证据收集，再生成研究计划")
+    selected_source_ids = _validate_selected_source_ids(task, [
+        str(item.get("chunk_id") or "") for item in task["sources"][:24] if isinstance(item, dict)
+    ])
+    row = await task_store.update_task(
+        task_id,
+        user_id,
+        {
+            "selected_source_ids": selected_source_ids,
+            "plan_markdown": _default_plan(task, selected_source_ids),
+        },
+    )
+    return _response(row or task)
+
+
+@router.put("/tasks/{task_id}/plan", response_model=EnterpriseResearchTaskResponse)
+async def update_plan(request: Request, task_id: str, body: EnterpriseResearchPlanUpdate) -> EnterpriseResearchTaskResponse:
+    """Persist the user's selected evidence and editable plan before report generation."""
+
+    user_id, _ = await _actor(request)
+    task_store = _task_store(request)
+    task = await task_store.get_task(task_id, user_id)
+    if task is None:
+        raise HTTPException(status_code=404, detail="研究任务不存在")
+    if not task.get("sources"):
+        raise HTTPException(status_code=409, detail="当前任务没有可用于编排的证据")
+    selected_source_ids = _validate_selected_source_ids(task, body.selected_source_ids)
+    plan_markdown = body.plan_markdown.strip()
+    if not plan_markdown:
+        raise HTTPException(status_code=422, detail="研究计划不能为空")
+    row = await task_store.update_task(
+        task_id,
+        user_id,
+        {"selected_source_ids": selected_source_ids, "plan_markdown": plan_markdown},
+    )
+    return _response(row or task)
+
+
+@router.post("/tasks/{task_id}/plan/ai", response_model=EnterpriseResearchTaskResponse)
+async def create_ai_plan(
+    request: Request, task_id: str, body: EnterpriseResearchAiPlanCreate
+) -> EnterpriseResearchTaskResponse:
+    """Use the configured model to improve a plan, without collecting new evidence."""
+
+    user_id, _ = await _actor(request)
+    task_store = _task_store(request)
+    task = await task_store.get_task(task_id, user_id)
+    if task is None:
+        raise HTTPException(status_code=404, detail="研究任务不存在")
+    if not task.get("sources"):
+        raise HTTPException(status_code=409, detail="请先完成证据收集")
+    selected_ids = list(task.get("selected_source_ids") or [])
+    if not selected_ids:
+        selected_ids = [
+            str(item.get("chunk_id") or "")
+            for item in task["sources"][:24]
+            if isinstance(item, dict) and item.get("chunk_id")
+        ]
+    selected_ids = _validate_selected_source_ids(task, selected_ids)
+    selected = {
+        str(item.get("chunk_id") or ""): item
+        for item in task.get("sources") or []
+        if isinstance(item, dict)
+    }
+    evidence_titles = "\n".join(
+        f"- {selected[item].get('title') or selected[item].get('filename') or item}"
+        for item in selected_ids
+        if item in selected
+    )
+    current = str(task.get("plan_markdown") or _default_plan(task, selected_ids))
+    backend = DeerFlowCompletionBackend(
+        DeepResearchConfig(mode="basic", language="zh-CN", smart_model=(body.model_name or "").strip() or None).clamp()
+    )
+    try:
+        result = await backend.stream_complete(
+            model_role="smart",
+            operation="enterprise_research_plan",
+            messages=[
+                {
+                    "role": "system",
+                    "content": (
+                        "你是企业研究规划师。仅优化研究目标、核心问题、章节结构和证据使用计划；"
+                        "不要撰写报告正文，不要联网，不要虚构资料。仅输出 Markdown 研究计划。"
+                    ),
+                },
+                {
+                    "role": "user",
+                    "content": (
+                        f"研究主题：{task['subject']}\n重点范围：{task.get('focus') or '未限定'}\n"
+                        f"用户补充要求：{body.instruction or '提高结构完整性和章节关联性'}\n\n"
+                        f"当前计划：\n{current}\n\n已选资料标题：\n{evidence_titles}"
+                    ),
+                },
+            ],
+        )
+    except Exception as exc:
+        logger.exception("enterprise research AI plan generation failed task=%s", task_id)
+        raise HTTPException(status_code=502, detail="研究计划生成失败，请检查模型配置后重试") from exc
+    plan = result.text.strip()
+    if plan.startswith("```"):
+        lines = plan.splitlines()[1:]
+        if lines and lines[-1].strip().startswith("```"):
+            lines.pop()
+        plan = "\n".join(lines).strip()
+    if not plan:
+        raise HTTPException(status_code=502, detail="模型没有返回可用研究计划")
+    row = await task_store.update_task(
+        task_id, user_id, {"plan_markdown": plan[:16_000], "selected_source_ids": selected_ids}
+    )
+    return _response(row or task)
+
+
+@router.get("/tasks/{task_id}/report-jobs", response_model=EnterpriseResearchReportJobListResponse)
+async def list_report_jobs(
+    request: Request, task_id: str, limit: int = Query(default=10, ge=1, le=20)
+) -> EnterpriseResearchReportJobListResponse:
+    user_id, _ = await _actor(request)
+    if await _task_store(request).get_task(task_id, user_id) is None:
+        raise HTTPException(status_code=404, detail="研究任务不存在")
+    rows = await _report_job_store(request).list_jobs(task_id, user_id, limit=limit)
+    return EnterpriseResearchReportJobListResponse(jobs=[_report_job_response(row) for row in rows])
+
+
+@router.post("/tasks/{task_id}/report-jobs", response_model=EnterpriseResearchReportJobResponse, status_code=status.HTTP_201_CREATED)
+async def create_report_job(
+    request: Request, task_id: str, body: EnterpriseResearchReportJobCreate
+) -> EnterpriseResearchReportJobResponse:
+    """Freeze the approved plan/evidence, then launch an independent report job."""
+
+    user_id, _ = await _actor(request)
+    task_store = _task_store(request)
+    job_store = _report_job_store(request)
+    executor = _report_executor(request)
+    async with _report_start_lock(request):
+        task = await task_store.get_task(task_id, user_id)
+        if task is None:
+            raise HTTPException(status_code=404, detail="研究任务不存在")
+        active = await job_store.get_active_for_task(task_id, user_id)
+        if active is not None:
+            await executor.ensure_started(active["id"], user_id)
+            return _report_job_response(active)
+        plan = str(task.get("plan_markdown") or "").strip()
+        selected_ids = _validate_selected_source_ids(task, list(task.get("selected_source_ids") or []))
+        if not plan:
+            raise HTTPException(status_code=409, detail="请先生成并保存研究计划")
+        if not selected_ids:
+            raise HTTPException(status_code=409, detail="请至少选择一条资料后再撰写报告")
+        sources_by_id = {
+            str(item.get("chunk_id") or ""): item
+            for item in task.get("sources") or []
+            if isinstance(item, dict)
+        }
+        sources = [sources_by_id[source_id] for source_id in selected_ids if source_id in sources_by_id]
+        if not sources:
+            raise HTTPException(status_code=409, detail="已选资料不可用，请重新保存研究计划")
+        if body.action in {"rewrite", "followup"} and not task.get("report_markdown"):
+            raise HTTPException(status_code=409, detail="当前任务还没有可供改写或追问的报告")
+        if body.action == "followup" and not body.instruction.strip():
+            raise HTTPException(status_code=422, detail="请填写报告追问")
+        row = await job_store.create_job(
+            {
+                "id": f"erj_{uuid4().hex}", "task_id": task_id, "user_id": user_id,
+                "model_name": (body.model_name or "").strip() or None,
+                "action": body.action, "instruction": body.instruction.strip(), "style": body.style,
+                "target_length": body.target_length, "plan_snapshot": plan, "sources": sources,
+                "original_report": task.get("report_markdown") if body.action in {"rewrite", "followup"} else None,
+            }
+        )
+        await task_store.update_task(task_id, user_id, {"active_report_job_id": row["id"]})
+    await executor.ensure_started(row["id"], user_id)
+    return _report_job_response(row)
+
+
+@router.get("/tasks/{task_id}/report-jobs/{job_id}/stream")
+async def stream_report_job(request: Request, task_id: str, job_id: str) -> StreamingResponse:
+    """SSE stream: durable draft snapshot first, then live Markdown/reasoning deltas."""
+
+    user_id, _ = await _actor(request)
+    job = await _report_job_store(request).get_job(job_id, user_id)
+    if job is None or job.get("task_id") != task_id:
+        raise HTTPException(status_code=404, detail="报告任务不存在")
+    if job["status"] == "queued":
+        await _report_executor(request).ensure_started(job_id, user_id)
+
+    async def body():
+        async for event in _report_executor(request).stream_events(job_id, user_id):
+            yield "event: enterprise_research\n"
+            yield "data: " + json.dumps(event, ensure_ascii=False, separators=(",", ":")) + "\n\n"
+
+    return StreamingResponse(
+        body(), media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+    )
+
+
+@router.post(
+    "/tasks/{task_id}/report-jobs/{job_id}/cancel",
+    response_model=EnterpriseResearchReportJobResponse,
+)
+async def cancel_report_job(request: Request, task_id: str, job_id: str) -> EnterpriseResearchReportJobResponse:
+    user_id, _ = await _actor(request)
+    job = await _report_job_store(request).get_job(job_id, user_id)
+    if job is None or job.get("task_id") != task_id:
+        raise HTTPException(status_code=404, detail="报告任务不存在")
+    cancelled = await _report_executor(request).cancel(job_id, user_id)
+    return _report_job_response(cancelled or job)
+
+
+@router.post(
+    "/tasks/{task_id}/report-jobs/{job_id}/restore",
+    response_model=EnterpriseResearchTaskResponse,
+)
+async def restore_report_version(request: Request, task_id: str, job_id: str) -> EnterpriseResearchTaskResponse:
+    """Restore the before-image frozen by a completed full-report rewrite."""
+
+    user_id, _ = await _actor(request)
+    job = await _report_job_store(request).get_job(job_id, user_id)
+    if job is None or job.get("task_id") != task_id:
+        raise HTTPException(status_code=404, detail="报告任务不存在")
+    original = str(job.get("original_report") or "")
+    if job.get("status") != "completed" or job.get("action") != "rewrite" or not original.strip():
+        raise HTTPException(status_code=409, detail="该任务没有可恢复的改写前版本")
+    if await _report_job_store(request).get_active_for_task(task_id, user_id) is not None:
+        raise HTTPException(status_code=409, detail="报告任务正在执行，请完成或停止后再恢复版本")
+    task_store = _task_store(request)
+    current = await task_store.get_task(task_id, user_id)
+    if current is None:
+        raise HTTPException(status_code=404, detail="研究任务不存在")
+    current_report = str(current.get("report_markdown") or "")
+    if not current_report.strip():
+        raise HTTPException(status_code=409, detail="当前没有可替换的报告")
+    # The restore itself is a versioned operation: its before-image allows the
+    # user to reverse a mistaken restore through the same version UI.
+    await _report_job_store(request).create_job(
+        {
+            "id": f"erj_{uuid4().hex}", "task_id": task_id, "user_id": user_id,
+            "status": "completed", "phase": "completed", "action": "rewrite",
+            "instruction": f"恢复报告任务 {job_id} 的改写前版本", "style": "formal",
+            "model_name": current.get("report_model_name"),
+            "plan_snapshot": str(current.get("plan_markdown") or ""),
+            "sources": [
+                item for item in current.get("sources") or []
+                if isinstance(item, dict) and item.get("chunk_id") in set(current.get("selected_source_ids") or [])
+            ],
+            "original_report": current_report, "report_markdown": original,
+            "report_summary": f"已恢复到报告任务 {job_id} 的改写前版本。",
+        }
+    )
+    task = await task_store.update_task(
+        task_id,
+        user_id,
+        {
+            "report_markdown": original,
+            "report_summary": f"已恢复到报告任务 {job_id} 的改写前版本。",
+            "active_report_job_id": None,
+        },
+    )
+    assert task is not None
+    return _response(task)
+
+
+@router.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_task(request: Request, task_id: str) -> None:
+    user_id, _ = await _actor(request)
+    if not await _task_store(request).delete_task(task_id, user_id):
+        raise HTTPException(status_code=404, detail="研究任务不存在")

offline-backend-20260512/backend/packages/harness/deerflow/persistence/enterprise_research/__init__.py

@@ -0,0 +1,31 @@
+"""Enterprise-research task persistence."""
+
+from deerflow.persistence.enterprise_research.base import EnterpriseResearchReportJobStore, EnterpriseResearchTaskStore
+from deerflow.persistence.enterprise_research.memory import MemoryEnterpriseResearchReportJobStore, MemoryEnterpriseResearchTaskStore
+from deerflow.persistence.enterprise_research.model import EnterpriseResearchReportJobRow, EnterpriseResearchTaskRow
+from deerflow.persistence.enterprise_research.sql import EnterpriseResearchReportJobRepository, EnterpriseResearchTaskRepository
+
+__all__ = [
+    "EnterpriseResearchTaskRow",
+    "EnterpriseResearchReportJobRow",
+    "EnterpriseResearchTaskRepository",
+    "EnterpriseResearchReportJobRepository",
+    "EnterpriseResearchTaskStore",
+    "EnterpriseResearchReportJobStore",
+    "MemoryEnterpriseResearchTaskStore",
+    "MemoryEnterpriseResearchReportJobStore",
+    "make_enterprise_research_task_store",
+    "make_enterprise_research_report_job_store",
+]
+
+
+def make_enterprise_research_task_store(session_factory):
+    if session_factory is None:
+        return MemoryEnterpriseResearchTaskStore()
+    return EnterpriseResearchTaskRepository(session_factory)
+
+
+def make_enterprise_research_report_job_store(session_factory):
+    if session_factory is None:
+        return MemoryEnterpriseResearchReportJobStore()
+    return EnterpriseResearchReportJobRepository(session_factory)

offline-backend-20260512/backend/packages/harness/deerflow/persistence/enterprise_research/base.py

@@ -0,0 +1,75 @@
+"""Persistence contract for the independent enterprise-research workbench."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class EnterpriseResearchTaskStore(ABC):
+    """Owner-scoped task storage; evidence snapshots are DeerFlow-owned."""
+
+    @abstractmethod
+    async def list_tasks(self, user_id: str, *, limit: int = 20) -> list[dict[str, Any]]:
+        raise NotImplementedError
+
+    @abstractmethod
+    async def get_task(self, task_id: str, user_id: str) -> dict[str, Any] | None:
+        raise NotImplementedError
+
+    @abstractmethod
+    async def create_task(self, data: dict[str, Any]) -> dict[str, Any]:
+        raise NotImplementedError
+
+    @abstractmethod
+    async def update_task(self, task_id: str, user_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
+        raise NotImplementedError
+
+    @abstractmethod
+    async def delete_task(self, task_id: str, user_id: str) -> bool:
+        raise NotImplementedError
+
+
+class EnterpriseResearchReportJobStore(ABC):
+    """Owner-scoped, frozen-input jobs for enterprise report writing."""
+
+    @abstractmethod
+    async def create_job(self, data: dict[str, Any]) -> dict[str, Any]:
+        raise NotImplementedError
+
+    @abstractmethod
+    async def get_job(self, job_id: str, user_id: str) -> dict[str, Any] | None:
+        raise NotImplementedError
+
+    @abstractmethod
+    async def get_active_for_task(self, task_id: str, user_id: str) -> dict[str, Any] | None:
+        raise NotImplementedError
+
+    @abstractmethod
+    async def claim_job(self, job_id: str, user_id: str) -> bool:
+        """Atomically move one queued job to running."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def requeue_job(self, job_id: str, user_id: str, *, expected_updated_at: str | None) -> bool:
+        """Atomically requeue the exact running snapshot seen during recovery."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def transition_job(
+        self, job_id: str, user_id: str, *, expected_statuses: set[str], data: dict[str, Any]
+    ) -> dict[str, Any] | None:
+        """Apply a terminal/state transition only from one of the expected states."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def update_job(self, job_id: str, user_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
+        raise NotImplementedError
+
+    @abstractmethod
+    async def list_jobs(self, task_id: str, user_id: str, *, limit: int = 10) -> list[dict[str, Any]]:
+        raise NotImplementedError
+
+    @abstractmethod
+    async def list_recoverable(self, *, limit: int = 100) -> list[dict[str, Any]]:
+        raise NotImplementedError

offline-backend-20260512/backend/packages/harness/deerflow/persistence/enterprise_research/memory.py

@@ -0,0 +1,172 @@
+"""In-memory fallback for enterprise-research tasks."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from deerflow.persistence.enterprise_research.base import EnterpriseResearchReportJobStore, EnterpriseResearchTaskStore
+
+
+def _now() -> str:
+    return datetime.now(UTC).isoformat()
+
+
+class MemoryEnterpriseResearchTaskStore(EnterpriseResearchTaskStore):
+    def __init__(self) -> None:
+        self._rows: dict[str, dict[str, Any]] = {}
+
+    async def list_tasks(self, user_id: str, *, limit: int = 20) -> list[dict[str, Any]]:
+        rows = [dict(row) for row in self._rows.values() if row["user_id"] == user_id]
+        return sorted(rows, key=lambda row: row["updated_at"], reverse=True)[:limit]
+
+    async def get_task(self, task_id: str, user_id: str) -> dict[str, Any] | None:
+        row = self._rows.get(task_id)
+        return dict(row) if row is not None and row["user_id"] == user_id else None
+
+    async def create_task(self, data: dict[str, Any]) -> dict[str, Any]:
+        now = _now()
+        row = {
+            "id": str(data["id"]),
+            "user_id": str(data["user_id"]),
+            "subject": str(data["subject"]),
+            "focus": str(data.get("focus") or ""),
+            "template_id": str(data["template_id"]),
+            "knowledge_base_ids": list(data.get("knowledge_base_ids") or []),
+            "queries": list(data.get("queries") or []),
+            "selected_source_ids": list(data.get("selected_source_ids") or []),
+            "sources": list(data.get("sources") or []),
+            "plan_markdown": data.get("plan_markdown"),
+            "report_markdown": data.get("report_markdown"),
+            "report_summary": data.get("report_summary"),
+            "report_model_name": data.get("report_model_name"),
+            "active_report_job_id": data.get("active_report_job_id"),
+            "status": str(data.get("status") or "draft"),
+            "warning": data.get("warning"),
+            "created_at": now,
+            "updated_at": now,
+        }
+        self._rows[row["id"]] = row
+        return dict(row)
+
+    async def update_task(self, task_id: str, user_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
+        row = self._rows.get(task_id)
+        if row is None or row["user_id"] != user_id:
+            return None
+        for key in (
+            "status", "warning", "sources", "selected_source_ids", "plan_markdown",
+            "report_markdown", "report_summary", "report_model_name", "active_report_job_id",
+        ):
+            if key in data:
+                row[key] = data[key]
+        row["updated_at"] = _now()
+        return dict(row)
+
+    async def delete_task(self, task_id: str, user_id: str) -> bool:
+        row = self._rows.get(task_id)
+        if row is None or row["user_id"] != user_id:
+            return False
+        del self._rows[task_id]
+        return True
+
+
+class MemoryEnterpriseResearchReportJobStore(EnterpriseResearchReportJobStore):
+    def __init__(self) -> None:
+        self._rows: dict[str, dict[str, Any]] = {}
+
+    async def create_job(self, data: dict[str, Any]) -> dict[str, Any]:
+        now = _now()
+        status = str(data.get("status") or "queued")
+        active_dedupe_key = (
+            f"{data['user_id']}:{data['task_id']}" if status in {"queued", "running"} else None
+        )
+        if active_dedupe_key:
+            existing = next(
+                (row for row in self._rows.values() if row.get("active_dedupe_key") == active_dedupe_key),
+                None,
+            )
+            if existing is not None:
+                return dict(existing)
+        row = {
+            "id": str(data["id"]), "task_id": str(data["task_id"]), "user_id": str(data["user_id"]),
+            "status": status, "phase": str(data.get("phase") or "queued"),
+            "model_name": data.get("model_name"), "action": str(data.get("action") or "initial"),
+            "active_dedupe_key": active_dedupe_key,
+            "instruction": data.get("instruction"), "style": data.get("style"),
+            "target_length": data.get("target_length"), "plan_snapshot": str(data["plan_snapshot"]),
+            "sources": list(data.get("sources") or []), "report_markdown": str(data.get("report_markdown") or ""),
+            "original_report": data.get("original_report"),
+            "report_summary": data.get("report_summary"), "error_message": data.get("error_message"),
+            "created_at": now, "updated_at": now,
+        }
+        self._rows[row["id"]] = row
+        return dict(row)
+
+    async def get_job(self, job_id: str, user_id: str) -> dict[str, Any] | None:
+        row = self._rows.get(job_id)
+        return dict(row) if row is not None and row["user_id"] == user_id else None
+
+    async def get_active_for_task(self, task_id: str, user_id: str) -> dict[str, Any] | None:
+        rows = [row for row in self._rows.values() if row["task_id"] == task_id and row["user_id"] == user_id and row["status"] in {"queued", "running"}]
+        return dict(max(rows, key=lambda row: row["updated_at"])) if rows else None
+
+    async def claim_job(self, job_id: str, user_id: str) -> bool:
+        row = self._rows.get(job_id)
+        if row is None or row["user_id"] != user_id or row["status"] != "queued":
+            return False
+        row["status"] = "running"
+        row["updated_at"] = _now()
+        return True
+
+    async def requeue_job(self, job_id: str, user_id: str, *, expected_updated_at: str | None) -> bool:
+        row = self._rows.get(job_id)
+        if (
+            row is None
+            or row["user_id"] != user_id
+            or row["status"] != "running"
+            or row.get("updated_at") != expected_updated_at
+        ):
+            return False
+        row.update({"status": "queued", "phase": "queued", "report_markdown": "", "error_message": None})
+        row["updated_at"] = _now()
+        return True
+
+    async def transition_job(
+        self, job_id: str, user_id: str, *, expected_statuses: set[str], data: dict[str, Any]
+    ) -> dict[str, Any] | None:
+        row = self._rows.get(job_id)
+        if row is None or row["user_id"] != user_id or row["status"] not in expected_statuses:
+            return None
+        for key in (
+            "status", "phase", "report_markdown", "report_summary", "error_message", "model_name",
+            "action", "instruction", "style", "target_length", "original_report",
+        ):
+            if key in data:
+                row[key] = data[key]
+        if data.get("status") in {"completed", "failed", "cancelled"}:
+            row["active_dedupe_key"] = None
+        row["updated_at"] = _now()
+        return dict(row)
+
+    async def update_job(self, job_id: str, user_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
+        row = self._rows.get(job_id)
+        if row is None or row["user_id"] != user_id:
+            return None
+        for key in (
+            "status", "phase", "report_markdown", "report_summary", "error_message", "model_name",
+            "action", "instruction", "style", "target_length", "original_report",
+        ):
+            if key in data:
+                row[key] = data[key]
+        if data.get("status") in {"completed", "failed", "cancelled"}:
+            row["active_dedupe_key"] = None
+        row["updated_at"] = _now()
+        return dict(row)
+
+    async def list_jobs(self, task_id: str, user_id: str, *, limit: int = 10) -> list[dict[str, Any]]:
+        rows = [dict(row) for row in self._rows.values() if row["task_id"] == task_id and row["user_id"] == user_id]
+        return sorted(rows, key=lambda row: row["updated_at"], reverse=True)[:limit]
+
+    async def list_recoverable(self, *, limit: int = 100) -> list[dict[str, Any]]:
+        rows = [dict(row) for row in self._rows.values() if row["status"] in {"queued", "running"}]
+        return sorted(rows, key=lambda row: row["updated_at"])[:limit]

offline-backend-20260512/backend/packages/harness/deerflow/persistence/enterprise_research/model.py

@@ -0,0 +1,77 @@
+"""ORM rows for the independent enterprise-research workbench."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import Index, String, Text, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from deerflow.persistence.base import Base
+from deerflow.persistence.types import BeijingDateTime, PortableJSON, PortableLongText
+
+
+class EnterpriseResearchTaskRow(Base):
+    """A user-owned research task plus its bounded WeKnora evidence snapshot."""
+
+    __tablename__ = "enterprise_research_tasks"
+    __table_args__ = (Index("ix_enterprise_research_tasks_owner_updated", "user_id", "updated_at"),)
+
+    id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    user_id: Mapped[str] = mapped_column(String(128), nullable=False, index=True)
+    subject: Mapped[str] = mapped_column(String(256), nullable=False)
+    focus: Mapped[str] = mapped_column(Text, nullable=False, default="", server_default="")
+    template_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    knowledge_base_ids: Mapped[list[str]] = mapped_column(PortableJSON(), nullable=False, default=list)
+    queries: Mapped[list[str]] = mapped_column(PortableJSON(), nullable=False, default=list)
+    # Nullable keeps upgrades safe: the generic schema reconciler can add this
+    # field to a phase-one table without an unsafe MySQL TEXT default.
+    selected_source_ids: Mapped[list[str] | None] = mapped_column(PortableJSON(), nullable=True, default=list)
+    # A WeKnora result can contain a long document chunk. MySQL TEXT is only
+    # 64 KiB, so evidence uses LONGTEXT through PortableLongText.
+    sources_json: Mapped[str] = mapped_column(PortableLongText(), nullable=False, default="[]", server_default="[]")
+    plan_markdown: Mapped[str | None] = mapped_column(PortableLongText(), nullable=True)
+    report_markdown: Mapped[str | None] = mapped_column(PortableLongText(), nullable=True)
+    report_summary: Mapped[str | None] = mapped_column(PortableLongText(), nullable=True)
+    report_model_name: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    active_report_job_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    status: Mapped[str] = mapped_column(String(32), nullable=False, default="draft", server_default="draft")
+    warning: Mapped[str | None] = mapped_column(PortableLongText(), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
+    updated_at: Mapped[datetime] = mapped_column(
+        BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC)
+    )
+
+
+class EnterpriseResearchReportJobRow(Base):
+    """A report job whose plan/evidence are frozen when the user starts it."""
+
+    __tablename__ = "enterprise_research_report_jobs"
+    __table_args__ = (
+        Index("ix_enterprise_research_report_jobs_task_owner_updated", "task_id", "user_id", "updated_at"),
+        UniqueConstraint("active_dedupe_key", name="uq_enterprise_research_report_jobs_active"),
+    )
+
+    id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    task_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    user_id: Mapped[str] = mapped_column(String(128), nullable=False, index=True)
+    status: Mapped[str] = mapped_column(String(32), nullable=False, default="queued", server_default="queued")
+    phase: Mapped[str] = mapped_column(String(32), nullable=False, default="queued", server_default="queued")
+    model_name: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    # Present only for queued/running rows. NULL terminal rows keep unlimited
+    # history while this key rejects cross-process duplicate starts.
+    active_dedupe_key: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    action: Mapped[str | None] = mapped_column(String(32), nullable=True, default="initial")
+    instruction: Mapped[str | None] = mapped_column(PortableLongText(), nullable=True)
+    style: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    target_length: Mapped[int | None] = mapped_column(nullable=True)
+    plan_snapshot: Mapped[str] = mapped_column(PortableLongText(), nullable=False)
+    sources_json: Mapped[str] = mapped_column(PortableLongText(), nullable=False)
+    original_report: Mapped[str | None] = mapped_column(PortableLongText(), nullable=True)
+    report_markdown: Mapped[str] = mapped_column(PortableLongText(), nullable=False, default="", server_default="")
+    report_summary: Mapped[str | None] = mapped_column(PortableLongText(), nullable=True)
+    error_message: Mapped[str | None] = mapped_column(PortableLongText(), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
+    updated_at: Mapped[datetime] = mapped_column(
+        BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC)
+    )

offline-backend-20260512/backend/packages/harness/deerflow/persistence/enterprise_research/sql.py

@@ -0,0 +1,320 @@
+"""SQLAlchemy store for enterprise-research tasks."""
+
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import delete, select, update
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from deerflow.persistence.enterprise_research.base import EnterpriseResearchReportJobStore, EnterpriseResearchTaskStore
+from deerflow.persistence.enterprise_research.model import EnterpriseResearchReportJobRow, EnterpriseResearchTaskRow
+
+
+def _to_dict(row: EnterpriseResearchTaskRow) -> dict[str, Any]:
+    try:
+        sources = json.loads(row.sources_json or "[]")
+    except (TypeError, ValueError):
+        sources = []
+    data: dict[str, Any] = {
+        "id": row.id,
+        "user_id": row.user_id,
+        "subject": row.subject,
+        "focus": row.focus,
+        "template_id": row.template_id,
+        "knowledge_base_ids": list(row.knowledge_base_ids or []),
+        "queries": list(row.queries or []),
+        "selected_source_ids": list(row.selected_source_ids or []),
+        "sources": sources if isinstance(sources, list) else [],
+        "plan_markdown": row.plan_markdown,
+        "report_markdown": row.report_markdown,
+        "report_summary": row.report_summary,
+        "report_model_name": row.report_model_name,
+        "active_report_job_id": row.active_report_job_id,
+        "status": row.status,
+        "warning": row.warning,
+        "created_at": row.created_at.isoformat() if row.created_at else None,
+        "updated_at": row.updated_at.isoformat() if row.updated_at else None,
+    }
+    return data
+
+
+class EnterpriseResearchTaskRepository(EnterpriseResearchTaskStore):
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
+        self._sf = session_factory
+
+    async def list_tasks(self, user_id: str, *, limit: int = 20) -> list[dict[str, Any]]:
+        stmt = (
+            select(EnterpriseResearchTaskRow)
+            .where(EnterpriseResearchTaskRow.user_id == user_id)
+            .order_by(EnterpriseResearchTaskRow.updated_at.desc())
+            .limit(limit)
+        )
+        async with self._sf() as session:
+            result = await session.execute(stmt)
+            return [_to_dict(row) for row in result.scalars()]
+
+    async def get_task(self, task_id: str, user_id: str) -> dict[str, Any] | None:
+        stmt = select(EnterpriseResearchTaskRow).where(
+            EnterpriseResearchTaskRow.id == task_id,
+            EnterpriseResearchTaskRow.user_id == user_id,
+        )
+        async with self._sf() as session:
+            row = (await session.execute(stmt)).scalar_one_or_none()
+            return _to_dict(row) if row is not None else None
+
+    async def create_task(self, data: dict[str, Any]) -> dict[str, Any]:
+        now = datetime.now(UTC)
+        row = EnterpriseResearchTaskRow(
+            id=str(data["id"]),
+            user_id=str(data["user_id"]),
+            subject=str(data["subject"]),
+            focus=str(data.get("focus") or ""),
+            template_id=str(data["template_id"]),
+            knowledge_base_ids=list(data.get("knowledge_base_ids") or []),
+            queries=list(data.get("queries") or []),
+            selected_source_ids=list(data.get("selected_source_ids") or []),
+            sources_json=json.dumps(list(data.get("sources") or []), ensure_ascii=False),
+            plan_markdown=data.get("plan_markdown"),
+            report_markdown=data.get("report_markdown"),
+            report_summary=data.get("report_summary"),
+            report_model_name=data.get("report_model_name"),
+            active_report_job_id=data.get("active_report_job_id"),
+            status=str(data.get("status") or "draft"),
+            warning=data.get("warning"),
+            created_at=now,
+            updated_at=now,
+        )
+        async with self._sf() as session:
+            session.add(row)
+            # Do not refresh after commit: a MySQL read/write split can route it
+            # to a lagging replica. The flushed in-memory row is authoritative.
+            await session.flush()
+            payload = _to_dict(row)
+            await session.commit()
+            return payload
+
+    async def update_task(self, task_id: str, user_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
+        stmt = select(EnterpriseResearchTaskRow).where(
+            EnterpriseResearchTaskRow.id == task_id,
+            EnterpriseResearchTaskRow.user_id == user_id,
+        )
+        async with self._sf() as session:
+            row = (await session.execute(stmt)).scalar_one_or_none()
+            if row is None:
+                return None
+            if "status" in data:
+                row.status = str(data["status"])
+            if "warning" in data:
+                row.warning = data["warning"]
+            if "sources" in data:
+                row.sources_json = json.dumps(list(data["sources"] or []), ensure_ascii=False)
+            if "selected_source_ids" in data:
+                row.selected_source_ids = list(data["selected_source_ids"] or [])
+            if "plan_markdown" in data:
+                row.plan_markdown = data["plan_markdown"]
+            for key in ("report_markdown", "report_summary", "report_model_name", "active_report_job_id"):
+                if key in data:
+                    setattr(row, key, data[key])
+            row.updated_at = datetime.now(UTC)
+            await session.flush()
+            payload = _to_dict(row)
+            await session.commit()
+            return payload
+
+    async def delete_task(self, task_id: str, user_id: str) -> bool:
+        async with self._sf() as session:
+            result = await session.execute(
+                delete(EnterpriseResearchTaskRow).where(
+                    EnterpriseResearchTaskRow.id == task_id,
+                    EnterpriseResearchTaskRow.user_id == user_id,
+                )
+            )
+            await session.commit()
+            return bool(result.rowcount)
+
+
+def _job_to_dict(row: EnterpriseResearchReportJobRow) -> dict[str, Any]:
+    try:
+        sources = json.loads(row.sources_json or "[]")
+    except (TypeError, ValueError):
+        sources = []
+    return {
+        "id": row.id, "task_id": row.task_id, "user_id": row.user_id, "status": row.status,
+        "phase": row.phase, "model_name": row.model_name, "action": row.action or "initial",
+        "instruction": row.instruction, "style": row.style, "target_length": row.target_length,
+        "plan_snapshot": row.plan_snapshot,
+        "sources": sources if isinstance(sources, list) else [], "report_markdown": row.report_markdown,
+        "original_report": row.original_report,
+        "report_summary": row.report_summary, "error_message": row.error_message,
+        "created_at": row.created_at.isoformat() if row.created_at else None,
+        "updated_at": row.updated_at.isoformat() if row.updated_at else None,
+    }
+
+
+class EnterpriseResearchReportJobRepository(EnterpriseResearchReportJobStore):
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
+        self._sf = session_factory
+
+    async def create_job(self, data: dict[str, Any]) -> dict[str, Any]:
+        now = datetime.now(UTC)
+        status = str(data.get("status") or "queued")
+        active_dedupe_key = (
+            f"{data['user_id']}:{data['task_id']}" if status in {"queued", "running"} else None
+        )
+        row = EnterpriseResearchReportJobRow(
+            id=str(data["id"]), task_id=str(data["task_id"]), user_id=str(data["user_id"]),
+            status=status, phase=str(data.get("phase") or "queued"),
+            model_name=data.get("model_name"), action=str(data.get("action") or "initial"),
+            active_dedupe_key=active_dedupe_key,
+            instruction=data.get("instruction"), style=data.get("style"), target_length=data.get("target_length"),
+            plan_snapshot=str(data["plan_snapshot"]),
+            sources_json=json.dumps(list(data.get("sources") or []), ensure_ascii=False),
+            original_report=data.get("original_report"),
+            report_markdown=str(data.get("report_markdown") or ""), report_summary=data.get("report_summary"),
+            error_message=data.get("error_message"), created_at=now, updated_at=now,
+        )
+        async with self._sf() as session:
+            session.add(row)
+            try:
+                await session.flush()
+                payload = _job_to_dict(row)
+                await session.commit()
+                return payload
+            except IntegrityError:
+                await session.rollback()
+                if not active_dedupe_key:
+                    raise
+                existing = (await session.execute(
+                    select(EnterpriseResearchReportJobRow)
+                    .where(EnterpriseResearchReportJobRow.active_dedupe_key == active_dedupe_key)
+                    .limit(1)
+                )).scalar_one_or_none()
+                if existing is None:
+                    raise
+                return _job_to_dict(existing)
+
+    async def get_job(self, job_id: str, user_id: str) -> dict[str, Any] | None:
+        async with self._sf() as session:
+            row = (await session.execute(select(EnterpriseResearchReportJobRow).where(
+                EnterpriseResearchReportJobRow.id == job_id, EnterpriseResearchReportJobRow.user_id == user_id,
+            ))).scalar_one_or_none()
+            return _job_to_dict(row) if row is not None else None
+
+    async def get_active_for_task(self, task_id: str, user_id: str) -> dict[str, Any] | None:
+        async with self._sf() as session:
+            row = (await session.execute(select(EnterpriseResearchReportJobRow).where(
+                EnterpriseResearchReportJobRow.task_id == task_id,
+                EnterpriseResearchReportJobRow.user_id == user_id,
+                EnterpriseResearchReportJobRow.status.in_(["queued", "running"]),
+            ).order_by(EnterpriseResearchReportJobRow.updated_at.desc()).limit(1))).scalar_one_or_none()
+            return _job_to_dict(row) if row is not None else None
+
+    async def claim_job(self, job_id: str, user_id: str) -> bool:
+        async with self._sf() as session:
+            result = await session.execute(
+                update(EnterpriseResearchReportJobRow)
+                .where(
+                    EnterpriseResearchReportJobRow.id == job_id,
+                    EnterpriseResearchReportJobRow.user_id == user_id,
+                    EnterpriseResearchReportJobRow.status == "queued",
+                )
+                .values(status="running", updated_at=datetime.now(UTC))
+            )
+            await session.commit()
+            return bool(result.rowcount)
+
+    async def requeue_job(self, job_id: str, user_id: str, *, expected_updated_at: str | None) -> bool:
+        if not expected_updated_at:
+            return False
+        try:
+            expected = datetime.fromisoformat(expected_updated_at)
+        except ValueError:
+            return False
+        async with self._sf() as session:
+            result = await session.execute(
+                update(EnterpriseResearchReportJobRow)
+                .where(
+                    EnterpriseResearchReportJobRow.id == job_id,
+                    EnterpriseResearchReportJobRow.user_id == user_id,
+                    EnterpriseResearchReportJobRow.status == "running",
+                    EnterpriseResearchReportJobRow.updated_at == expected,
+                )
+                .values(
+                    status="queued",
+                    phase="queued",
+                    report_markdown="",
+                    error_message=None,
+                    updated_at=datetime.now(UTC),
+                )
+            )
+            await session.commit()
+            return bool(result.rowcount)
+
+    async def transition_job(
+        self, job_id: str, user_id: str, *, expected_statuses: set[str], data: dict[str, Any]
+    ) -> dict[str, Any] | None:
+        allowed = {
+            "status", "phase", "report_markdown", "report_summary", "error_message", "model_name",
+            "action", "instruction", "style", "target_length", "original_report",
+        }
+        values = {key: value for key, value in data.items() if key in allowed}
+        if data.get("status") in {"completed", "failed", "cancelled"}:
+            values["active_dedupe_key"] = None
+        values["updated_at"] = datetime.now(UTC)
+        async with self._sf() as session:
+            result = await session.execute(
+                update(EnterpriseResearchReportJobRow)
+                .where(
+                    EnterpriseResearchReportJobRow.id == job_id,
+                    EnterpriseResearchReportJobRow.user_id == user_id,
+                    EnterpriseResearchReportJobRow.status.in_(expected_statuses),
+                )
+                .values(**values)
+            )
+            if not result.rowcount:
+                await session.rollback()
+                return None
+            await session.commit()
+        return await self.get_job(job_id, user_id)
+
+    async def update_job(self, job_id: str, user_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
+        async with self._sf() as session:
+            row = (await session.execute(select(EnterpriseResearchReportJobRow).where(
+                EnterpriseResearchReportJobRow.id == job_id, EnterpriseResearchReportJobRow.user_id == user_id,
+            ))).scalar_one_or_none()
+            if row is None:
+                return None
+            for key in (
+                "status", "phase", "report_markdown", "report_summary", "error_message", "model_name",
+                "action", "instruction", "style", "target_length", "original_report",
+            ):
+                if key in data:
+                    setattr(row, key, data[key])
+            if data.get("status") in {"completed", "failed", "cancelled"}:
+                row.active_dedupe_key = None
+            row.updated_at = datetime.now(UTC)
+            await session.flush()
+            payload = _job_to_dict(row)
+            await session.commit()
+            return payload
+
+    async def list_jobs(self, task_id: str, user_id: str, *, limit: int = 10) -> list[dict[str, Any]]:
+        async with self._sf() as session:
+            rows = (await session.execute(select(EnterpriseResearchReportJobRow).where(
+                EnterpriseResearchReportJobRow.task_id == task_id, EnterpriseResearchReportJobRow.user_id == user_id,
+            ).order_by(EnterpriseResearchReportJobRow.updated_at.desc()).limit(limit))).scalars()
+            return [_job_to_dict(row) for row in rows]
+
+    async def list_recoverable(self, *, limit: int = 100) -> list[dict[str, Any]]:
+        async with self._sf() as session:
+            rows = (await session.execute(
+                select(EnterpriseResearchReportJobRow)
+                .where(EnterpriseResearchReportJobRow.status.in_(["queued", "running"]))
+                .order_by(EnterpriseResearchReportJobRow.updated_at.asc())
+                .limit(limit)
+            )).scalars()
+            return [_job_to_dict(row) for row in rows]

offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py

@@ -46,6 +46,7 @@ from deerflow.persistence.deep_research_sessions.model import DeepResearchSessio
 from deerflow.persistence.deep_research_sources.model import DeepResearchSourceRow
 from deerflow.persistence.document_rewrite_versions.model import DocumentRewriteVersionRow
 from deerflow.persistence.embed_sessions.model import EmbedSessionRow
+from deerflow.persistence.enterprise_research.model import EnterpriseResearchReportJobRow, EnterpriseResearchTaskRow
 from deerflow.persistence.feedback.model import FeedbackRow
 from deerflow.persistence.intent_turns.model import IntentTurnRow
 from deerflow.persistence.light_apps.model import LightAppRow
@@ -162,6 +163,8 @@ __all__ = [
     "TaskCopTaskRow",
     "TaskReportRow",
     "EmbedSessionRow",
+    "EnterpriseResearchTaskRow",
+    "EnterpriseResearchReportJobRow",
     "IntentTurnRow",
     "ThreadMetaRow",
     "ThreadShareRow",

offline-backend-20260512/backend/tests/test_enterprise_research_tasks.py

@@ -0,0 +1,396 @@
+"""Regression tests for owner-scoped enterprise-research task persistence."""
+
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+import pytest_asyncio
+from fastapi import FastAPI
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from deerflow.persistence.base import Base
+from deerflow.persistence.enterprise_research import (
+    EnterpriseResearchReportJobRepository,
+    EnterpriseResearchTaskRepository,
+    MemoryEnterpriseResearchReportJobStore,
+    MemoryEnterpriseResearchTaskStore,
+    make_enterprise_research_task_store,
+)
+
+
+@pytest_asyncio.fixture
+async def store(tmp_path):
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'enterprise-research.db'}")
+    async with engine.begin() as conn:
+        import deerflow.persistence.enterprise_research.model  # noqa: F401
+
+        await conn.run_sync(Base.metadata.create_all)
+    return EnterpriseResearchTaskRepository(async_sessionmaker(engine, expire_on_commit=False))
+
+
+async def _create(store, *, task_id: str, user_id: str = "u1"):
+    return await store.create_task(
+        {
+            "id": task_id,
+            "user_id": user_id,
+            "subject": "新能源汽车供应链风险",
+            "focus": "关键原材料",
+            "template_id": "topic",
+            "knowledge_base_ids": ["kb-1"],
+            "queries": ["新能源汽车供应链风险", "关键原材料"],
+            "status": "draft",
+        }
+    )
+
+
+@pytest.mark.asyncio
+async def test_task_evidence_roundtrip_and_owner_isolation(store: EnterpriseResearchTaskRepository):
+    created = await _create(store, task_id="ert-1")
+    assert created["sources"] == []
+    assert created["status"] == "draft"
+
+    updated = await store.update_task(
+        "ert-1",
+        "u1",
+        {
+            "status": "ready",
+            "selected_source_ids": ["chunk-1"],
+            "plan_markdown": "# 供应链风险研究计划",
+            "sources": [
+                {
+                    "chunk_id": "chunk-1",
+                    "title": "供应链报告",
+                    "content": "证据内容",
+                    "knowledge_base_id": "kb-1",
+                }
+            ],
+        },
+    )
+    assert updated is not None
+    assert updated["status"] == "ready"
+    assert updated["sources"][0]["chunk_id"] == "chunk-1"
+    assert updated["selected_source_ids"] == ["chunk-1"]
+    assert updated["plan_markdown"] == "# 供应链风险研究计划"
+    assert await store.get_task("ert-1", "u2") is None
+    assert await store.update_task("ert-1", "u2", {"status": "failed"}) is None
+
+
+@pytest.mark.asyncio
+async def test_list_is_owner_scoped_and_delete_requires_owner(store: EnterpriseResearchTaskRepository):
+    await _create(store, task_id="ert-u1", user_id="u1")
+    await _create(store, task_id="ert-u2", user_id="u2")
+    assert [task["id"] for task in await store.list_tasks("u1")] == ["ert-u1"]
+    assert await store.delete_task("ert-u1", "u2") is False
+    assert await store.delete_task("ert-u1", "u1") is True
+    assert await store.get_task("ert-u1", "u1") is None
+
+
+@pytest.mark.asyncio
+async def test_memory_store_matches_sql_contract():
+    store = make_enterprise_research_task_store(None)
+    assert isinstance(store, MemoryEnterpriseResearchTaskStore)
+    await _create(store, task_id="memory-1")
+    assert (await store.get_task("memory-1", "u1"))["subject"] == "新能源汽车供应链风险"
+    assert await store.delete_task("memory-1", "u1") is True
+
+
+@pytest.mark.asyncio
+async def test_report_job_freezes_sources_and_is_owner_scoped(tmp_path):
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'enterprise-report-job.db'}")
+    async with engine.begin() as conn:
+        import deerflow.persistence.enterprise_research.model  # noqa: F401
+
+        await conn.run_sync(Base.metadata.create_all)
+    jobs = EnterpriseResearchReportJobRepository(async_sessionmaker(engine, expire_on_commit=False))
+    created = await jobs.create_job(
+        {
+            "id": "erj-1", "task_id": "ert-1", "user_id": "u1", "plan_snapshot": "# 计划",
+            "sources": [{"chunk_id": "source-1", "content": "冻结证据"}],
+            "action": "rewrite", "instruction": "优化结构", "original_report": "# 原报告",
+        }
+    )
+    assert created["status"] == "queued"
+    assert created["sources"][0]["content"] == "冻结证据"
+    assert created["action"] == "rewrite"
+    assert created["original_report"] == "# 原报告"
+    deduplicated = await jobs.create_job(
+        {
+            "id": "erj-duplicate", "task_id": "ert-1", "user_id": "u1", "plan_snapshot": "# 其他计划",
+            "sources": [], "action": "initial",
+        }
+    )
+    assert deduplicated["id"] == "erj-1"
+    assert await jobs.get_job("erj-1", "u2") is None
+    assert (await jobs.get_active_for_task("ert-1", "u1"))["id"] == "erj-1"
+    assert await jobs.claim_job("erj-1", "u1") is True
+    assert await jobs.claim_job("erj-1", "u1") is False
+    running = (await jobs.list_recoverable())[0]
+    assert running["status"] == "running"
+    assert await jobs.requeue_job("erj-1", "u1", expected_updated_at="stale-snapshot") is False
+    assert await jobs.requeue_job("erj-1", "u1", expected_updated_at=running["updated_at"]) is True
+    assert await jobs.requeue_job("erj-1", "u1", expected_updated_at=running["updated_at"]) is False
+    assert await jobs.claim_job("erj-1", "u1") is True
+    completed = await jobs.transition_job(
+        "erj-1", "u1", expected_statuses={"running"},
+        data={"status": "completed", "phase": "completed", "report_markdown": "# 报告"},
+    )
+    assert completed is not None and completed["report_markdown"] == "# 报告"
+    assert await jobs.transition_job(
+        "erj-1", "u1", expected_statuses={"running"}, data={"status": "cancelled"}
+    ) is None
+    assert await jobs.get_active_for_task("ert-1", "u1") is None
+    next_job = await jobs.create_job(
+        {"id": "erj-2", "task_id": "ert-1", "user_id": "u1", "plan_snapshot": "# 新计划", "sources": []}
+    )
+    assert next_job["id"] == "erj-2"
+
+
+@pytest.mark.asyncio
+async def test_memory_report_job_store_has_matching_owner_contract():
+    jobs = MemoryEnterpriseResearchReportJobStore()
+    await jobs.create_job({"id": "memory-job", "task_id": "task", "user_id": "u1", "plan_snapshot": "x"})
+    assert await jobs.get_job("memory-job", "u2") is None
+    assert (await jobs.list_jobs("task", "u1"))[0]["id"] == "memory-job"
+    assert await jobs.claim_job("memory-job", "u1") is True
+    assert await jobs.claim_job("memory-job", "u1") is False
+    running = await jobs.get_job("memory-job", "u1")
+    assert running is not None
+    assert await jobs.requeue_job("memory-job", "u1", expected_updated_at=running["updated_at"]) is True
+    assert await jobs.claim_job("memory-job", "u1") is True
+    completed = await jobs.transition_job(
+        "memory-job", "u1", expected_statuses={"running"}, data={"status": "completed", "phase": "completed"}
+    )
+    assert completed is not None and completed["status"] == "completed"
+
+
+def test_evidence_snapshot_deduplicates_and_never_exposes_unmapped_remote_bases():
+    from app.gateway.routers.enterprise_research import _dedupe_evidence
+
+    mappings = {"remote-kb-1": {"id": "local-kb-1", "name": "用户可见知识库", "weknora_id": "remote-kb-1"}}
+    evidence = _dedupe_evidence(
+        [
+            {"chunk_id": "chunk-1", "content": "第一段", "knowledge_base_id": "remote-kb-1", "title": "文档 A"},
+            {"chunk_id": "chunk-1", "content": "重复段", "knowledge_base_id": "remote-kb-1", "title": "文档 A"},
+            {"chunk_id": "private-chunk", "content": "不应显示", "knowledge_base_id": "other-user-kb", "title": "私有文档"},
+        ],
+        mappings,
+    )
+    assert len(evidence) == 1
+    assert evidence[0]["knowledge_base_id"] == "local-kb-1"
+    assert "other-user-kb" not in str(evidence)
+
+
+def test_report_executor_uses_distinct_initial_rewrite_and_followup_prompts():
+    from app.gateway.enterprise_research_report_executor import EnterpriseResearchReportExecutor
+
+    base = {
+        "plan_snapshot": "# 计划",
+        "sources": [{"chunk_id": "source-1", "title": "内部资料", "content": "证据正文"}],
+        "original_report": "# 原报告\n原始正文",
+        "instruction": "压缩重复段落",
+        "target_length": 3_000,
+        "style": "analytical",
+    }
+    initial = EnterpriseResearchReportExecutor._messages({**base, "action": "initial"})
+    rewrite = EnterpriseResearchReportExecutor._messages({**base, "action": "rewrite"})
+    followup = EnterpriseResearchReportExecutor._messages({**base, "action": "followup"})
+
+    assert "完整 Markdown 报告" in initial[0]["content"]
+    assert "原报告" not in initial[1]["content"]
+    assert "全文改写" in rewrite[1]["content"] and "# 原报告" in rewrite[1]["content"]
+    assert "不要重写整篇报告" in followup[0]["content"]
+    assert "压缩重复段落" in followup[1]["content"]
+    summary = EnterpriseResearchReportExecutor._summary(
+        "# 新报告\n\n新正文", 2, "rewrite", original_report="# 旧报告\n\n旧正文", instruction="调整结构"
+    )
+    assert "原文" in summary and "标题节点 1→1" in summary and "调整结构" in summary
+
+
+@pytest.mark.asyncio
+async def test_executor_recovery_requeues_exact_running_snapshot(monkeypatch):
+    from app.gateway.enterprise_research_report_executor import EnterpriseResearchReportExecutor
+
+    tasks = MemoryEnterpriseResearchTaskStore()
+    jobs = MemoryEnterpriseResearchReportJobStore()
+    await jobs.create_job({"id": "recover-job", "task_id": "task", "user_id": "u1", "plan_snapshot": "# 计划"})
+    assert await jobs.claim_job("recover-job", "u1") is True
+    executor = EnterpriseResearchReportExecutor(tasks, jobs)
+    started: list[tuple[str, str]] = []
+
+    async def remember(job_id: str, user_id: str) -> None:
+        started.append((job_id, user_id))
+
+    monkeypatch.setattr(executor, "ensure_started", remember)
+    await executor.recover_pending()
+
+    recovered = await jobs.get_job("recover-job", "u1")
+    assert recovered is not None and recovered["status"] == "queued"
+    assert started == [("recover-job", "u1")]
+
+
+@pytest.mark.asyncio
+async def test_enterprise_research_http_plan_job_dedupe_and_cancel(monkeypatch):
+    import app.gateway.routers.enterprise_research as enterprise_router
+
+    tasks = MemoryEnterpriseResearchTaskStore()
+    jobs = MemoryEnterpriseResearchReportJobStore()
+
+    class NoopExecutor:
+        def __init__(self) -> None:
+            self.started: list[tuple[str, str]] = []
+
+        async def ensure_started(self, job_id: str, user_id: str) -> None:
+            self.started.append((job_id, user_id))
+
+        async def cancel(self, job_id: str, user_id: str):
+            job = await jobs.get_job(job_id, user_id)
+            if job is None:
+                return None
+            cancelled = await jobs.transition_job(
+                job_id, user_id, expected_statuses={"queued", "running"},
+                data={"status": "cancelled", "phase": "cancelled"},
+            )
+            await tasks.update_task(job["task_id"], user_id, {"active_report_job_id": None})
+            return cancelled
+
+    async def anonymous_actor(_request):
+        return None
+
+    monkeypatch.setattr(enterprise_router, "get_optional_user_from_request", anonymous_actor)
+    app = FastAPI()
+    app.include_router(enterprise_router.router)
+    app.state.enterprise_research_task_store = tasks
+    app.state.enterprise_research_report_job_store = jobs
+    app.state.enterprise_research_report_executor = NoopExecutor()
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
+        created = await client.post(
+            "/api/enterprise-research/tasks",
+            json={"subject": "企业风险", "queries": ["企业风险"], "knowledge_base_ids": []},
+        )
+        assert created.status_code == 201
+        task_id = created.json()["id"]
+        await tasks.update_task(
+            task_id, "default",
+            {
+                "status": "ready", "selected_source_ids": ["source-1"],
+                "sources": [{"chunk_id": "source-1", "title": "内部材料", "content": "已核验事实"}],
+            },
+        )
+        plan = await client.post(f"/api/enterprise-research/tasks/{task_id}/plan/draft")
+        assert plan.status_code == 200 and plan.json()["plan_markdown"]
+        saved = await client.put(
+            f"/api/enterprise-research/tasks/{task_id}/plan",
+            json={"plan_markdown": "# 企业风险报告", "selected_source_ids": ["source-1"]},
+        )
+        assert saved.status_code == 200
+        first = await client.post(
+            f"/api/enterprise-research/tasks/{task_id}/report-jobs",
+            json={"action": "initial", "style": "formal", "target_length": 2_000},
+        )
+        assert first.status_code == 201
+        duplicate = await client.post(
+            f"/api/enterprise-research/tasks/{task_id}/report-jobs",
+            json={"action": "initial"},
+        )
+        assert duplicate.status_code == 201 and duplicate.json()["id"] == first.json()["id"]
+        cancelled = await client.post(
+            f"/api/enterprise-research/tasks/{task_id}/report-jobs/{first.json()['id']}/cancel"
+        )
+        assert cancelled.status_code == 200 and cancelled.json()["status"] == "cancelled"
+
+
+@pytest.mark.asyncio
+async def test_executor_stream_commit_separates_reasoning_and_followup(monkeypatch):
+    import app.gateway.enterprise_research_report_executor as executor_module
+
+    tasks = MemoryEnterpriseResearchTaskStore()
+    jobs = MemoryEnterpriseResearchReportJobStore()
+    await _create(tasks, task_id="stream-task")
+    await tasks.update_task(
+        "stream-task", "u1",
+        {"plan_markdown": "# 研究计划", "report_markdown": "# 旧报告", "report_summary": "旧摘要"},
+    )
+
+    class FakeBackend:
+        def __init__(self, _config) -> None:
+            pass
+
+        async def stream_complete(self, *, messages, on_delta, on_reasoning, **_kwargs):
+            is_followup = "不要重写整篇报告" in messages[0]["content"]
+            text = "追问回答" if is_followup else "# 新报告\n正文 [资料:source-1]"
+            await on_reasoning("模型内部思考")
+            for delta in (text[:4], text[4:]):
+                await on_delta(delta)
+            return SimpleNamespace(text=text, model="fake-model")
+
+    monkeypatch.setattr(executor_module, "DeerFlowCompletionBackend", FakeBackend)
+    executor = executor_module.EnterpriseResearchReportExecutor(tasks, jobs)
+    await jobs.create_job(
+        {
+            "id": "initial-job", "task_id": "stream-task", "user_id": "u1", "plan_snapshot": "# 研究计划",
+            "sources": [{"chunk_id": "source-1", "content": "证据"}], "action": "initial",
+        }
+    )
+    await executor._run("initial-job", "u1")
+    completed = await jobs.get_job("initial-job", "u1")
+    task = await tasks.get_task("stream-task", "u1")
+    assert completed is not None and completed["status"] == "completed"
+    assert completed["report_markdown"] == "# 新报告\n正文 [资料:source-1]"
+    assert "模型内部思考" not in completed["report_markdown"]
+    assert task is not None and task["report_markdown"] == completed["report_markdown"]
+
+    await jobs.create_job(
+        {
+            "id": "followup-job", "task_id": "stream-task", "user_id": "u1", "plan_snapshot": "# 研究计划",
+            "sources": [{"chunk_id": "source-1", "content": "证据"}], "action": "followup",
+            "instruction": "结论是什么", "original_report": completed["report_markdown"],
+        }
+    )
+    await executor._run("followup-job", "u1")
+    followup = await jobs.get_job("followup-job", "u1")
+    unchanged = await tasks.get_task("stream-task", "u1")
+    assert followup is not None and followup["report_markdown"] == "追问回答"
+    assert unchanged is not None and unchanged["report_markdown"] == completed["report_markdown"]
+
+
+@pytest.mark.asyncio
+async def test_cross_process_cancel_cannot_be_overwritten_by_late_model_result(monkeypatch):
+    import app.gateway.enterprise_research_report_executor as executor_module
+
+    tasks = MemoryEnterpriseResearchTaskStore()
+    jobs = MemoryEnterpriseResearchReportJobStore()
+    await _create(tasks, task_id="cancel-task")
+    await tasks.update_task("cancel-task", "u1", {"report_markdown": "# 已保存报告"})
+    await jobs.create_job(
+        {"id": "cancel-job", "task_id": "cancel-task", "user_id": "u1", "plan_snapshot": "# 计划"}
+    )
+    started = asyncio.Event()
+    release = asyncio.Event()
+
+    class SlowBackend:
+        def __init__(self, _config) -> None:
+            pass
+
+        async def stream_complete(self, *, on_delta, **_kwargs):
+            started.set()
+            await release.wait()
+            await on_delta("# 不应提交的迟到结果")
+            return SimpleNamespace(text="# 不应提交的迟到结果", model="fake-model")
+
+    monkeypatch.setattr(executor_module, "DeerFlowCompletionBackend", SlowBackend)
+    writer = executor_module.EnterpriseResearchReportExecutor(tasks, jobs)
+    remote_gateway = executor_module.EnterpriseResearchReportExecutor(tasks, jobs)
+    running = asyncio.create_task(writer._run("cancel-job", "u1"))
+    await started.wait()
+    cancelled = await remote_gateway.cancel("cancel-job", "u1")
+    assert cancelled is not None and cancelled["status"] == "cancelled"
+    release.set()
+    with pytest.raises(asyncio.CancelledError):
+        await running
+    final_job = await jobs.get_job("cancel-job", "u1")
+    final_task = await tasks.get_task("cancel-task", "u1")
+    assert final_job is not None and final_job["status"] == "cancelled"
+    assert final_task is not None and final_task["report_markdown"] == "# 已保存报告"