feat(open-3qfx): 开放接口按 taskId 直接发起 3qfx 第一次问答

新增 POST /api/open/3qfx/ask（无需登录，/api/open/ 白名单 + CSRF 豁免）：只传一个 taskId，后端解析 3Q 业务配置的智能体 + 按 id 拉取 cop-task-detail 拼出与前端一致的开场白，新建带 metadata.{taskId,agent_id} 的真实会话并后台启动持久化运行，立即返回 thread_id（随后在 3qfx 任务工作区对话列表观看）。 - start_run() 增加 keyword-only enforce_access（默认 True）；开放接口传 False 绕过按当前（无）登录用户的智能体可见性校验 - task_deeplink 配置新增 cop_task_detail_url / _authorization / _test + mock - tests/test_open_3qfx.py（10 项，tests/ 被 .gitignore 故未入库） Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-25 19:19:17 +08:00
1 parent d35d121261
commit fe7c6cb8f7
@@ -244,6 +244,7 @@ FastAPI application on port 8001 with health check at `GET /health`. Set `GATEWA
 | **Feedback** (`/api/threads/{id}/runs/{rid}/feedback`) | `PUT /` - upsert feedback; `DELETE /` - delete user feedback; `POST /` - create feedback; `GET /` - list feedback; `GET /stats` - aggregate stats; `DELETE /{fid}` - delete specific |
 | **Runs** (`/api/runs`) | `POST /stream` - stateless run + SSE; `POST /wait` - stateless run + block; `GET /{rid}/messages` - paginated messages by run_id `{data, has_more}` (cursor: `after_seq`/`before_seq`); `GET /{rid}/feedback` - list feedback by run_id |
 | **Open Chat** (`/api/open/chat`) | **Unauthenticated** open Q&A entry for external systems (the `/api/open/` prefix is whitelisted in `auth_middleware._PUBLIC_PATH_PREFIXES` + CSRF-exempt in `csrf_middleware.should_check_csrf`). `POST /chat` `{message \| messages[], model_name?, agent_name?, agent_id?, thinking_enabled=false, strip_think=true, thread_id?, recursion_limit=100}` → resolves the agent by **Chinese display name** (`agent_name`, matched against each `.deer-flow/agents/*/config.yaml` `name` via `_resolve_agent_id`; explicit `agent_id` wins if given), builds the lead agent via `make_lead_agent` (so the chosen **agent's** SOUL/tools/skills all apply), `ainvoke`s it once (stateless — no checkpointer, no thread persistence; runs under the `"default"` user bucket), and returns **only the final** `{answer, model_name, agent_name, agent_id, thinking_enabled, thread_id}` (no intermediate tool-call/process). **`POST /chat/stream`** is the **SSE streaming variant** (same request body) for **long tasks** — the blocking `/chat` produces no downstream bytes for the whole run, so a front nginx `proxy_read_timeout` (default 60s) cuts it off with a **504**; the stream keeps the connection warm (a `: keepalive` SSE comment every `_HEARTBEAT_SECONDS=15`s via a background producer-task + queue, plus `event: delta` AI-text increments and `event: progress`), and ends with **`event: result`** carrying the same final `{answer, …}` JSON (or `event: error`). Clients only need to read `event: result`; intermediate events are ignorable. Both endpoints share `_build_run_context` (config/agent/thread). `GET /agents` lists `{name, id}` so callers know what `agent_name` to send. **Thinking off is reliable**: when `thinking_enabled=false` the route also sets `thinking_force_disabled=true` so the lead agent calls `create_chat_model(force_disable_thinking=True)`, which injects **three** OpenAI-compatible "thinking off" shapes into `extra_body` — top-level `enable_thinking=false` (**SiliconFlow / DashScope hybrid models** like `deepseek-ai/DeepSeek-V4-*` on `api.siliconflow.cn` — the knob the nested `chat_template_kwargs` shape does NOT reach), `chat_template_kwargs.enable_thinking=false` (vLLM/Qwen native), and `thinking.type="disabled"` (Zhipu GLM) — even for models that only declare `supports_thinking: true`. For the **web-UI path** (which sends `thinking_enabled=false` but NOT `force_disable`), SiliconFlow models need `when_thinking_disabled: {extra_body: {enable_thinking: false}}` in their `config.yaml` entry (added to `deepseek-ai/DeepSeek-V4-Flash`) for thinking-off to take effect. Sets `is_scheduled_run=true` internally to skip the auto-title LLM call. **Generated-file return (`include_files=true`, default on)**: an agent that produces its report as a file (writes to `/mnt/user-data/outputs/` + `present_files`) returns only a short summary in `answer`; to recover the real deliverable both endpoints now also return `files: [{name, virtual_path, encoding(text|base64), content}]` — `_collect_output_files` scans the run's `state['thread_data']['outputs_path']` (the open thread is stateless so artifacts can't be fetched by thread, but the endpoint runs server-side and the sandbox already wrote the files to disk), text decoded as utf-8 else base64, capped by `_MAX_OUTPUT_FILES=20` / `_MAX_OUTPUT_FILE_BYTES=5MB`. `/chat` puts `files` on the response; `/chat/stream` puts it in the `event: result` payload. A run with no answer **and** no files now 502s ("…文本答案或文件"). Router `app/gateway/routers/open_chat.py`; standalone caller `test_open_chat_api.py` (repo root, streaming by default); tests `tests/test_open_chat.py`. |
+| **Open 3qfx Ask** (`/api/open/3qfx/ask`) | **Unauthenticated** open entry that, given just a **任务id**, directly kicks off the 3qfx（三情分析）「第一次问答」as a **real persisted conversation** (sibling of `/api/open/chat`; same `/api/open/` auth-whitelist + CSRF-exempt). `POST /ask` `{task_id, message?, model_name?, thinking_enabled=false}` → (1) resolves the `3Q` agent from `app.state.business_mapping_store.get_mapping("3Q")` (the **same** agent the frontend deep-link `goPath=3qfx` uses; 503 if unconfigured); (2) **fetches the task detail server-side** (consumer `cop-task-detail`) and builds the opening text **identical to the frontend** `buildOpeningText` 3qfx branch — `任务名称：…\n任务内容：…\n任务id（task_id）为{task_id}` (an explicit `message` overrides and **skips** the fetch); (3) creates a fresh thread tagged `metadata.{taskId, agent_id}` (the exact filter `useTaskThreads` uses, and `metadata.taskId` makes it task-deeplink-shared / creator-path-resolved) and **starts a background persisted run** via `start_run(..., enforce_access=False)` (writes the checkpointer, runs under the `"default"` bucket); (4) **returns immediately** `{thread_id, run_id, agent_id, status, opening_text}` — the caller just needs "started", then watches the run in the 3qfx 任务工作区 conversation list. `start_run` gained a keyword-only `enforce_access` (default `True`); the open entry passes `False` so the admin-configured agent isn't rejected by the per-(missing)-login visibility check. **Consumer fetch config** lives in `config.yaml → task_deeplink`: `cop_task_detail_url` (full URL, appends `?taskId=`), `cop_task_detail_authorization` (default `Bearer admin`), `cop_task_detail_test` + `mock.cop_task_detail` (offline fake when the upstream is unreachable — on in this deployment). Router `app/gateway/routers/open_3qfx.py`; tests `tests/test_open_3qfx.py`. |
 | **Thread Shares** (`/api/thread-shares`) | Per-conversation sharing with two modes (`thread_shares.mode`). **`import`**: `POST /` `{thread_id, mode}` - create (or reuse) a revocable share code; `GET /{code}/preview` + `POST /{code}/import` - copy that conversation into the caller's account (fresh `thread_id` via `app/gateway/thread_copy.py` + new `threads_meta` row + `user-data` copy; dedup on `metadata.shared_origin`). **`view`**: a public, read-only static page ? see Public Shares below. Both modes: `GET /` - list my codes; `DELETE /{code}` - revoke. |
 | **Public Shares** (`/api/public/shares`) | **Unauthenticated** (the `/api/public/` prefix is whitelisted in `auth_middleware._PUBLIC_PATH_PREFIXES`). `GET /{code}` - read-only conversation snapshot (title + serialized messages + artifacts) for a `view`-mode share; messages are read straight from the latest checkpoint. `GET /{code}/files/{path}` - serves a file referenced by that conversation (images, Markdown, text, PDF, ?) so the shared page's links are openable; path resolution is confined to the owner's thread directory. Active content (`text/html`, `application/xhtml+xml`, `image/svg+xml`) is forced to download (`_public_share_disposition`); everything else is served inline. Missing/revoked/non-`view` codes all return 404. |
 | **Roundtable Drafts** (`/api/roundtable-drafts`) | **Strictly per-user** persistence for the roundtable-planning page (replaces the old browser `localStorage`). `GET /` - list drafts (lightweight meta — incl. `chain_title`, the business-chain name used this session, cheaply regex-extracted from the in-memory `step2` blob in `_draft_to_meta`/`_extract_chain_title`; null = no chain / 自由模式; surfaced on the 第一步「分析记录」history card + sidebar list instead of the old「第 X 步」); `POST /` - create; `GET/PUT/DELETE /{id}` - full read / partial update / delete (all owner-scoped). A draft = one full session (Step 1 intent + Step 2 roundtable), step snapshots stored as JSON in `PortableLongText`. Recommendation history bound to a draft: `GET /{id}/recommendations` - list newest-first; `POST /{id}/recommendations` - append one run (objective/status/rationale/picks/candidates). taskId 深链会商记录 are **not** here — they live in the separate `/api/roundtable-task-drafts` store (see below) so personal history never mixes with task records. Backed by `deerflow.persistence.roundtable_drafts` (tables `roundtable_drafts`, `roundtable_recommend_history`); store wired in `deps.py` as `app.state.roundtable_draft_store`. Tests: `tests/test_roundtable_task_drafts.py`. |
@@ -50,6 +50,7 @@ from app.gateway.routers import (
    models,
    multi_agent,
    notifications,
+    open_3qfx,
    open_chat,
    parallel_agent_tasks,
    parallel_agents,
@@ -1478,6 +1479,9 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
    # 开放问答接口 /api/open/chat —— 无需登录,指定模型 + 智能体做问答,可关闭思考模式
    app.include_router(open_chat.router)

+    # 开放接口 /api/open/3qfx/ask —— 无需登录,按 taskId 直接发起 3qfx 第一次问答(后台运行,落成真实会话)
+    app.include_router(open_3qfx.router)
+
    # --- ?????????? /page/workspace/roundtable/planning?---
    # ??? app/gateway/routers/{multi_agent,intent,recommend}.py?
    # ??? frontend-web/docs/multi-agent-backend-dev.md
@@ -0,0 +1,186 @@
+"""开放接口：按 taskId 直接发起 3qfx（三情分析）的「第一次问答」（无需登录）。
+
+外部系统只需把一个 **任务id** 传进来，后端就会：
+
+1. 解析 ``3Q`` 业务配置的智能体（``business_mapping`` 里 ``business_code=="3Q"`` 那一行的
+   ``agent_id``，与前端任务深链 ``goPath=3qfx`` 取的是同一个）；
+2. 按 taskId 拉取任务详情（consumer 的 ``cop-task-detail``，地址 / 鉴权头 / mock 都在
+   ``config.yaml`` 的 ``task_deeplink`` 段），拼出与前端深链一致的开场白
+   ``任务名称：…\n任务内容：…\n任务id（task_id）为…``；
+3. **新建一条真实会话**（thread 打 ``metadata.{taskId, agent_id}`` 标签，与前端
+   ``taskThreadMetadata`` / ``useTaskThreads`` 的过滤口径一致），以开场白作为第一条用户消息
+   **后台启动一次持久化运行**（写 checkpointer，不阻塞）；
+4. **立即返回** ``{thread_id, run_id, agent_id, status}`` —— 调用方拿到「调用成功」即可，
+   随后在 3qfx 任务工作区的对话列表里就能看到这条问答并实时观看其进展。
+
+该路由前缀 ``/api/open/`` 已在 ``auth_middleware._PUBLIC_PATH_PREFIXES`` 白名单里（且
+``csrf_middleware`` 豁免），**绕过登录鉴权**，以 ``"default"`` 用户桶运行；因 thread 带
+``metadata.taskId``（任务深链对话全局共享、按创建者解析文件路径），任何已登录用户打开该任务
+都能看到同一条会话。
+
+请求示例::
+
+    POST /api/open/3qfx/ask
+    {"task_id": "2055"}
+
+返回::
+
+    {"thread_id": "…", "run_id": "…", "agent_id": "…", "status": "running",
+     "opening_text": "任务名称：…\n任务内容：…\n任务id（task_id）为2055"}
+"""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from typing import Any
+
+import httpx
+from fastapi import APIRouter, HTTPException, Request
+from pydantic import BaseModel, Field
+
+from app.gateway.routers.thread_runs import RunCreateRequest
+from app.gateway.services import start_run
+from deerflow.config import get_app_config
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/open/3qfx", tags=["open"])
+
+# 与前端任务深链 goPath=3qfx → businessCode 的映射保持一致。
+_THREEQ_BUSINESS_CODE = "3Q"
+
+
+class ThreeQAskRequest(BaseModel):
+    """3qfx 开放问答请求体。"""
+
+    task_id: str = Field(..., description="任务id（cop-task-detail 的 taskId）", min_length=1)
+    message: str | None = Field(
+        default=None,
+        description="可选：自定义首条问题文本。留空（默认）则后端按 task_id 拉取任务详情自动拼开场白。",
+    )
+    model_name: str | None = Field(default=None, description="可选：覆盖模型（models[] 中的 name）")
+    thinking_enabled: bool = Field(default=False, description="是否开启思考；默认关闭")
+
+
+class ThreeQAskResponse(BaseModel):
+    thread_id: str
+    run_id: str
+    agent_id: str
+    status: str
+    opening_text: str
+
+
+class CopTaskDetail(BaseModel):
+    """consumer ``cop-task-detail`` 返回的 ``data`` 字段（与前端 CopTaskDetail 对齐）。"""
+
+    id: str | int | None = None
+    overview: str = ""
+    task_direction: str = ""
+    task_name: str = ""
+    task_content: str = ""
+
+
+async def _resolve_threeq_agent_id(request: Request) -> str:
+    """取 ``business_code=="3Q"`` 配置的智能体 id；未配置 → 503。"""
+    store = getattr(request.app.state, "business_mapping_store", None)
+    if store is None:
+        raise HTTPException(status_code=503, detail="业务映射未就绪（business_mapping_store 不可用）。")
+    mapping = await store.get_mapping(_THREEQ_BUSINESS_CODE)
+    agent_id = (mapping or {}).get("agent_id")
+    if not agent_id:
+        raise HTTPException(status_code=503, detail="3Q（三情分析）业务尚未在「业务链条管理」里配置智能体。")
+    return str(agent_id)
+
+
+def _parse_cop_task_detail(body: Any) -> CopTaskDetail:
+    """从 ``{data: {...}}`` 整包里抽出任务详情（字段缺失按空串）。"""
+    data = body.get("data") if isinstance(body, dict) else None
+    if not isinstance(data, dict):
+        return CopTaskDetail()
+    return CopTaskDetail(
+        id=data.get("id"),
+        overview=data.get("overview") if isinstance(data.get("overview"), str) else "",
+        task_direction=data.get("taskDirection") if isinstance(data.get("taskDirection"), str) else "",
+        task_name=data.get("taskName") if isinstance(data.get("taskName"), str) else "",
+        task_content=data.get("taskContent") if isinstance(data.get("taskContent"), str) else "",
+    )
+
+
+async def _fetch_cop_task_detail(task_id: str) -> CopTaskDetail:
+    """按 taskId 取任务详情。``cop_task_detail_test`` 为真走 mock；否则打真实 consumer 接口。"""
+    cfg = get_app_config().task_deeplink
+    if cfg.cop_task_detail_test:
+        return _parse_cop_task_detail((cfg.mock or {}).get("cop_task_detail"))
+
+    url = (cfg.cop_task_detail_url or "").strip()
+    if not url:
+        raise HTTPException(status_code=503, detail="未配置任务详情接口（task_deeplink.cop_task_detail_url），无法按 id 拉取。")
+    try:
+        async with httpx.AsyncClient(timeout=20.0) as client:
+            resp = await client.get(
+                url,
+                params={"taskId": task_id},
+                headers={"Authorization": cfg.cop_task_detail_authorization or "Bearer admin"},
+            )
+            resp.raise_for_status()
+            body = resp.json()
+    except Exception as exc:  # noqa: BLE001 — 开放接口需把上游错误归一成 502
+        logger.warning("拉取 cop-task-detail 失败 (taskId=%s): %s", task_id, exc)
+        raise HTTPException(status_code=502, detail=f"拉取任务详情失败：{exc}") from exc
+    return _parse_cop_task_detail(body)
+
+
+def build_opening_text(detail: CopTaskDetail, task_id: str) -> str:
+    """拼 3qfx 开场白，与前端 ``buildOpeningText``（3qfx 分支）逐行对齐。"""
+    parts = [
+        f"任务名称：{detail.task_name}" if detail.task_name else "",
+        f"任务内容：{detail.task_content}" if detail.task_content else "",
+        f"任务id（task_id）为{task_id}" if task_id else "",
+    ]
+    return "\n".join(p for p in parts if p)
+
+
+@router.post("/ask", response_model=ThreeQAskResponse)
+async def ask_3qfx(body: ThreeQAskRequest, request: Request) -> ThreeQAskResponse:
+    """按 taskId 直接发起 3qfx 第一次问答；后台运行，立即返回 thread_id。"""
+    task_id = body.task_id.strip()
+    if not task_id:
+        raise HTTPException(status_code=422, detail="task_id 不能为空。")
+
+    agent_id = await _resolve_threeq_agent_id(request)
+
+    # 首条问题：默认按 id 自动拉取任务详情拼开场白；显式传 message 则用它覆盖。
+    if body.message and body.message.strip():
+        opening_text = body.message.strip()
+    else:
+        detail = await _fetch_cop_task_detail(task_id)
+        opening_text = build_opening_text(detail, task_id)
+    if not opening_text:
+        raise HTTPException(status_code=502, detail="任务详情为空，无法生成首条问题（可改用 message 显式传入）。")
+
+    thread_id = str(uuid.uuid4())
+    context: dict[str, Any] = {"agent_id": agent_id, "thinking_enabled": body.thinking_enabled}
+    if body.model_name:
+        context["model_name"] = body.model_name
+
+    run_body = RunCreateRequest(
+        assistant_id=agent_id,
+        input={"messages": [{"role": "user", "content": opening_text}]},
+        # taskId + agent_id 让该会话被 3qfx 任务工作区的 useTaskThreads 过滤命中（按任务+业务隔离），
+        # 且 metadata.taskId 触发后端「任务深链对话全局共享 / 按创建者解析文件」语义。
+        metadata={"taskId": task_id, "agent_id": agent_id},
+        context=context,
+        multitask_strategy="reject",
+    )
+
+    # enforce_access=False：开放接口已自行解析 admin 配置的 3Q 智能体，不再按当前（无）登录用户做可见性校验。
+    record = await start_run(run_body, thread_id, request, enforce_access=False)
+
+    return ThreeQAskResponse(
+        thread_id=record.thread_id,
+        run_id=record.run_id,
+        agent_id=agent_id,
+        status=record.status.value,
+        opening_text=opening_text,
+    )
@@ -528,6 +528,8 @@ async def start_run(
    body: Any,
    thread_id: str,
    request: Request,
+    *,
+    enforce_access: bool = True,
 ) -> RunRecord:
    """Create a RunRecord and launch the background agent task.

@@ -559,7 +561,13 @@ async def start_run(
    disconnect = DisconnectMode.cancel if body.on_disconnect == "cancel" else DisconnectMode.continue_

    t_pre_enforce = time.perf_counter()
-    await _enforce_agent_access(body, request)
+    # ``enforce_access=False`` is for intentionally open/unauthenticated entry
+    # points (e.g. /api/open/3qfx/ask) that resolve the agent themselves from an
+    # admin-configured mapping rather than from a per-user-visible assistant_id.
+    # Those run under the "default" bucket, where a user-owned unpublished agent
+    # would otherwise 404 the access check.
+    if enforce_access:
+        await _enforce_agent_access(body, request)
    t_post_enforce = time.perf_counter()

    try:
@@ -435,6 +435,9 @@ task_deeplink:
  test: true            # 目的树查询走 mock.purpose_detail 假数据
  delete_test: true     # 删除接口走 mock.delete 假数据（该接口尚未开发完）
  action_test: true     # 行动→任务解析走 mock.action_detail 假数据（xdfx 行动id→任务id）
+  # 开放接口 POST /api/open/3qfx/ask 按 taskId 拉取任务详情拼开场白；上游不可达时打开下面开关走 mock.cop_task_detail。
+  cop_task_detail_test: true
+  cop_task_detail_url: "https://ch1.b.uat.4.cn/consumer/taskAnalyseSearch/cop-task-detail"
  purpose_detail_url: "https://ch1.b.uat.4.cn/consumer/taskAnalyseSearch/getTaskPurposeDetail"
  # rwfx 与 xdfx 共用同一删除接口：rwfx 追加 ?taskId=（不传 actionId）；xdfx 追加 ?actionId=（不传 taskId，直接不出现）。
  delete_url: "https://ch1.b.uat.4.cn/bwjt/dropSixOrSevenData"
@@ -486,3 +489,11 @@ task_deeplink:
        actionName: ""
        actionStatus: 25
        narration: ""
+    # cop_task_detail_test=true 时原样返回给开放接口 /api/open/3qfx/ask（结构同真实 cop-task-detail 返回）。
+    cop_task_detail:
+      data:
+        id: 2055
+        overview: "暑期文旅消费高峰期间的舆情态势综合研判与引导任务"
+        taskDirection: "正向引导 + 风险预警"
+        taskName: "任务01：暑期文旅舆情综合研判"
+        taskContent: "围绕暑期文旅消费高峰，对重点城市、重点景区的公众情绪与舆情态势做综合研判，识别情绪拐点与潜在风险，提出正向引导与处置建议。"
@@ -41,6 +41,18 @@ class TaskDeeplinkConfig(BaseModel):
        default=False,
        description="为真时前端「行动→任务」解析走 mock.action_detail 假数据（不打真实接口）",
    )
+    cop_task_detail_test: bool = Field(
+        default=False,
+        description="为真时开放接口 POST /api/open/3qfx/ask 按 taskId 取任务详情走 mock.cop_task_detail 假数据（上游 consumer 不可达时用）",
+    )
+    cop_task_detail_url: str = Field(
+        default="",
+        description="任务详情接口完整 URL（开放接口 /api/open/3qfx/ask 后端按 taskId 拉取，追加 ?taskId=）；与前端 cop-task-detail 同一上游。留空则该开放接口报 503。",
+    )
+    cop_task_detail_authorization: str = Field(
+        default="Bearer admin",
+        description="拉取任务详情时下发的 Authorization 头（与前端 axios 拦截器一致，默认 Bearer admin）",
+    )
    purpose_detail_url: str = Field(
        default="",
        description="目的树查询完整 URL，调用时追加 ?taskId=",