feat: add unified strategy project workflow

offline-backend-20260512/backend/CLAUDE.md

@@ -542,6 +542,8 @@ Self-contained product knowledge base. **All knowledge capability lives in this
 
 Custom agent metadata lives in the `agents` ORM table (`deerflow.persistence.agents`). `agents.id` is the stable runtime id and the filesystem directory name under `.deer-flow/agents/{id}/`; `agents.name` is display-only, can contain Chinese characters, and is not unique. `user_id IS NULL` marks built-in agents, user-owned agents are visible only to their owner unless `published = true`, and update/delete operations require ownership. The runtime accepts `agent_id` in context/configurable, and non-default `assistant_id` values are treated as agent ids after visibility checks.
 
+Strategy-selection fan-out (`routers/parallel_agents.py`) must inspect its final AI message with the shared `is_llm_error_message` detector and require substantive visible text before constructing a success result. A middleware fallback, empty/thinking-only response, or punctuation-only provider degeneration is a failed seat run: raise so the stream emits `agent_error`; never wrap it in `agent_done`. Regression coverage: `tests/test_roundtable_model_fallback.py`.
+
 **Listing cache + pagination** ? to keep `GET /api/agents` fast, the gallery's per-card extras (skills + tool_groups) are denormalized into two cache tables instead of reading every agent's `config.yaml` on each request: `agent_skills` (relational, one row per (agent, skill) with `position`) and `agent_extras` (one row per agent: `tool_groups` JSON + `model` + `skills_synced_at`). `config.yaml` remains authoritative for the agent runtime; these tables are a read cache, write-through on create/update and backfilled from disk by `_sync_legacy_agents` (mtime-vs-`skills_synced_at` drift check). The list endpoint enriches via `AgentStore.get_extras_for([ids])` (two batched `IN` queries). Both tables are auto-created by `Base.metadata.create_all` (no migration needed). `GET /api/agents` supports **opt-in server-side pagination**: `scope` (`mine` = owned + favorited / `square` = built-in/published public), `page`, `page_size`, returning `{agents, total, page, page_size}`; omitting `scope`/`page` preserves the legacy "return everything" behavior. Name `search` matches the raw OR desensitized name (the frontend's static term?abbrev map is mirrored server-side in `deerflow/config/agent_name_desensitize.py`); `tag_ids` is an SQL OR-filter; `owner` keeps only agents published by that user id (the card's ??? ? built-in agents never match), powering the gallery's click-???-to-filter. `POST /api/agents/private-square/list` takes the same pagination params (incl. `owner`). Repository entry point: `AgentStore.list_paginated(...)`. Tests: `tests/test_agent_listing_cache.py::test_list_paginated_owner_filter_keeps_one_publisher`.
 
 **Homepage selector (agent picker on the new-chat page)** ? single admin-curated list backed by `agents.featured_order` (nullable BIGINT). NULL = not featured. Only `published = true` agents can be featured (router enforces on writes); any published agent is eligible regardless of ownership. Cleared with `order: null`. The list rendered to end users is `WHERE featured_order IS NOT NULL` + visibility-filter, sorted by `featured_order ASC`. Admins reorder by sending the desired ids to `PUT /api/agents/featured/order` (assigns indices 0..N-1). BIGINT is used rather than INT so the management UI can seed initial values from `Date.now()` (overflows INT on MySQL).
@@ -1000,6 +1002,15 @@ scheduler runs once per deployment. Per run:
 - Supports `supports_vision` flag for image understanding models
 - Config values starting with `$` resolved as environment variables
 - Missing provider modules surface actionable install hints from reflection resolvers (for example `uv add langchain-google-genai`)
+- Generic `langchain_openai:ChatOpenAI` entries are transparently wrapped by
+  `PatchedChatOpenAI`. Besides preserving reasoning/tool-call metadata, the
+  patch recognizes compatible gateways that encode failures inside HTTP 200
+  payloads (`base_resp.status_code != 0` with null/empty `choices`) in both
+  streaming and non-streaming paths. It raises the stable
+  `OpenAICompatibleUpstreamError`; `LLMErrorHandlingMiddleware` retries
+  transient concurrency/inference failures, while genuine quota/balance
+  failures are classified as non-retriable and surfaced immediately. Embedded
+  vendor failures therefore cannot become empty successful assistant turns.
 
 ### Roundtable Model Fault Tolerance (会商模型自动容错 / 换下一个模型)
 

offline-backend-20260512/backend/README.md

@@ -471,7 +471,7 @@ Custom agents are indexed in the `agents` database table. The `id` column is the
 
 The built-in `forced-research-responder` (「强制检索输出助手」) is seeded at Gateway startup. Unlike an ordinary skill-enabled custom agent, it has a runtime-enforced per-turn collection gate: the model cannot finish a response until a retrieval/knowledge/MCP tool or a skill Python script returns usable information. It then answers directly in the user's requested chat format rather than generating an artifact. The agent may create or edit only temporary `.py` files below `/mnt/user-data/workspace/` and may execute `.py` skill scripts through `bash`; writes to `outputs`, Markdown/document creation, `present_files`, general shell commands, and skill mutation are denied. Administrators can edit its model and skill allowlist from agent management; the runtime safety policy remains fixed. Regression coverage: `tests/test_forced_research_agent.py`.
 
