From b8ab097ca26cc8b017bee5306104de28caaaa53e Mon Sep 17 00:00:00 2001 From: YxinMiracle <56109594+YxinMiracle@users.noreply.github.com> Date: Tue, 22 Sep 2026 17:53:21 +0800 Subject: [PATCH] fix(models): support official DeepSeek managed profiles (#5718) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: YxinMiracle <“939157765@qq.com”> --- README.md | 25 +- backend/app/gateway/AGENTS.md | 2 +- backend/app/gateway/routers/managed_models.py | 19 +- .../harness/deerflow/config/AGENTS.md | 4 +- .../config/managed_model_providers.py | 29 ++ .../harness/deerflow/config/managed_models.py | 7 +- .../harness/deerflow/models/AGENTS.md | 24 +- backend/tests/AGENTS.md | 10 + .../tests/blocking_io/test_managed_models.py | 53 +++- backend/tests/test_managed_deepseek.py | 281 ++++++++++++++++++ backend/tests/test_managed_deepseek_live.py | 75 +++++ 11 files changed, 512 insertions(+), 17 deletions(-) create mode 100644 backend/packages/harness/deerflow/config/managed_model_providers.py create mode 100644 backend/tests/test_managed_deepseek.py create mode 100644 backend/tests/test_managed_deepseek_live.py diff --git a/README.md b/README.md index fd62fd852..ecba7c541 100644 --- a/README.md +++ b/README.md @@ -162,7 +162,30 @@ It is disabled by default; see the linked guide to enable it. saving refreshes the chat model list. Connection testing sends a short streaming tool-call request and may incur provider charges. It does not save the draft or verify image support; set image support and token limits from provider documentation. - Native provider adapters and advanced reasoning settings remain YAML-configured. + Official DeepSeek models at `https://api.deepseek.com` or + `https://api.deepseek.com/v1` (default HTTPS port) automatically use DeerFlow's + DeepSeek adapter, preserving reasoning content across tool calls and honoring + output token limits. Chat uses the selected thinking mode; the connection test + temporarily disables thinking because DeepSeek rejects forced tool selection + in thinking mode. The test checks streaming tool connectivity, not every agent + workflow or thinking-mode behavior. Existing saved DeepSeek profiles receive + this adapter without re-entering credentials. DeepSeek-specific settings for + third-party proxies, other native adapters, and advanced reasoning settings + remain YAML-configured. + + DeepSeek regression tests run offline with the normal backend suite. To verify + the real provider explicitly, set `DEEPSEEK_TEST_API_KEY` in your environment + and run from `backend/`: + + ```bash + DEER_FLOW_RUN_LIVE_TESTS=1 uv run --no-sync pytest tests/test_managed_deepseek_live.py -q + ``` + + These opt-in tests send short requests to DeepSeek and may incur charges; + they use temporary state, never save credentials to the deployment catalog, + and are skipped in CI. `DEEPSEEK_TEST_MODEL` optionally selects a different + DeepSeek model ID (default: `deepseek-flash`). The same tests can be run on + unfixed and fixed revisions; success is always the expected result. YAML models remain read-only in this page and take precedence on name conflicts. Managed models are appended after YAML models; edits apply to new configuration diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index 0ad044b42..2bb9984de 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -90,7 +90,7 @@ owner-scoped assistant version selection remains enabled. | Router | Endpoints | |--------|-----------| -| **Managed Models** (`/api/managed-models`) | Admin-only GET catalog, PUT create/update with revision checks, POST `/test` streaming tool-call probe. YAML names are reserved; credentials stay private. | +| **Managed Models** (`/api/managed-models`) | Admin-only GET/PUT with revisions; streaming POST `/test` (DeepSeek thinking off). Clients/storage off-loop; YAML names reserved; keys private. | | **Models** (`/api/models`) | `GET /` - list models; `GET /{name}` - model details | | **Features** (`/api/features`) | `GET /` - UI capabilities: hot-reloaded agents, guarded browser, startup MCP tasks, separate batch repository/worker states so history stays readable without a worker, `conversation_references` (whether `read_conversation` is configured, plus the per-run reference cap), and knowledge scope selection | | **Knowledge** (`/api/knowledge/retrieval-catalog`) | Authenticated, allowlist-safe, read-only dataset/document catalog used only by main and custom-agent chat scope selection; knowledge management remains in RAGFlow | diff --git a/backend/app/gateway/routers/managed_models.py b/backend/app/gateway/routers/managed_models.py index 9ab72ead1..5346f65d4 100644 --- a/backend/app/gateway/routers/managed_models.py +++ b/backend/app/gateway/routers/managed_models.py @@ -9,6 +9,7 @@ from pydantic import BaseModel, ConfigDict from app.gateway.deps import require_admin_user from deerflow.config.app_config import get_app_config from deerflow.config.managed_models import ManagedModel, ManagedModelStore +from deerflow.reflection import resolve_class router = APIRouter(prefix="/api/managed-models", tags=["models"]) _ADMIN = "Admin privileges are required to manage shared models." @@ -70,16 +71,24 @@ def _probe_config(body: SaveModelRequest): return profile.runtime_config() +def _build_probe(body: SaveModelRequest): + """Resolve credentials and construct provider clients off the event loop.""" + config = _probe_config(body) + settings = config.model_dump(include={"model", "api_key", "base_url", "api_base"}, exclude_none=True) + # Forced tool selection requires thinking off on DeepSeek. Reuse the resolved + # profile's disable settings without changing the saved chat configuration. + settings.update(config.when_thinking_disabled or {}) + settings.update(timeout=15, max_retries=0, max_tokens=32) + model = resolve_class(config.use)(**settings) + return model.bind_tools([{"type": "function", "function": {"name": "connection_check", "description": "Check the connection", "parameters": {"type": "object", "properties": {}}}}], tool_choice="connection_check") + + @router.post("/test") async def test_model(request: Request, body: SaveModelRequest): """Send a bounded streaming tool-call probe without saving the profile.""" await require_admin_user(request, detail=_ADMIN) try: - config = await asyncio.to_thread(_probe_config, body) - from langchain_openai import ChatOpenAI - - model = ChatOpenAI(model=config.model, base_url=config.base_url, api_key=config.api_key, timeout=15, max_retries=0, max_tokens=32) - probe = model.bind_tools([{"type": "function", "function": {"name": "connection_check", "description": "Check the connection", "parameters": {"type": "object", "properties": {}}}}], tool_choice="connection_check") + probe = await asyncio.to_thread(_build_probe, body) response = None async with asyncio.timeout(20): async for chunk in probe.astream([HumanMessage(content="Call connection_check.")], config={"callbacks": []}): diff --git a/backend/packages/harness/deerflow/config/AGENTS.md b/backend/packages/harness/deerflow/config/AGENTS.md index e507c4fdd..c33c44ab3 100644 --- a/backend/packages/harness/deerflow/config/AGENTS.md +++ b/backend/packages/harness/deerflow/config/AGENTS.md @@ -127,4 +127,6 @@ The file-backed singleton entrypoints additionally merge administrator-managed s models from the encrypted runtime-home catalog. YAML entries win name conflicts; managed changes create new effective snapshots and do not alter an active runtime or an explicitly injected AppConfig. See `../models/AGENTS.md` for storage and reload -boundaries. `AppConfig.from_file()` remains YAML-only. +boundaries. `managed_model_providers.py` derives provider defaults from validated +endpoints; `ManagedModel.runtime_config()` combines them with profile fields. +`AppConfig.from_file()` remains YAML-only. diff --git a/backend/packages/harness/deerflow/config/managed_model_providers.py b/backend/packages/harness/deerflow/config/managed_model_providers.py new file mode 100644 index 000000000..81c6ebfbb --- /dev/null +++ b/backend/packages/harness/deerflow/config/managed_model_providers.py @@ -0,0 +1,29 @@ +"""Provider defaults for administrator-managed OpenAI-compatible endpoints.""" + +from typing import Any +from urllib.parse import urlsplit + + +def _is_official_deepseek_endpoint(base_url: str) -> bool: + endpoint = urlsplit(base_url) + if endpoint.scheme != "https" or endpoint.hostname != "api.deepseek.com": + return False + return endpoint.port in (None, 443) and endpoint.path.rstrip("/") in ("", "/v1") + + +def resolve_managed_model_provider(base_url: str) -> dict[str, Any]: + """Derive settings from a validated endpoint, never from the model name. + + Only the official Chat Completions endpoint opts into DeepSeek semantics; + third-party proxies retain the generic OpenAI-compatible contract. + """ + if _is_official_deepseek_endpoint(base_url): + return { + "use": "deerflow.models.patched_deepseek:PatchedChatDeepSeek", + "api_base": base_url, + "supports_thinking": True, + "supports_reasoning_effort": True, + "when_thinking_enabled": {"extra_body": {"thinking": {"type": "enabled"}}}, + "when_thinking_disabled": {"extra_body": {"thinking": {"type": "disabled"}}}, + } + return {"use": "langchain_openai:ChatOpenAI", "base_url": base_url} diff --git a/backend/packages/harness/deerflow/config/managed_models.py b/backend/packages/harness/deerflow/config/managed_models.py index 6771ea406..acdebf684 100644 --- a/backend/packages/harness/deerflow/config/managed_models.py +++ b/backend/packages/harness/deerflow/config/managed_models.py @@ -19,6 +19,7 @@ from pydantic import BaseModel, ConfigDict, Field, SecretStr, field_validator from deerflow.config.extensions_config import extensions_config_file_lock from deerflow.config.file_signature import get_config_signature +from deerflow.config.managed_model_providers import resolve_managed_model_provider from deerflow.config.model_config import ModelConfig from deerflow.config.runtime_paths import runtime_home @@ -59,16 +60,16 @@ class ManagedModel(BaseModel): return {**self.model_dump(exclude={"api_key"}), "has_api_key": bool(self.api_key and self.api_key.get_secret_value()), "source": "managed"} def runtime_config(self) -> ModelConfig: + api_key = self.api_key.get_secret_value() if self.api_key else None return ModelConfig( name=self.name, display_name=self.display_name or self.name, - use="langchain_openai:ChatOpenAI", model=self.model, - base_url=self.base_url, - api_key=self.api_key.get_secret_value() if self.api_key and self.api_key.get_secret_value() else "not-required", + api_key=api_key or "not-required", supports_vision=self.supports_vision, context_window=self.context_window, max_tokens=self.max_tokens, + **resolve_managed_model_provider(self.base_url), ) diff --git a/backend/packages/harness/deerflow/models/AGENTS.md b/backend/packages/harness/deerflow/models/AGENTS.md index e61ebdf1e..4dfaf95bd 100644 --- a/backend/packages/harness/deerflow/models/AGENTS.md +++ b/backend/packages/harness/deerflow/models/AGENTS.md @@ -46,8 +46,22 @@ returned AppConfig: runtime-scoped and explicitly injected configurations remain authoritative. `_managed_model_names` is private source metadata, not provider kwargs. Direct `AppConfig.from_file()` continues to read only operator configuration. -The MVP uses only the registered `langchain_openai:ChatOpenAI` adapter. Full updates -require the current revision; omission means create, an omitted API key retains the -saved key and an empty string clears it. Never serialize SecretStr masking as a saved -key. Read APIs return `has_api_key`, never a credential. Tests live in -`tests/test_managed_models.py` and `tests/blocking_io/test_managed_models.py`. +`config/managed_model_providers.py` owns endpoint detection and provider defaults. +Managed profiles normally use `langchain_openai:ChatOpenAI`. Official HTTPS +`api.deepseek.com` endpoints (default port, root or `/v1` path) instead resolve to +`PatchedChatDeepSeek` with explicit thinking on/off settings and reasoning-effort +support. Resolve from the parsed endpoint, never a model-name substring; proxies, +lookalike hosts and other paths retain the generic contract. This is derived runtime +configuration: no catalog migration or new API fields. The native `api_base` field +preserves the administrator's endpoint; the adapter preserves `reasoning_content` +and sends `max_tokens` rather than OpenAI's `max_completion_tokens`. + +Full updates require the current revision; omission means create, an omitted API +key retains the saved key and an empty string clears it. Never serialize SecretStr +masking as a saved key. Read APIs return `has_api_key`, never a credential. +The Gateway probe constructs the resolved class off-loop and applies the profile's +`when_thinking_disabled` settings to its bounded forced-tool request. It does not +repeat provider selection or persist the probe override. +Tests: `test_managed_models.py`, `test_managed_deepseek.py` (real SDK serialization +with an HTTP double), opt-in `test_managed_deepseek_live.py`, and +`tests/blocking_io/test_managed_models.py`. diff --git a/backend/tests/AGENTS.md b/backend/tests/AGENTS.md index a95c58721..93f16d0db 100644 --- a/backend/tests/AGENTS.md +++ b/backend/tests/AGENTS.md @@ -57,3 +57,13 @@ missing token fence; always drain paused tasks and restore session patches. Use explicit synchronization such as `threading.Event` rather than sleep-based timing thresholds for worker lifecycle assertions. Every test must release blocked workers and restore any process-global monkeypatches so teardown cannot leak threads or state into later tests. Stress/soak testing, AnyIO worker instrumentation, Uvicorn multi-process behavior, and broad production executor redesign are separate concerns and should not be folded into these deterministic regressions. + +## Managed DeepSeek compatibility + +`test_managed_deepseek.py` exercises real SDK request serialization and SSE parsing +with an HTTP double; do not replace the provider classes with successful stubs. +`test_managed_deepseek_live.py` uses the same production probe/model configuration +against DeepSeek only with `DEER_FLOW_RUN_LIVE_TESTS=1` and +`DEEPSEEK_TEST_API_KEY`, never in CI. Keep credentials and provider payloads out of +committed evidence. A passing connectivity probe does not establish full agent +compatibility; distinguish protocol assertions from observed live behavior. diff --git a/backend/tests/blocking_io/test_managed_models.py b/backend/tests/blocking_io/test_managed_models.py index da7b2c949..bbade961c 100644 --- a/backend/tests/blocking_io/test_managed_models.py +++ b/backend/tests/blocking_io/test_managed_models.py @@ -1,12 +1,17 @@ """Managed model encryption, persistence and config merging stay off the loop.""" +import asyncio +import json from types import SimpleNamespace +import httpx import pytest +from langchain_openai import ChatOpenAI from app.gateway.routers import managed_models as router from deerflow.config.app_config import AppConfig -from deerflow.config.managed_models import ManagedModel +from deerflow.config.managed_models import ManagedModel, ManagedModelStore +from deerflow.models.patched_deepseek import PatchedChatDeepSeek @pytest.mark.asyncio @@ -21,3 +26,49 @@ async def test_admin_catalog_round_trip_offloads_storage(tmp_path, monkeypatch): catalog = await router.list_managed_models(request) assert catalog["models"][0]["name"] == "test" assert "secret" not in str(catalog) + + +@pytest.mark.asyncio +async def test_deepseek_probe_offloads_credentials_and_client_construction(tmp_path, monkeypatch): + monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path)) + monkeypatch.setenv("LANGSMITH_TRACING", "false") + store = await asyncio.to_thread(ManagedModelStore) + profile = ManagedModel(name="flash", model="deepseek-flash", base_url="https://api.deepseek.com", api_key="test-secret") + saved = await asyncio.to_thread(store.save, profile, expected_revision=None) + original_catalog = await asyncio.to_thread(store.path.read_bytes) + constructed = [] + + def observe_constructor(original): + def initialize(self, *args, **kwargs): + with pytest.raises(RuntimeError, match="no running event loop"): + asyncio.get_running_loop() + original(self, *args, **kwargs) + constructed.append(type(self).__name__) + + return initialize + + monkeypatch.setattr(ChatOpenAI, "__init__", observe_constructor(ChatOpenAI.__init__)) + monkeypatch.setattr(PatchedChatDeepSeek, "__init__", observe_constructor(PatchedChatDeepSeek.__init__)) + + async def send(client, request, **kwargs): + assert request.headers["authorization"] == "Bearer test-secret" + body = json.loads(request.content) + assert body["thinking"] == {"type": "disabled"} + chunk = { + "id": "chat-test", + "object": "chat.completion.chunk", + "created": 1, + "model": "deepseek-flash", + "choices": [ + {"index": 0, "delta": {"role": "assistant", "content": "", "tool_calls": [{"index": 0, "id": "call-test", "type": "function", "function": {"name": "connection_check", "arguments": "{}"}}]}, "finish_reason": "tool_calls"} + ], + } + return httpx.Response(200, request=request, headers={"content-type": "text/event-stream"}, content=f"data: {json.dumps(chunk)}\n\ndata: [DONE]\n\n".encode()) + + monkeypatch.setattr(httpx.AsyncClient, "send", send) + request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(system_role="admin"))) + draft = ManagedModel(name="flash", model="deepseek-flash", base_url="https://api.deepseek.com") + result = await router.test_model(request, router.SaveModelRequest(config=draft, expected_revision=saved.revision)) + assert result == {"ok": True, "message": "success"} + assert constructed == ["PatchedChatDeepSeek"] + assert await asyncio.to_thread(store.path.read_bytes) == original_catalog diff --git a/backend/tests/test_managed_deepseek.py b/backend/tests/test_managed_deepseek.py new file mode 100644 index 000000000..d787da497 --- /dev/null +++ b/backend/tests/test_managed_deepseek.py @@ -0,0 +1,281 @@ +"""DeepSeek managed-model regressions through real LangChain/OpenAI serialization. + +Only the HTTP boundary is replaced. The fake provider enforces the documented +thinking/tool-choice restriction; the real SDK builds and parses every request. +No API credentials or network access are needed. +""" + +import asyncio +import json +from types import SimpleNamespace + +import httpx +import pytest +from fastapi import HTTPException +from langchain_core.messages import HumanMessage, ToolMessage + +from app.gateway.routers import managed_models as router +from deerflow.config.app_config import AppConfig +from deerflow.config.managed_models import ManagedModel, ManagedModelStore, merge_managed_models +from deerflow.models.factory import create_chat_model + +_KEY = "diagnostic-only-not-a-real-key" +_ADMIN = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(system_role="admin"))) +_TOOL = {"type": "function", "function": {"name": "connection_check", "description": "Check the connection", "parameters": {"type": "object", "properties": {}}}} + + +def _profile(**overrides): + return ManagedModel(**{"name": "flash", "model": "deepseek-flash", "base_url": "https://api.deepseek.com", "api_key": _KEY, **overrides}) + + +def _base_config(): + return AppConfig.model_validate({"sandbox": {"use": "test"}, "models": []}) + + +@pytest.fixture +def store(tmp_path, monkeypatch): + monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path)) + monkeypatch.setenv("LANGSMITH_TRACING", "false") + monkeypatch.setenv("LANGCHAIN_TRACING_V2", "false") + return ManagedModelStore() + + +@pytest.fixture +def provider(monkeypatch): + """Inspect outbound HTTP and feed split SSE frames into the real SDK.""" + state = SimpleNamespace(requests=[], mode="tool", status=200, exception=None) + + async def send(client, request, **kwargs): + body = json.loads(request.content) + state.requests.append((request, body)) + if state.exception is not None: + raise state.exception + if state.status != 200: + return httpx.Response(state.status, request=request, json={"error": {"message": f"Provider failure containing {_KEY}", "type": "invalid_request_error"}}) + if request.headers.get("authorization") != f"Bearer {_KEY}": + return httpx.Response(401, request=request, json={"error": {"message": "Authentication failed", "type": "authentication_error"}}) + thinking = body.get("thinking", {}).get("type", "enabled") == "enabled" + if request.url.host == "api.deepseek.com" and thinking and body.get("tool_choice") not in (None, "auto", "none"): + return httpx.Response(400, request=request, json={"error": {"message": "Thinking mode does not support this tool_choice", "type": "invalid_request_error"}}) + deltas = [{"role": "assistant", "content": ""}] + if thinking: + deltas.extend([{"reasoning_content": "Checking "}, {"reasoning_content": "connection."}]) + finish = "stop" + if body["messages"][-1]["role"] == "tool" or state.mode == "text": + deltas.append({"content": "OK"}) + elif state.mode != "empty": + name = "wrong_tool" if state.mode == "wrong_tool" else "connection_check" + args = '{"unexpected":true}' if state.mode == "wrong_args" else "not-json" if state.mode == "malformed" else "{" + deltas.append({"tool_calls": [{"index": 0, "id": "call-test", "type": "function", "function": {"name": name, "arguments": args}}]}) + if state.mode != "wrong_args": + deltas.append({"tool_calls": [{"index": 0, "function": {"arguments": "}"}}]}) + finish = "tool_calls" + chunks = [{"id": "chat-test", "object": "chat.completion.chunk", "created": 1, "model": body["model"], "choices": [{"index": 0, "delta": delta, "finish_reason": None}]} for delta in deltas] + chunks.append({"id": "chat-test", "object": "chat.completion.chunk", "created": 1, "model": body["model"], "choices": [{"index": 0, "delta": {}, "finish_reason": finish}]}) + content = "".join(f"data: {json.dumps(chunk)}\n\n" for chunk in chunks) + "data: [DONE]\n\n" + return httpx.Response(200, request=request, headers={"content-type": "text/event-stream"}, content=content.encode()) + + monkeypatch.setattr(httpx.AsyncClient, "send", send) + return state + + +@pytest.mark.asyncio +@pytest.mark.parametrize("base_url", ["https://api.deepseek.com", "https://api.deepseek.com/v1/", "https://API.DEEPSEEK.COM:443"]) +@pytest.mark.parametrize("model_id", ["deepseek-flash", "deepseek-v4-flash", "deepseek-v4-pro"]) +async def test_probe_uses_non_thinking_bounded_streaming_tool_request(store, provider, base_url, model_id): + draft = _profile(base_url=base_url, model=model_id, max_tokens=8192) + result = await router.test_model(_ADMIN, router.SaveModelRequest(config=draft)) + assert result == {"ok": True, "message": "success"} + assert len(provider.requests) == 1 + request, body = provider.requests[0] + assert request.url == httpx.URL(base_url.rstrip("/") + "/chat/completions") + assert body["model"] == model_id + assert body["stream"] is True + assert body["thinking"] == {"type": "disabled"} + assert body["max_tokens"] == 32 + assert "max_completion_tokens" not in body + assert body["tool_choice"] == {"type": "function", "function": {"name": "connection_check"}} + assert request.extensions["timeout"]["read"] == 15 + assert draft.max_tokens == 8192 + assert "supports_thinking" not in draft.model_dump() + assert not store.path.exists() + assert not store.key_path.exists() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", ["text", "empty", "wrong_tool", "wrong_args", "malformed"]) +async def test_probe_requires_the_expected_tool_and_empty_arguments(store, provider, mode): + provider.mode = mode + result = await router.test_model(_ADMIN, router.SaveModelRequest(config=_profile())) + assert result == {"ok": False, "message": "tool_call_missing"} + assert len(provider.requests) == 1 + assert not store.path.exists() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", [400, 401, 403, 429, 500]) +async def test_probe_does_not_retry_or_expose_provider_errors(store, provider, status): + provider.status = status + result = await router.test_model(_ADMIN, router.SaveModelRequest(config=_profile())) + assert result == {"ok": False, "message": "connection_failed"} + assert _KEY not in json.dumps(result) + assert len(provider.requests) == 1 + assert not store.path.exists() + + +@pytest.mark.asyncio +async def test_probe_timeout_and_cancellation_are_distinct(store, provider): + provider.exception = httpx.ReadTimeout("private provider detail") + assert await router.test_model(_ADMIN, router.SaveModelRequest(config=_profile())) == {"ok": False, "message": "connection_failed"} + provider.exception = asyncio.CancelledError() + with pytest.raises(asyncio.CancelledError): + await router.test_model(_ADMIN, router.SaveModelRequest(config=_profile())) + assert not store.path.exists() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("key_mode", ["retained", "replacement", "cleared"]) +async def test_probe_uses_draft_or_saved_key_without_persisting(store, provider, key_mode): + saved = store.save(_profile(api_key="old-key" if key_mode == "replacement" else _KEY), expected_revision=None) + original_catalog = store.path.read_bytes() + draft = _profile(api_key={"retained": None, "replacement": _KEY, "cleared": ""}[key_mode]) + result = await router.test_model(_ADMIN, router.SaveModelRequest(config=draft, expected_revision=saved.revision)) + assert result == ({"ok": False, "message": "connection_failed"} if key_mode == "cleared" else {"ok": True, "message": "success"}) + assert len(provider.requests) == 1 + expected_key = "not-required" if key_mode == "cleared" else _KEY + assert provider.requests[0][0].headers["authorization"] == f"Bearer {expected_key}" + assert store.path.read_bytes() == original_catalog + assert store.list()[0].revision == saved.revision + assert (draft.api_key.get_secret_value() if draft.api_key is not None else None) == {"retained": None, "replacement": _KEY, "cleared": ""}[key_mode] + + +@pytest.mark.asyncio +async def test_stale_revision_is_rejected_before_provider_contact(store, provider): + saved = store.save(_profile(), expected_revision=None) + store.save(_profile(), expected_revision=saved.revision) + with pytest.raises(HTTPException) as exc: + await router.test_model(_ADMIN, router.SaveModelRequest(config=_profile(api_key=None), expected_revision=saved.revision)) + assert exc.value.status_code == 409 + assert not provider.requests + + +@pytest.mark.asyncio +@pytest.mark.parametrize("state", [SimpleNamespace(user=SimpleNamespace(system_role="user")), SimpleNamespace(user=SimpleNamespace(system_role="admin"), auth_source="pat")]) +async def test_probe_authorization_precedes_provider_contact(store, provider, state): + with pytest.raises(HTTPException) as exc: + await router.test_model(SimpleNamespace(state=state), router.SaveModelRequest(config=_profile())) + assert exc.value.status_code == 403 + assert not provider.requests + assert not store.path.exists() + + +async def _collect(model, messages): + result = None + async for chunk in model.astream(messages, config={"callbacks": []}): + result = chunk if result is None else result + chunk + assert result is not None + return result + + +@pytest.mark.asyncio +@pytest.mark.parametrize("thinking", [False, True]) +async def test_saved_model_streaming_replays_reasoning_and_honors_token_limit(store, provider, thinking): + saved = store.save(_profile(max_tokens=1536), expected_revision=None) + original_catalog = store.path.read_bytes() + base = _base_config() + config = merge_managed_models(base) + model_config = config.get_model_config(saved.name) + model = create_chat_model(saved.name, thinking_enabled=thinking, reasoning_effort="high", app_config=config, attach_tracing=False) + bound = model.bind_tools([_TOOL]) + history = [HumanMessage(content="Call connection_check, then say OK.")] + first = await _collect(bound, history) + assert first.tool_calls[0]["name"] == "connection_check" + expected_reasoning = "Checking connection." if thinking else None + assert first.additional_kwargs.get("reasoning_content") == expected_reasoning + history.extend([first, ToolMessage(content="healthy", tool_call_id=first.tool_calls[0]["id"])]) + final = await _collect(bound, history) + assert final.content == "OK" + assert len(provider.requests) == 2 + for request, body in provider.requests: + assert str(request.url) == "https://api.deepseek.com/chat/completions" + assert body["thinking"] == {"type": "enabled" if thinking else "disabled"} + assert body["max_tokens"] == 1536 + assert "max_completion_tokens" not in body + if thinking: + assert body["reasoning_effort"] == "high" + assistant = provider.requests[1][1]["messages"][1] + assert assistant["content"] == "" + assert assistant.get("reasoning_content") == expected_reasoning + assert store.path.read_bytes() == original_catalog + assert base.models == [] + assert model_config.supports_thinking is True + assert model_config.supports_reasoning_effort is True + + +@pytest.mark.parametrize( + "base_url", + [ + "https://api.deepseek.com.example.org/v1", + "https://other.api.deepseek.com/v1", + "https://api.deepseek.com/anthropic", + "https://api.deepseek.com/proxy/v1", + "https://api.deepseek.com:8443/v1", + "https://openrouter.ai/api/v1", + "http://localhost:8000/v1", + ], +) +def test_other_endpoints_do_not_gain_deepseek_specific_parameters(store, base_url): + saved = store.save(_profile(base_url=base_url), expected_revision=None) + config = merge_managed_models(_base_config()) + model = create_chat_model(saved.name, app_config=config, attach_tracing=False) + payload = model._get_request_payload([HumanMessage(content="Hello")]) + assert "thinking" not in payload + assert not payload.get("extra_body") + assert config.get_model_config(saved.name).supports_thinking is False + assert str(model.openai_api_base).rstrip("/") == base_url.rstrip("/") + + +@pytest.mark.asyncio +async def test_generic_openai_probe_keeps_existing_request_contract(store, provider): + result = await router.test_model(_ADMIN, router.SaveModelRequest(config=_profile(base_url="https://example.com/v1"))) + assert result == {"ok": True, "message": "success"} + body = provider.requests[0][1] + assert "thinking" not in body + assert body["max_completion_tokens"] == 32 + assert body["tool_choice"]["function"]["name"] == "connection_check" + + +@pytest.mark.asyncio +async def test_new_draft_without_key_does_not_use_environment_credentials(store, provider, monkeypatch): + monkeypatch.setenv("DEEPSEEK_API_KEY", _KEY) + monkeypatch.setenv("OPENAI_API_KEY", _KEY) + result = await router.test_model(_ADMIN, router.SaveModelRequest(config=_profile(api_key=None))) + assert result == {"ok": False, "message": "connection_failed"} + assert provider.requests[0][0].headers["authorization"] == "Bearer not-required" + assert not store.path.exists() + + +@pytest.mark.asyncio +async def test_stream_failure_after_valid_tool_chunks_is_not_success(store, provider, monkeypatch): + send = httpx.AsyncClient.send + + class InterruptedStream(httpx.AsyncByteStream): + async def __aiter__(self): + yield self.content + raise httpx.ReadError("private upstream stream failure") + + def __init__(self, content): + self.content = content + + async def interrupted(client, request, **kwargs): + response = await send(client, request, **kwargs) + if response.status_code == 200: + content = response.content.replace(b"data: [DONE]\n\n", b"") + return httpx.Response(200, request=request, headers=response.headers, stream=InterruptedStream(content)) + return response + + monkeypatch.setattr(httpx.AsyncClient, "send", interrupted) + result = await router.test_model(_ADMIN, router.SaveModelRequest(config=_profile())) + assert result == {"ok": False, "message": "connection_failed"} + assert len(provider.requests) == 1 + assert not store.path.exists() diff --git a/backend/tests/test_managed_deepseek_live.py b/backend/tests/test_managed_deepseek_live.py new file mode 100644 index 000000000..fe061131f --- /dev/null +++ b/backend/tests/test_managed_deepseek_live.py @@ -0,0 +1,75 @@ +"""Opt-in verification of the same managed DeepSeek path before/after a fix. + +From backend/ (test credentials are read only from the process environment): + DEER_FLOW_RUN_LIVE_TESTS=1 uv run --no-sync pytest \ + tests/test_managed_deepseek_live.py -q -s + +Set DEEPSEEK_TEST_API_KEY separately; optional DEEPSEEK_TEST_MODEL defaults to +"deepseek-flash". This sends a few short requests to api.deepseek.com and may +incur charges. It never starts an agent or modifies the real managed catalog. +""" + +import os +from types import SimpleNamespace + +import pytest +from langchain_core.messages import HumanMessage, ToolMessage + +from app.gateway.routers import managed_models as router +from deerflow.config.app_config import AppConfig +from deerflow.config.managed_models import ManagedModel +from deerflow.models.factory import create_chat_model + +pytestmark = [ + pytest.mark.live, + pytest.mark.skipif( + os.getenv("DEER_FLOW_RUN_LIVE_TESTS") != "1" or not os.getenv("DEEPSEEK_TEST_API_KEY") or bool(os.getenv("CI")), + reason="Requires explicit live opt-in and DEEPSEEK_TEST_API_KEY; never runs in CI", + ), +] + + +@pytest.fixture +def profile(tmp_path, monkeypatch): + monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path)) + monkeypatch.setenv("LANGSMITH_TRACING", "false") + monkeypatch.setenv("LANGCHAIN_TRACING_V2", "false") + return ManagedModel(name="deepseek-live", model=os.getenv("DEEPSEEK_TEST_MODEL", "deepseek-flash"), base_url="https://api.deepseek.com", api_key=os.environ["DEEPSEEK_TEST_API_KEY"], max_tokens=512) + + +@pytest.mark.asyncio +async def test_managed_deepseek_connection_probe_live(profile, tmp_path): + request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(system_role="admin"))) + result = await router.test_model(request, router.SaveModelRequest(config=profile)) + assert result == {"ok": True, "message": "success"} + assert not (tmp_path / "managed-models").exists() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("thinking", [False, True]) +async def test_managed_deepseek_tool_round_trip_live(profile, thinking, tmp_path): + config = AppConfig.model_validate({"sandbox": {"use": "test"}, "models": [profile.runtime_config().model_dump(exclude_none=True)]}) + model = create_chat_model(profile.name, thinking_enabled=thinking, app_config=config, attach_tracing=False, timeout=30, max_retries=0) + tool = {"type": "function", "function": {"name": "connection_check", "description": "Check the connection", "parameters": {"type": "object", "properties": {}}}} + bound = model.bind_tools([tool]) + history = [HumanMessage(content="Call connection_check exactly once. After its result, reply only OK.")] + first = None + async for chunk in bound.astream(history, config={"callbacks": []}): + first = chunk if first is None else first + chunk + assert first is not None and first.tool_calls + history.append(first) + history.extend(ToolMessage(content="Connection is healthy.", tool_call_id=call["id"]) for call in first.tool_calls) + payload = model._get_request_payload(history, **bound.kwargs) + assert payload["max_tokens"] == 512 + assert "max_completion_tokens" not in payload + assert payload["extra_body"]["thinking"]["type"] == ("enabled" if thinking else "disabled") + assistant = payload["messages"][1] + assert assistant["content"] is not None + if thinking: + # Reasoning can be empty. Verify faithful replay, not model verbosity. + assert assistant["reasoning_content"] == first.additional_kwargs.get("reasoning_content", "") + final = None + async for chunk in bound.astream(history, config={"callbacks": []}): + final = chunk if final is None else final + chunk + assert final is not None and final.content and not final.tool_calls + assert not (tmp_path / "managed-models").exists()