deerflow-code/offline-backend-20260512/backend/app/gateway/workflow_deep_research_adapter.py

"""Bridge a workflow report node onto the existing Deep Research job system.

The workflow engine intentionally has no dependency on Deep Research storage or
the web app.  This adapter is the only integration seam: it turns an immutable
workflow Evidence Pack into selected Deep Research source rows, creates (or
reattaches to) one durable report-writing job, and projects safe progress back
to the caller.  It never calls the HTTP router and therefore cannot create an
internal request/auth loop.
"""

from __future__ import annotations

import asyncio
import hashlib
import json
import logging
from collections.abc import Awaitable, Callable
from typing import Any

from deerflow.agents.deep_research.config import (
    DEFAULT_MAX_ACTIVE_JOBS_PER_USER,
    DeepResearchConfig,
)
from deerflow.persistence.deep_research_jobs.sql import TERMINAL_STATUSES
from deerflow.workflows.errors import WorkflowError
from deerflow.workflows.runtime.context import CancelToken

logger = logging.getLogger(__name__)

_POLL_SECONDS = 0.25
_EVENT_PAGE_SIZE = 200
_MAX_EVIDENCE_ITEMS = 64
_MAX_EVIDENCE_ITEM_CHARS = 20_000

ProgressCallback = Callable[[dict[str, Any]], Awaitable[None]]


def _stable_id(prefix: str, *parts: str) -> str:
    digest = hashlib.sha256("\x1f".join(parts).encode("utf-8")).hexdigest()
    return f"{prefix}_{digest[:40]}"


def _text(value: Any, *, limit: int) -> str:
    value = str(value or "").strip()
    return value[:limit]