-The strategy-selection workbench owns nine folder-backed functional agents: one intent agent, six independent angle agents (governance, diplomacy, economy/supply-chain, security risk, public opinion, and law/policy), one judge, and one final solution editor. Defaults ship in `app/gateway/routers/_strategy_selection_seed_assets/` and are copied only when an agent directory is absent, before `_sync_legacy_agents`; existing administrator edits are never overwritten. The angle agents use the existing `policy-knowledge` skill and read/search tool groups, intent uses `intent-clarification-policy`, judging is input-only, and the solution editor has file-write access for `/mnt/user-data/outputs/综合策略方案.md`. Every default SOUL requires neutral evidence handling, de-escalation and civilian protection, and prohibits operational combat guidance or targeted political manipulation. Regression coverage: `tests/test_strategy_selection_seed.py`.
+The strategy-selection workbench owns nine folder-backed functional agents: one intent agent, six independent angle agents (governance, diplomacy, economy/supply-chain, security risk, public opinion, and law/policy), one judge, and one final solution editor. Defaults ship in `app/gateway/routers/_strategy_selection_seed_assets/` and are copied only when an agent directory is absent, before `_sync_legacy_agents`; existing administrator edits are never overwritten. The angle agents use the existing `policy-knowledge` skill and read/search tool groups, intent uses `intent-clarification-policy`, judging is input-only, and the solution editor has file-write access for `/mnt/user-data/outputs/综合策略方案.md`. Every default SOUL requires neutral evidence handling, de-escalation and civilian protection, and prohibits operational combat guidance or targeted political manipulation. The parallel strategy stream treats a final `is_llm_error_message` fallback, empty/thinking-only response, or punctuation-only provider degeneration as `agent_error`, never as a successful `agent_done`. Regression coverage: `tests/test_strategy_selection_seed.py` and `tests/test_roundtable_model_fallback.py`.
 
 ### AI Writing Intranet Retrieval
 
@@ -545,6 +545,13 @@ the configured model has an explicit, proxy-supported thinking control. This
 prevents DeerFlow's connection check from sending vendor-specific `thinking`
 fields that a generic OpenAI-compatible gateway may reject.
 
+OpenAI-compatible gateways that return vendor errors inside an HTTP 200
+response (for example, a non-zero `base_resp.status_code` with empty/null
+`choices`) are treated as upstream failures in both streaming and
+non-streaming calls. DeerFlow retries transient upstream responses instead of
+accepting an empty assistant turn, while genuine quota/balance failures are
+classified as non-retriable and surfaced immediately instead of being retried.
+
 ### Running
 
 **Full Application** (from project root):

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

@@ -15,7 +15,8 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request
 from pydantic import BaseModel, Field
 
 from app.gateway.auth.password import verify_password_async
-from app.gateway.deps import get_agent_store, get_config, get_tag_store
+from app.gateway.deps import get_agent_store, get_config, get_skill_store, get_tag_store
+from app.gateway.routers._strategy_selection_seed import required_agent_ids as required_strategy_agent_ids
 from deerflow.agents.lead_agent.prompt import refresh_skills_system_prompt_cache_async
 from deerflow.config.agents_api_config import get_agents_api_config
 from deerflow.config.agents_config import CommonQuestion, list_custom_agents, load_agent_config, load_agent_soul, validate_agent_id
@@ -23,13 +24,17 @@ from deerflow.config.app_config import AppConfig
 from deerflow.config.paths import get_paths
 from deerflow.config.system_settings import load_system_settings, normalize_square_id, resolve_default_square_id
 from deerflow.persistence.agents import AgentStore
+from deerflow.persistence.skills import SkillStore
 from deerflow.persistence.tags import TagStore
 from deerflow.runtime.user_context import get_effective_user_id
 from deerflow.skills.cascade import sync_agent_skills_to_available
 from deerflow.skills.storage import get_or_new_skill_storage
+from deerflow.skills.types import SkillCategory
+from deerflow.skills.usage import all_entries as get_skill_usage_entries
 
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api", tags=["agents"])
+_STRATEGY_SELECTION_AGENT_IDS = frozenset(required_strategy_agent_ids())
 
 
 class AgentResponse(BaseModel):
@@ -276,6 +281,109 @@ def _validate_storage_id(agent_id: str) -> str:
         raise HTTPException(status_code=422, detail=str(exc)) from exc
 
 