class WorkflowDeepResearchAdapter:
    """App-layer implementation injected into ``WorkflowRuntimeDeps``."""

    def __init__(self, app: Any) -> None:
        self._app = app

    async def run(
        self,
        *,
        run_id: str,
        node_id: str,
        owner_id: str,
        topic: str,
        evidence_pack: dict[str, Any],
        research_config: dict[str, Any],
        report_instruction: str,
        cancel: CancelToken,
        on_event: ProgressCallback,
    ) -> dict[str, Any]:
        stores = self._stores()
        config = self._config(research_config, report_instruction)
        # A retry of the same node reattaches to its existing job, while a
        # meaningful topic/evidence/write-config change receives a new report
        # session instead of accidentally returning stale prose from an old
        # snapshot.
        execution_fingerprint = self._execution_fingerprint(
            topic=topic,
            evidence_pack=evidence_pack,
            config=config,
            instruction=report_instruction,
        )
        session_id = _stable_id("drs_wf", run_id, node_id, owner_id, execution_fingerprint)
        source_count = await self._ensure_session_and_sources(
            stores=stores,
            session_id=session_id,
            run_id=run_id,
            node_id=node_id,
            owner_id=owner_id,
            topic=topic,
            evidence_pack=evidence_pack,
            config=config,
        )

        session = await stores["sessions"].get(session_id, user_id=owner_id)
        if session is not None and session.get("status") == "completed":
            report = str(session.get("report_markdown") or "").strip()
            if report:
                return {
                    "session_id": session_id,
                    "job_id": session.get("active_job_id") or (session.get("usage_snapshot") or {}).get("lastJobId"),
                    "report_markdown": report,
                    "source_count": source_count,
                    "reused": True,
                }

        job = await self._ensure_job(
            stores=stores,
            session_id=session_id,
            owner_id=owner_id,
            topic=topic,
            config=config,
        )
        job_id = str(job.get("id") or "")
        if not job_id:
            raise RuntimeError("Deep Research job creation returned no id")
        await on_event(
            {
                "type": "job_status",
                "phase": str(job.get("phase") or "initializing"),
                "payload": {"status": str(job.get("status") or "queued"), "jobId": job_id},
            }
        )
        return await self._wait_for_completion(
            stores=stores,
            session_id=session_id,
            job_id=job_id,
            owner_id=owner_id,
            source_count=source_count,
            cancel=cancel,
            on_event=on_event,
        )

    def _stores(self) -> dict[str, Any]:
        state = self._app.state
        stores = {
            "sessions": getattr(state, "deep_research_session_store", None),
            "jobs": getattr(state, "deep_research_job_store", None),
            "events": getattr(state, "deep_research_event_store", None),
            "sources": getattr(state, "deep_research_source_store", None),
            "dispatcher": getattr(state, "deep_research_dispatcher", None),
            "live_hub": getattr(state, "deep_research_live_hub", None),
        }
        missing = [name for name in ("sessions", "jobs", "events", "sources", "dispatcher") if stores[name] is None]
        if missing:
            raise WorkflowError(
                "WORKFLOW_RESOURCE_MISSING",
                "深度研究运行环境不可用",
                details={"missing": missing},
            )
        return stores

    @staticmethod
    def _config(raw: dict[str, Any], instruction: str) -> DeepResearchConfig:
        # A workflow report is a selected-material regeneration.  It must not
        # launch a second collection channel or pause for Deep Research's own
        # multi-agent plan approval; workflow human_input remains the sole
        # interaction protocol at this layer.
        candidate = dict(raw or {})
        if candidate.get("mode") == "multi_agent":
            raise WorkflowError(
                "WORKFLOW_DEEP_RESEARCH_FAILED",
                "工作流写作节点暂不支持 Deep Research 的 multi_agent 计划中断模式",
            )
        candidate.update(
            {
                "collection_mode": "legacy",
                "allowed_material_channels": ["other"],
                "generate_images": False,
                "include_human_feedback": False,
            }
        )
        if instruction:
            candidate["report_instruction"] = instruction[:4_000]
        try:
            return DeepResearchConfig.model_validate(candidate).clamp()
        except Exception as exc:  # noqa: BLE001 - expose a stable workflow error
            raise WorkflowError(
                "WORKFLOW_DEEP_RESEARCH_FAILED",
                "深度研究写作配置无效",
                details={"reason": type(exc).__name__},
            ) from exc

    @staticmethod
    def _execution_fingerprint(
        *,
        topic: str,
        evidence_pack: dict[str, Any],
        config: DeepResearchConfig,
        instruction: str,
    ) -> str:
        items = evidence_pack.get("items") if isinstance(evidence_pack, dict) else []
        compact: list[dict[str, str]] = []
        for item in items[:_MAX_EVIDENCE_ITEMS] if isinstance(items, list) else []:
            if not isinstance(item, dict):
                continue
            source = item.get("source") if isinstance(item.get("source"), dict) else {}
            compact.append(
                {
                    "claim": _text(item.get("claim"), limit=_MAX_EVIDENCE_ITEM_CHARS),
                    "role": _text(item.get("role"), limit=400),
                    "node": _text(source.get("nodeId"), limit=120),
                    "uri": _text(source.get("uri"), limit=2_000),
                }
            )
        return hashlib.sha256(
            json.dumps(
                {
                    "topic": _text(topic, limit=8_000),
                    "evidence": compact,
                    "config": config.model_dump(),
                    "instruction": _text(instruction, limit=4_000),
                },
                ensure_ascii=False,
                sort_keys=True,
            ).encode("utf-8")
        ).hexdigest()[:32]

    async def _ensure_session_and_sources(
        self,
        *,
        stores: dict[str, Any],
        session_id: str,
        run_id: str,
        node_id: str,
        owner_id: str,
        topic: str,
        evidence_pack: dict[str, Any],
        config: DeepResearchConfig,
    ) -> int:
        items = evidence_pack.get("items")
        if not isinstance(items, list) or not items:
            raise WorkflowError(
                "WORKFLOW_DEEP_RESEARCH_FAILED",
                "Evidence Pack 没有可供写作的证据项",
            )
        usable = [item for item in items if isinstance(item, dict)][: _MAX_EVIDENCE_ITEMS]
        if not usable:
            raise WorkflowError(
                "WORKFLOW_DEEP_RESEARCH_FAILED",
                "Evidence Pack 没有有效的证据项",
            )

        session = await stores["sessions"].get(session_id, user_id=owner_id)
        if session is None:
            await stores["sessions"].create(
                id=session_id,
                user_id=owner_id,
                title=_text(topic, limit=60) or "工作流深度研究报告",
                query=topic,
                mode=config.mode,
                config_snapshot={
                    **config.model_dump(),
                    "workflow_managed": True,
                    "workflow_run_id": run_id,
                    "workflow_node_id": node_id,
                },
            )

        written = 0
        for index, item in enumerate(usable, start=1):
            claim = _text(item.get("claim"), limit=_MAX_EVIDENCE_ITEM_CHARS)
            if not claim:
                continue
            source = item.get("source") if isinstance(item.get("source"), dict) else {}
            source_node = _text(source.get("nodeId"), limit=120)
            source_uri = _text(source.get("uri"), limit=2_000) or None
            role = _text(item.get("role"), limit=400) or source_node or f"证据 {index}"
            content = "\n".join(
                [
                    "[工作流 Evidence Pack]",
                    f"角色：{role}",
                    f"来源节点：{source_node or 'unknown'}",
                    "以下内容是上游智能体或工具返回的材料，不自动视为已核验事实。",
                    "",
                    claim,
                ]
            )
            source_id = _stable_id("drs_src", session_id, str(index), source_node, claim)
            content_hash = hashlib.sha256(content.encode("utf-8")).hexdigest()
            await stores["sources"].upsert(
                session_id=session_id,
                job_id=None,
                id=source_id,
                user_id=owner_id,
                first_seen_query=topic,
                title=role,
                url=source_uri,
                raw_content=content,
                snippet=claim[:1_000],
                source="workflow_evidence_pack",
                source_type="other",
                relevance_score=None,
                rec_uuid=None,
                content_hash=content_hash,
                selected=True,
                selection_reason="workflow_evidence_pack",
                citation_key=str(index),
            )
            written += 1
        if written < 1:
            raise WorkflowError(
                "WORKFLOW_DEEP_RESEARCH_FAILED",
                "Evidence Pack 中没有可写入的文本证据",
            )
        await stores["sessions"].update(
            session_id,
            user_id=owner_id,
            source_count=written,
        )
        return written

    async def _ensure_job(
        self,
        *,
        stores: dict[str, Any],
        session_id: str,
        owner_id: str,
        topic: str,
        config: DeepResearchConfig,
    ) -> dict[str, Any]:
        active = await stores["jobs"].get_active_for_session(session_id, user_id=owner_id)
        if active is not None:
            return active

        active_count = await stores["sessions"].count_active_by_user(user_id=owner_id)
        if active_count >= DEFAULT_MAX_ACTIVE_JOBS_PER_USER:
            raise WorkflowError(
                "WORKFLOW_LIMIT_EXCEEDED",
                "当前深度研究任务过多，请稍后重试",
                retryable=True,
                details={"limit": DEFAULT_MAX_ACTIVE_JOBS_PER_USER},
            )

        request_id = f"workflow:{session_id}"
        request_hash = hashlib.sha256(
            json.dumps(
                {"topic": topic, "config": config.model_dump()},
                ensure_ascii=False,
                sort_keys=True,
            ).encode("utf-8")
        ).hexdigest()[:32]
        job_id = _stable_id("drj_wf", session_id, request_hash)
        input_snapshot = {
            "session_id": session_id,
            "user_id": owner_id,
            "query": topic,
            "config": config.model_dump(),
            "title": _text(topic, limit=60),
            "runtime_thread_id": None,
            # The selected source rows are the exact Evidence Pack snapshot.
            "entry": "regenerate",
        }
        job, _created = await stores["jobs"].try_create_or_get_active(
            id=job_id,
            session_id=session_id,
            user_id=owner_id,
            request_id=request_id,
            request_hash=request_hash,
            input_snapshot=input_snapshot,
        )
        await stores["sessions"].update(
            session_id,
            user_id=owner_id,
            status="running",
            active_job_id=job["id"],
            error=None,
        )
        stores["dispatcher"].nudge()
        return job

    async def _wait_for_completion(
        self,
        *,
        stores: dict[str, Any],
        session_id: str,
        job_id: str,
        owner_id: str,
        source_count: int,
        cancel: CancelToken,
        on_event: ProgressCallback,
    ) -> dict[str, Any]:
        cursor = 0
        live_queue = stores["live_hub"].subscribe(job_id) if stores["live_hub"] is not None else None
        live_report_seen = False
        try:
            while True:
                if cancel.cancelled:
                    await stores["jobs"].request_cancel(job_id, user_id=owner_id)
                    cancel.raise_if_cancelled()

                # A local subscriber sees token deltas immediately. Persisted
                # envelopes from the same hub are ignored; the database replay
                # below owns their sequence and prevents double delivery.
                if live_queue is not None:
                    try:
                        live_event = live_queue.get_nowait()
                    except asyncio.QueueEmpty:
                        live_event = None
                    if isinstance(live_event, dict) and int(live_event.get("seq") or 0) <= 0:
                        event_type = str(live_event.get("type") or "")
                        if event_type == "report_delta":
                            live_report_seen = True
                        await self._emit_event(on_event, live_event)

                events = await stores["events"].list_after(
                    job_id,
                    after=cursor,
                    limit=_EVENT_PAGE_SIZE,
                )
                for event in events:
                    cursor = max(cursor, int(event.get("seq") or 0))
                    if str(event.get("event_type") or "") == "report_chunk" and live_report_seen:
                        continue
                    await self._emit_event(on_event, event)

                job = await stores["jobs"].get(job_id, user_id=owner_id)
                if job is None:
                    raise RuntimeError("Deep Research job no longer exists")
                status = str(job.get("status") or "")
                if status == "completed":
                    session = await stores["sessions"].get(session_id, user_id=owner_id)
                    report = str((session or {}).get("report_markdown") or "").strip()
                    if not report:
                        raise RuntimeError("Deep Research completed without a durable report")
                    return {
                        "session_id": session_id,
                        "job_id": job_id,
                        "report_markdown": report,
                        "source_count": source_count,
                    }
                if status in TERMINAL_STATUSES:
                    if status == "cancelled":
                        cancel.cancel("deep_research_cancelled")
                        cancel.raise_if_cancelled()
                    raise RuntimeError(str(job.get("error_message") or "Deep Research job failed"))

                await asyncio.sleep(_POLL_SECONDS)
        finally:
            if live_queue is not None:
                stores["live_hub"].unsubscribe(job_id, live_queue)

    @staticmethod
    async def _emit_event(on_event: ProgressCallback, event: dict[str, Any]) -> None:
        await on_event(
            {
                "type": str(event.get("event_type") or event.get("type") or "progress"),
                "phase": str(event.get("phase") or ""),
                "payload": event.get("payload") if isinstance(event.get("payload"), dict) else {},
            }
        )


__all__ = ["WorkflowDeepResearchAdapter"]