+async def _validate_strategy_agent_skills(
+    requested_skills: list[str] | None,
+    *,
+    app_config: AppConfig,
+    skill_store: SkillStore,
+) -> list[str] | None:
+    """Validate a global strategy role's skill allowlist.
+
+    Strategy-selection agents are built-in roles shared by every project and
+    every user.  Their allowlists may therefore reference only canonical skill
+    names that exist on disk, are enabled/active, and are globally visible:
+    deployment-owned public skills, explicit built-in rows, or custom skills
+    currently published by their owner.  In particular, an administrator must
+    not be able to attach a different user's unpublished skill merely because
+    ``?admin=true`` exposes it in the management catalog.
+    """
+
+    if requested_skills is None:
+        return None
+
+    normalized: list[str] = []
+    for raw_name in requested_skills:
+        name = raw_name.strip()
+        if not name:
+            raise HTTPException(
+                status_code=422,
+                detail="Strategy agent skill names cannot be blank",
+            )
+        if name not in normalized:
+            normalized.append(name)
+
+    try:
+        skills = get_or_new_skill_storage(app_config=app_config).load_skills(
+            enabled_only=False
+        )
+        ownership = {
+            str(record.get("name") or ""): record
+            for record in await skill_store.list_all()
+            if record.get("name")
+        }
+        usage_entries = get_skill_usage_entries()
+    except HTTPException:
+        raise
+    except Exception as exc:  # noqa: BLE001 - stable API response
+        logger.exception("Failed to validate strategy-agent skill allowlist")
+        raise HTTPException(status_code=503, detail="Skill catalog is unavailable") from exc
+
+    by_name = {skill.name: skill for skill in skills}
+    unknown = [name for name in normalized if name not in by_name]
+    if unknown:
+        raise HTTPException(
+            status_code=422,
+            detail=f"Unknown strategy agent skills: {', '.join(unknown)}",
+        )
+
+    unavailable = [
+        name
+        for name in normalized
+        if not by_name[name].enabled
+        or str((usage_entries.get(name) or {}).get("state") or "active") != "active"
+    ]
+    if unavailable:
+        raise HTTPException(
+            status_code=422,
+            detail=f"Disabled or archived strategy agent skills: {', '.join(unavailable)}",
+        )
+
+    not_global: list[str] = []
+    for name in normalized:
+        skill = by_name[name]
+        if skill.category != SkillCategory.CUSTOM:
+            continue
+        record = ownership.get(name)
+        if record is not None and (
+            record.get("owner_user_id") is None or bool(record.get("published"))
+        ):
+            continue
+        not_global.append(name)
+    if not_global:
+        raise HTTPException(
+            status_code=422,
+            detail=(
+                "Strategy agent skills must be globally available; "
+                f"unpublished or unregistered custom skills: {', '.join(not_global)}"
+            ),
+        )
+
+    return normalized
+
+
+def _ensure_strategy_agent_deletable(agent_id: str) -> None:
+    """Keep fixed strategy role ids alive; their SOUL/skills remain editable."""
+
+    if agent_id in _STRATEGY_SELECTION_AGENT_IDS:
+        raise HTTPException(
+            status_code=409,
+            detail=(
+                f"Strategy role agent '{agent_id}' cannot be deleted; "
+                "edit its SOUL and skills instead"
+            ),
+        )
+
+
 def _write_agent_config(
     agent_id: str,
     *,
@@ -953,7 +1061,14 @@ async def update_agent(agent_id: str, request: AgentUpdateRequest, http_request:
         next_description = update_data.get("description", current.get("description") or "")
 
         existing_extras = _load_config_extras(agent_id)
-        next_skills = request.skills if "skills" in fields_set else existing_extras["skills"]
+        if "skills" in fields_set and agent_id in _STRATEGY_SELECTION_AGENT_IDS:
+            next_skills = await _validate_strategy_agent_skills(
+                request.skills,
+                app_config=get_config(http_request),
+                skill_store=get_skill_store(http_request),
+            )
+        else:
+            next_skills = request.skills if "skills" in fields_set else existing_extras["skills"]
         next_model = request.model if "model" in fields_set else existing_extras["model"]
         next_tool_groups = request.tool_groups if "tool_groups" in fields_set else existing_extras["tool_groups"]
         next_common_questions = request.common_questions if "common_questions" in fields_set else existing_extras["common_questions"]
@@ -1079,6 +1194,7 @@ async def delete_agent(agent_id: str, request: Request) -> None:
         _, edit_mode = await _get_editable_or_raise(
             store, agent_id, user_id, is_admin=is_admin
         )
+        _ensure_strategy_agent_deletable(agent_id)
 
         if edit_mode == "admin":
             deleted = await store.delete_any(agent_id)

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

@@ -31,13 +31,15 @@ from __future__ import annotations
 import asyncio
 import json
 import logging
-from typing import Any, AsyncIterator
+from collections.abc import AsyncIterator
+from typing import Any
 
 import httpx
 from fastapi import APIRouter, Body, HTTPException, Request
 from fastapi.responses import StreamingResponse
 from pydantic import BaseModel, Field
 
+from app.gateway.roundtable_model_fallback import is_llm_error_message, reason_text
 from app.gateway.routers.multi_agent import (
     _ROUNDTABLE_THREAD_METADATA,
     _auth_headers,
@@ -50,6 +52,7 @@ from app.gateway.routers.multi_agent import (
     _stream_upstream,
     _validate_agent_id,
 )
+from deerflow.agents.roundtable_orchestrator.delivery import visible_seat_text
 from deerflow.tools.builtins.clarification_utils import resolve_allow_multiple
 
 logger = logging.getLogger(__name__)
@@ -58,6 +61,17 @@ router = APIRouter(prefix="/api/parallel-agents", tags=["parallel-agents"])
 # Soft cap on parallel degree — protects remote model concurrency + sandbox.
 MAX_PARALLEL_AGENTS = 8
 
+# Strategy roles are expected to return a substantive task/strategy/evaluation,
+# not merely punctuation or a broken provider placeholder.  Counting Unicode
+# alphanumerics keeps the check language-neutral while rejecting outputs such as
+# a long run of "!!!", which otherwise look non-empty to the UI.
+MIN_MEANINGFUL_CONTENT_CHARS = 20
+
+
+def _has_substantive_content(content: str) -> bool:
+    visible = visible_seat_text(content)
+    return sum(character.isalnum() for character in visible) >= MIN_MEANINGFUL_CONTENT_CHARS
+
 # 禁两类工具：①**写文件**（write_file / str_replace）—— 并行席位互相写文件会打架；
 # ②**派活 / 脚本逃逸**（agent_orchestration / bash）—— 本面板里每个智能体只做自己那
 # 份分析交付，绝不该扮演「总控」去调度别人。弱模型（deepseek 等）拿到带「总控 / 业务
@@ -231,6 +245,9 @@ async def _run_one_agent(
         if isinstance(msg, dict) and (msg.get("type") or "").lower() in ("ai", "aimessage", "aimessagechunk"):
             last_ai = msg
             break
+    failure_reason = is_llm_error_message(last_ai)
+    if failure_reason:
+        raise RuntimeError(f"模型调用失败：{reason_text(failure_reason)}")
     content = _flatten_content(last_ai.get("content", ""))
     # 收集**所有** AI 消息的 tool_calls（全量 args，按顺序）—— 用于在收口时补发所有步骤
     # （含 ls / skill_list 这类无参/中间步骤），保证步骤条不丢步、且 args 完整。
@@ -250,6 +267,8 @@ async def _run_one_agent(
             "tool_calls": final_tool_calls,
         }
     else:
+        if not _has_substantive_content(content):
+            raise RuntimeError("智能体未生成有效正文，请重试或更换全局模型")
         yield {
             "status": f"子智能体 {agent_id} 完成",
             "content": content,
@@ -408,8 +427,24 @@ async def run_parallel_agents_stream(request: Request, payload: dict = Body(...)
                             await _emit_tool(tc)
             if not done_sent:
                 # 流被抢占/截断,没收到终态帧 —— 用已累积正文兜底收口本席位。
+                buffered_content = "".join(buffer)
+                failure_reason = is_llm_error_message(
+                    {"type": "ai", "content": buffered_content}
+                )
+                if failure_reason:
+                    raise RuntimeError(
+                        f"模型调用失败：{reason_text(failure_reason)}"
+                    )
+                if not _has_substantive_content(buffered_content):
+                    raise RuntimeError(
+                        "智能体未生成有效正文，请重试或更换全局模型"
+                    )
                 await queue.put(
-                    {"type": "agent_done", "agent_id": agent_id, "content": "".join(buffer)}
+                    {
+                        "type": "agent_done",
+                        "agent_id": agent_id,
+                        "content": buffered_content,
+                    }
                 )
         except Exception as exc:  # noqa: BLE001 — 单个 agent 失败不拖垮其它
             logger.warning("[parallel-agents] %s failed: %r", agent_id, exc)

offline-backend-20260512/backend/extensions_config.json

@@ -26,7 +26,7 @@
       "enabled": true
     },
     "policy-knowledge": {
-      "enabled": false
+      "enabled": true
     },
     "html-page-builder": {
       "enabled": true

offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py

@@ -96,9 +96,11 @@ _BUSY_PATTERNS = (
     "稍后重试",
     "请稍后重试",
 )
-_QUOTA_PATTERNS = (
-    "insufficient_quota",
-    "quota",
+_QUOTA_PATTERNS = (
+    "insufficient_quota",
+    "insufficient balance",
+    "token plan limit",
+    "quota",
     "billing",
     "credit",
     "payment",
@@ -122,8 +124,10 @@ _AUTH_PATTERNS = (
 # error wrapped as `bad_response_status_code`, or a truncated/garbled inference
 # response). These are almost always transient inference-side issues and safe
 # to retry.
-_UPSTREAM_GARBAGE_PATTERNS = (
-    "bad_response_status_code",
+_UPSTREAM_GARBAGE_PATTERNS = (
+    "openai-compatible upstream application error",
+    "null value for 'choices'",
+    "bad_response_status_code",
     "bad response status",
     "expecting ',' delimiter",
     "expecting value",

offline-backend-20260512/backend/packages/harness/deerflow/models/patched_openai.py

@@ -25,7 +25,8 @@ message that originally carried it:
 
 from __future__ import annotations
 
-from typing import Any
+from collections.abc import Mapping
+from typing import Any
 
 from langchain_core.language_models import LanguageModelInput
 from langchain_core.messages import AIMessage, AIMessageChunk
@@ -34,6 +35,65 @@ from langchain_openai import ChatOpenAI
 
 from deerflow.models.message_ordering import normalize_chat_payload_messages
 
+
+class OpenAICompatibleUpstreamError(RuntimeError):
+    """An OpenAI-compatible gateway reported an error inside a HTTP 200 body.
+
+    Several compatible gateways do not use an HTTP error status for quota,
+    concurrency, or inference failures. Instead they return ``choices=null``
+    (or an empty streamed chunk) together with a vendor ``base_resp`` object.
+    LangChain raises a generic ``TypeError`` for the non-streaming shape and
+    treats the streaming shape as an empty successful response. Raising one
+    stable exception here lets DeerFlow's model middleware retry both paths.
+    """
+
+    def __init__(self, *, code: Any, message: str | None = None) -> None:
+        self.code = code
+        detail = _bounded_error_message(message)
+        suffix = f": {detail}" if detail else ""
+        super().__init__(f"OpenAI-compatible upstream application error (code={code}){suffix}")
+
+
+def _bounded_error_message(value: Any, *, limit: int = 500) -> str:
+    """Return a log-safe, bounded one-line provider message."""
+    if not isinstance(value, str):
+        return ""
+    normalized = " ".join(value.split())
+    return normalized[:limit]
+
+
+def _application_error(payload: Mapping[str, Any]) -> tuple[Any, str | None] | None:
+    """Extract a non-success ``base_resp`` error from a compatible payload."""
+    raw = payload.get("base_resp") or payload.get("baseResp")
+    if not isinstance(raw, Mapping):
+        return None
+
+    code = raw.get("status_code")
+    if code is None:
+        code = raw.get("statusCode")
+    if code is None:
+        code = raw.get("code")
+
+    # MiniMax-compatible responses use 0 for success. Be conservative for
+    # gateways that put informational metadata in ``base_resp``.
+    normalized_code = str(code).strip().lower() if code is not None else ""
+    if normalized_code in {"", "0", "200", "ok", "success"}:
+        return None
+
+    message = raw.get("status_msg")
+    if message is None:
+        message = raw.get("statusMsg")
+    if message is None:
+        message = raw.get("message")
+    return code, message if isinstance(message, str) else None
+
+
+def _raise_for_application_error(payload: Mapping[str, Any]) -> None:
+    error = _application_error(payload)
+    if error is not None:
+        code, message = error
+        raise OpenAICompatibleUpstreamError(code=code, message=message)
+
 
 class PatchedChatOpenAI(ChatOpenAI):
     """ChatOpenAI with ``thought_signature`` preservation for Gemini thinking via OpenAI gateway.
@@ -100,16 +160,20 @@ class PatchedChatOpenAI(ChatOpenAI):
         normalize_chat_payload_messages(payload)
         return payload
 
-    def _convert_chunk_to_generation_chunk(
-        self,
-        chunk: dict,
-        default_chunk_class: type,
-        base_generation_info: dict | None,
-    ) -> ChatGenerationChunk | None:
-        """Preserve streamed ``reasoning_content`` in ``additional_kwargs``."""
-        generation_chunk = super()._convert_chunk_to_generation_chunk(
-            chunk,
-            default_chunk_class,
+    def _convert_chunk_to_generation_chunk(
+        self,
+        chunk: dict,
+        default_chunk_class: type,
+        base_generation_info: dict | None,
+    ) -> ChatGenerationChunk | None:
+        """Preserve streamed ``reasoning_content`` in ``additional_kwargs``."""
+        # Some compatible gateways return a vendor error event followed by a
+        # normal-looking usage chunk, all under HTTP 200. Detect the first
+        # event before LangChain converts it into an empty AIMessageChunk.
+        _raise_for_application_error(chunk)
+        generation_chunk = super()._convert_chunk_to_generation_chunk(
+            chunk,
+            default_chunk_class,
             base_generation_info,
         )
         if generation_chunk is None:
@@ -130,11 +194,12 @@ class PatchedChatOpenAI(ChatOpenAI):
         self,
         response: dict | Any,
         generation_info: dict | None = None,
-    ) -> ChatResult:
-        """Preserve non-stream ``reasoning_content`` in ``additional_kwargs``."""
-        result = super()._create_chat_result(response, generation_info)
-        response_dict = response if isinstance(response, dict) else response.model_dump()
-        choices = response_dict.get("choices", [])
+    ) -> ChatResult:
+        """Preserve non-stream ``reasoning_content`` in ``additional_kwargs``."""
+        response_dict = response if isinstance(response, dict) else response.model_dump()
+        _raise_for_application_error(response_dict)
+        result = super()._create_chat_result(response, generation_info)
+        choices = response_dict.get("choices", [])
 
         generations: list[ChatGeneration] = []
         for index, generation in enumerate(result.generations):

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

@@ -11,9 +11,11 @@ import asyncio
 from types import SimpleNamespace
 
 import httpx
-from langchain_core.messages import AIMessage
+import pytest
+from langchain_core.messages import AIMessage, AIMessageChunk
 
 from deerflow.agents.middlewares.llm_error_handling_middleware import LLM_ERROR_MARKER, LLMErrorHandlingMiddleware
+from deerflow.models.patched_openai import OpenAICompatibleUpstreamError, PatchedChatOpenAI
 
 
 def _middleware() -> LLMErrorHandlingMiddleware:
@@ -26,6 +28,10 @@ def _middleware() -> LLMErrorHandlingMiddleware:
     return middleware
 
 
+def _openai_model() -> PatchedChatOpenAI:
+    return PatchedChatOpenAI(model="test-model", api_key="test-key")
+
+
 def test_httpx_read_timeout_is_transient() -> None:
     assert _middleware()._classify_error(httpx.ReadTimeout("read timed out")) == (True, "transient")
 
@@ -46,6 +52,98 @@ def test_value_error_stays_generic() -> None:
     assert _middleware()._classify_error(ValueError("boom")) == (False, "generic")
 
 
+def test_openai_compatible_embedded_error_is_retriable_upstream() -> None:
+    error = RuntimeError("OpenAI-compatible upstream application error (code=2056)")
+    assert _middleware()._classify_error(error) == (True, "upstream")
+
+
+def test_openai_compatible_null_choices_error_is_retriable_upstream() -> None:
+    error = TypeError("Received response with null value for 'choices'.")
+    assert _middleware()._classify_error(error) == (True, "upstream")
+
+
+def test_insufficient_balance_stays_non_retriable_quota() -> None:
+    """The 200-body compatibility fix must not hide a real billing failure."""
+
+    class PaymentError(RuntimeError):
+        status_code = 402
+
+    error = PaymentError("Insufficient Balance")
+    assert _middleware()._classify_error(error) == (False, "quota")
+
+
+def test_minimax_token_plan_limit_stays_non_retriable_quota() -> None:
+    error = OpenAICompatibleUpstreamError(
+        code=2056,
+        message="Token Plan limit reached",
+    )
+
+    assert _middleware()._classify_error(error) == (False, "quota")
+
+
+def test_non_stream_embedded_application_error_is_raised() -> None:
+    response = {
+        "id": "failed-response",
+        "object": "chat.completion",
+        "created": 0,
+        "model": "test-model",
+        "choices": None,
+        "base_resp": {
+            "status_code": 2056,
+            "status_msg": "Token Plan limit reached",
+        },
+    }
+
+    with pytest.raises(OpenAICompatibleUpstreamError, match=r"code=2056") as exc_info:
+        _openai_model()._create_chat_result(response)
+
+    assert exc_info.value.code == 2056
+    assert "Token Plan limit reached" in str(exc_info.value)
+
+
+def test_stream_embedded_application_error_is_raised() -> None:
+    chunk = {
+        "id": "failed-response",
+        "object": "chat.completion.chunk",
+        "created": 0,
+        "model": "test-model",
+        "choices": [],
+        "base_resp": {
+            "status_code": 2056,
+            "status_msg": "Token Plan limit reached",
+        },
+    }
+
+    with pytest.raises(OpenAICompatibleUpstreamError, match=r"code=2056"):
+        _openai_model()._convert_chunk_to_generation_chunk(chunk, AIMessageChunk, None)
+
+
+def test_normal_empty_usage_chunk_is_not_treated_as_error() -> None:
+    chunk = {
+        "id": "usage-response",
+        "object": "chat.completion.chunk",
+        "created": 0,
+        "model": "test-model",
+        "choices": [],
+        "usage": {
+            "prompt_tokens": 3,
+            "completion_tokens": 2,
+            "total_tokens": 5,
+        },
+    }
+
+    result = _openai_model()._convert_chunk_to_generation_chunk(chunk, AIMessageChunk, None)
+
+    assert result is not None
+    assert result.message.usage_metadata == {
+        "input_tokens": 3,
+        "output_tokens": 2,
+        "total_tokens": 5,
+        "input_token_details": {},
+        "output_token_details": {},
+    }
+
+
 def test_read_timeout_is_retried_and_recovers() -> None:
     """第一次 ReadTimeout、第二次成功：必须重试并返回真实结果。"""
     middleware = _middleware()

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

@@ -13,9 +13,14 @@
 from __future__ import annotations
 
 import asyncio
+import json
+from contextlib import asynccontextmanager
 from types import SimpleNamespace
 
+import pytest
+
 import app.gateway.roundtable_model_fallback as mf
+import app.gateway.routers.parallel_agents as parallel_agents
 from app.gateway.roundtable_model_fallback import (
     fallback_model_chain,
     is_llm_error_message,
@@ -56,6 +61,181 @@ def test_is_llm_error_message_normal_is_none() -> None:
     assert is_llm_error_message({"type": "ai", "content": ""}) is None
 
 
+def test_parallel_agent_run_raises_instead_of_completing_llm_fallback(monkeypatch) -> None:
+    fallback_message = {
+        "type": "ai",
+        "content": "大模型本次调用未能成功完成，请稍后再试或更换模型。",
+    }
+
+    async def fake_stream_upstream(*_args, **_kwargs):
+        yield None, [f"data: {json.dumps({'messages': [fallback_message]}, ensure_ascii=False)}"]
+
+    monkeypatch.setattr(parallel_agents, "_stream_upstream", fake_stream_upstream)
+
+    async def collect() -> list[object]:
+        return [
+            item
+            async for item in parallel_agents._run_one_agent(
+                None,  # type: ignore[arg-type]
+                "http://gateway.test",
+                {"Authorization": "Bearer test"},
+                "strategy-selection-intent",
+                "thread-id",
+                "test task",
+                "",
+            )
+        ]
+
+    try:
+        asyncio.run(collect())
+    except RuntimeError as exc:
+        assert "模型调用失败" in str(exc)
+    else:
+        raise AssertionError("LLM fallback message must not be emitted as agent_done")
+
+
+def test_parallel_agent_run_rejects_punctuation_only_delivery(monkeypatch) -> None:
+    invalid_message = {"type": "ai", "content": "!" * 500}
+
+    async def fake_stream_upstream(*_args, **_kwargs):
+        yield None, [f"data: {json.dumps({'messages': [invalid_message]}, ensure_ascii=False)}"]
+
+    monkeypatch.setattr(parallel_agents, "_stream_upstream", fake_stream_upstream)
+
+    async def collect() -> list[object]:
+        return [
+            item
+            async for item in parallel_agents._run_one_agent(
+                None,  # type: ignore[arg-type]
+                "http://gateway.test",
+                {"Authorization": "Bearer test"},
+                "strategy-selection-angle-security",
+                "thread-id",
+                "test task",
+                "",
+            )
+        ]
+
+    try:
+        asyncio.run(collect())
+    except RuntimeError as exc:
+        assert "未生成有效正文" in str(exc)
+    else:
+        raise AssertionError("Punctuation-only output must not be emitted as agent_done")
+
+
+def test_parallel_agent_run_allows_empty_content_clarification(monkeypatch) -> None:
+    clarification_message = {
+        "type": "ai",
+        "content": "",
+        "tool_calls": [
+            {
+                "id": "clarify-1",
+                "name": "ask_clarification",
+                "args": {
+                    "question": "请确认策略时间范围。",
+                    "clarification_type": "single_choice",
+                    "options": ["短期", "长期"],
+                    "allow_custom": True,
+                },
+            }
+        ],
+    }
+
+    async def fake_stream_upstream(*_args, **_kwargs):
+        yield None, [
+            f"data: {json.dumps({'messages': [clarification_message]}, ensure_ascii=False)}"
+        ]
+
+    monkeypatch.setattr(parallel_agents, "_stream_upstream", fake_stream_upstream)
+
+    async def collect() -> list[object]:
+        return [
+            item
+            async for item in parallel_agents._run_one_agent(
+                None,  # type: ignore[arg-type]
+                "http://gateway.test",
+                {"Authorization": "Bearer test"},
+                "strategy-selection-intent",
+                "thread-id",
+                "test task",
+                "",
+            )
+        ]
+
+    events = asyncio.run(collect())
+
+    assert len(events) == 1
+    event = events[0]
+    assert isinstance(event, dict)
+    assert event["type"] == "clarification"
+    assert event["content"] == ""
+    assert event["clarification"]["question"] == "请确认策略时间范围。"
+
+
+def test_parallel_agent_substantive_content_accepts_chinese_and_english() -> None:
+    assert parallel_agents._has_substantive_content("这是一份包含目标、前置条件、阶段路径和验收标准的完整策略。")
+    assert parallel_agents._has_substantive_content(
+        "A complete strategy with objectives, prerequisites, milestones, and measurable outcomes."
+    )
+    assert not parallel_agents._has_substantive_content("<think>internal reasoning only</think>!!!")
+
+
+@pytest.mark.parametrize(
+    ("buffered_content", "expected_error"),
+    [
+        ("大模型本次调用未能成功完成，请稍后再试或更换模型。", "模型调用失败"),
+        ("", "未生成有效正文"),
+        ("<think>internal reasoning only</think>", "未生成有效正文"),
+        ("!" * 500, "未生成有效正文"),
+    ],
+)
+def test_parallel_agent_truncated_stream_rejects_invalid_buffer(
+    monkeypatch,
+    buffered_content: str,
+    expected_error: str,
+) -> None:
+    async def fake_run_one_agent(*_args, **_kwargs):
+        yield "event: messages"
+        yield f"data: {json.dumps([{'type': 'ai', 'content': buffered_content}], ensure_ascii=False)}"
+
+    @asynccontextmanager
+    async def fake_client(*_args, **_kwargs):
+        yield object()
+
+    async def fake_create_thread(*_args, **_kwargs):
+        return "thread-id"
+
+    monkeypatch.setattr(parallel_agents, "_run_one_agent", fake_run_one_agent)
+    monkeypatch.setattr(parallel_agents, "_make_loopback_client", fake_client)
+    monkeypatch.setattr(parallel_agents, "_create_thread", fake_create_thread)
+    monkeypatch.setattr(parallel_agents, "_loopback_base", lambda _request: "http://gateway.test")
+    monkeypatch.setattr(parallel_agents, "_auth_headers", lambda _request: {})
+
+    async def collect() -> list[dict[str, object]]:
+        response = await parallel_agents.run_parallel_agents_stream(
+            SimpleNamespace(),
+            {
+                "task_name": "test task",
+                "brief": "test brief",
+                "agent_ids": ["strategy-selection-intent"],
+            },
+        )
+        events: list[dict[str, object]] = []
+        async for chunk in response.body_iterator:
+            text = chunk.decode() if isinstance(chunk, bytes) else chunk
+            for line in text.splitlines():
+                if line.startswith("data:"):
+                    events.append(json.loads(line[5:].strip()))
+        return events
+
+    events = asyncio.run(collect())
+    assert not any(event.get("type") == "agent_done" for event in events)
+    errors = [event for event in events if event.get("type") == "agent_error"]
+    assert len(errors) == 1
+    assert expected_error in str(errors[0].get("message"))
+
+
 # ── 中间件给兜底 AIMessage 打标记 ────────────────────────────────────────────
 
 

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

@@ -6,9 +6,13 @@ from pathlib import Path
 
 import pytest
 import yaml
+from fastapi import HTTPException
 
 from app.gateway.routers import _strategy_selection_seed as seed
+from app.gateway.routers import agents as agents_router
 from deerflow.config.paths import Paths
+from deerflow.persistence.skills.memory import MemorySkillStore
+from deerflow.skills.types import Skill, SkillCategory
 
 
 @pytest.fixture()
@@ -73,6 +77,18 @@ def test_seed_assets_use_role_appropriate_existing_skills_and_tools() -> None:
     assert solution["tool_groups"] == ["file:write"]
 
 
+def test_default_strategy_skills_are_enabled_in_the_system_catalog() -> None:
+    extensions_path = Path(__file__).resolve().parents[1] / "extensions_config.json"
+    extensions = yaml.safe_load(extensions_path.read_text(encoding="utf-8")) or {}
+    configured = extensions.get("skills") or {}
+
+    # A missing entry means enabled by default.  An explicit false would make
+    # the seeded allowlist impossible to save under the server-side validator
+    # and silently remove the skill from strategy runs.
+    for skill_name in ("intent-clarification-policy", "policy-knowledge"):
+        assert (configured.get(skill_name) or {}).get("enabled", True) is True
+
+
 def test_seed_assets_define_machine_readable_handoff_contracts() -> None:
     intent_soul = _asset(seed.STRATEGY_INTENT_AGENT_ID, "SOUL.md").read_text(encoding="utf-8")
     judge_soul = _asset(seed.STRATEGY_JUDGE_AGENT_ID, "SOUL.md").read_text(encoding="utf-8")
@@ -113,3 +129,120 @@ def test_seed_creates_once_and_never_overwrites_admin_changes(tmp_paths: Paths)
 
     assert seed.ensure_strategy_selection_agents() == []
     assert customized_soul.read_text(encoding="utf-8") == "管理员现场修改"
+
+
+class _SkillStorage:
+    def __init__(self, skills: list[Skill]) -> None:
+        self._skills = skills
+
+    def load_skills(self, *, enabled_only: bool) -> list[Skill]:
+        if enabled_only:
+            return [skill for skill in self._skills if skill.enabled]
+        return list(self._skills)
+
+
+def _skill(name: str, *, category: SkillCategory, enabled: bool = True) -> Skill:
+    directory = Path("skills") / category.value / name
+    return Skill(
+        name=name,
+        description=name,
+        license=None,
+        skill_dir=directory,
+        skill_file=directory / "SKILL.md",
+        relative_path=Path(name),
+        category=category,
+        enabled=enabled,
+    )
+
+
+@pytest.mark.asyncio
+async def test_strategy_skill_validation_accepts_only_global_catalog_names(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    skill_store = MemorySkillStore()
+    await skill_store.create(
+        {
+            "name": "published-custom",
+            "owner_user_id": "publisher",
+            "published": True,
+        }
+    )
+    storage = _SkillStorage(
+        [
+            _skill("public-skill", category=SkillCategory.PUBLIC),
+            _skill("published-custom", category=SkillCategory.CUSTOM),
+        ]
+    )
+    monkeypatch.setattr(agents_router, "get_or_new_skill_storage", lambda **_: storage)
+    monkeypatch.setattr(
+        agents_router,
+        "get_skill_usage_entries",
+        lambda: {},
+    )
+
+    result = await agents_router._validate_strategy_agent_skills(
+        [" public-skill ", "published-custom", "public-skill"],
+        app_config=object(),  # type: ignore[arg-type]
+        skill_store=skill_store,
+    )
+
+    assert result == ["public-skill", "published-custom"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("requested", "detail_fragment"),
+    [
+        ("missing", "Unknown strategy agent skills"),
+        ("disabled", "Disabled or archived strategy agent skills"),
+        ("stale", "Disabled or archived strategy agent skills"),
+        ("private-custom", "must be globally available"),
+        ("orphan-custom", "must be globally available"),
+    ],
+)
+async def test_strategy_skill_validation_rejects_unselectable_names(
+    monkeypatch: pytest.MonkeyPatch,
+    requested: str,
+    detail_fragment: str,
+) -> None:
+    skill_store = MemorySkillStore()
+    await skill_store.create(
+        {
+            "name": "private-custom",
+            "owner_user_id": "owner",
+            "published": False,
+        }
+    )
+    storage = _SkillStorage(
+        [
+            _skill("disabled", category=SkillCategory.PUBLIC, enabled=False),
+            _skill("stale", category=SkillCategory.PUBLIC),
+            _skill("private-custom", category=SkillCategory.CUSTOM),
+            _skill("orphan-custom", category=SkillCategory.CUSTOM),
+        ]
+    )
+    monkeypatch.setattr(agents_router, "get_or_new_skill_storage", lambda **_: storage)
+    monkeypatch.setattr(
+        agents_router,
+        "get_skill_usage_entries",
+        lambda: {"stale": {"state": "stale"}},
+    )
+
+    with pytest.raises(HTTPException) as exc_info:
+        await agents_router._validate_strategy_agent_skills(
+            [requested],
+            app_config=object(),  # type: ignore[arg-type]
+            skill_store=skill_store,
+        )
+
+    assert exc_info.value.status_code == 422
+    assert detail_fragment in str(exc_info.value.detail)
+
+
+def test_fixed_strategy_role_agents_cannot_be_deleted() -> None:
+    for agent_id in seed.required_agent_ids():
+        with pytest.raises(HTTPException) as exc_info:
+            agents_router._ensure_strategy_agent_deletable(agent_id)
+        assert exc_info.value.status_code == 409
+
+    agents_router._ensure_strategy_agent_deletable("ordinary-custom-agent")