From 51d59a1cdff92626bd9f03f1b2254f34a824ca3e Mon Sep 17 00:00:00 2001 From: lilei <93921041+xihongshichaojidan8@users.noreply.github.com> Date: Tue, 27 Oct 2026 10:11:56 +0800 Subject: [PATCH] fix(safety): handle native Responses content-filter termination (#6965) * docs: describe native Responses safety stops * fix(safety): handle native Responses content-filter termination * test(safety): cover native Responses stops and replay --- README.md | 2 + .../deerflow/agents/middlewares/AGENTS.md | 2 +- .../safety_finish_reason_middleware.py | 9 +- .../safety_termination_detectors.py | 23 ++- .../test_safety_finish_reason_middleware.py | 4 +- backend/tests/test_safety_responses.py | 187 ++++++++++++++++++ .../test_safety_termination_detectors.py | 6 +- 7 files changed, 223 insertions(+), 10 deletions(-) create mode 100644 backend/tests/test_safety_responses.py diff --git a/README.md b/README.md index c2ec4dcc99..8bd7b5371c 100644 --- a/README.md +++ b/README.md @@ -3204,6 +3204,8 @@ when retries are exhausted. Turn Live off and on to start a fresh retry budget. **Strict Tool-Call Recovery**: When a provider or middleware interrupts a tool-call loop, DeerFlow now strips provider-level raw tool-call metadata on forced-stop assistant messages and injects placeholder tool results for dangling calls before the next model invocation. This keeps OpenAI-compatible reasoning models that strictly validate `tool_call_id` sequences from failing with malformed history errors. +**Provider Safety Stops**: With `safety_finish_reason.enabled`, native OpenAI Responses `status=incomplete` with `incomplete_details.reason=content_filter` suppresses tool calls, including malformed calls, and removes their native frames from subsequent requests. Empty filtered responses receive an explanation; visible partial answers without tool calls are preserved. + **Visible Tool-Run Completion**: For interactive turns, DeerFlow retries an empty post-tool final response once, then surfaces a visible error instead of reporting a silent successful run. **Run History**: Image-only user input is recorded once per run, even when the agent makes multiple model calls. Its media content remains in history without requiring a text summary. diff --git a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md index 3d11e8be96..05e63052f1 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md +++ b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md @@ -164,7 +164,7 @@ Before changing a later authorization phase, read the [authorization RFC](../../ 35. **Configured extension middlewares** - `extensions.middlewares` in `config.yaml` or `extensions_config.json` optionally accepts `module.path:ClassName` strings or `{class, kwargs}` objects. `deerflow.reflection.resolve_class` loads `AgentMiddleware` classes; import, class, and constructor errors fail agent creation. `kwargs` must be JSON-compatible; YAML dates/timestamps become ISO strings. Order: built-ins/custom and loop/token guards → extensions → terminal-response/safety/clarification tail. Subagents share the list before their safety tail; separate lead/subagent lists are unsupported. Trusted operator config only: paths instantiate arbitrary code. Gateway skill/MCP toggles preserve it in raw JSON; adding an API write path requires explicit trust-boundary review. 36. **TerminalResponseMiddleware** - When a provider returns an empty terminal `AIMessage` after tool execution, injects a hidden recovery prompt and retries the model once; a second empty response is replaced in checkpoint state by a visible error fallback marked for the run worker, so the run finishes as an error instead of a silent success 37. **ModelLengthFinishReasonMiddleware** - Responses: same-container `status=incomplete` + `incomplete_details.reason=max_output_tokens`. Suppress calls via shared cloning after clearing invalid calls; keep non-call blocks/metadata. Stamp cap markers; add notice. Also registered on subagent chains (before Safety), where `consume_stop_reason` surfaces `model_length_capped` to the executor. -38. **SafetyFinishReasonMiddleware** - *(optional, if `safety_finish_reason.enabled`)* Suppresses tool execution when the provider safety-terminated the response (e.g. `finish_reason=content_filter`); registered after terminal-response/custom/configured middlewares so LangChain's reverse-order `after_model` dispatch runs it first +38. **SafetyFinishReasonMiddleware** - `safety_finish_reason.enabled`: suppresses valid/invalid safety-stopped calls. Responses: `status=incomplete` + `incomplete_details.reason=content_filter`. Clear invalid calls before shared cloning. Place after terminal/custom/configured hooks; `after_model` runs in reverse. 39. **ClarificationMiddleware** - Intercepts `ask_clarification`, writes a readable `ToolMessage.content` fallback plus a structured `ToolMessage.artifact.human_input` payload, and interrupts via `Command(goto=END)` (must be last). `after_model` drops same-turn sibling tool calls so they cannot run before the user answers; a malformed `ask_clarification` parked on `invalid_tool_calls` is the same stop signal. `disable_clarification` runs keep the siblings. Payloads are versioned — legacy `free_text`/`choice_with_other` stay `version: 1`; the v2 `form` mode (from `fields`) is `version: 2` so older frontends reject it and fall back to plain text. Field normalization is deterministic and lives in the middleware (it short-circuits before tool execution, so tool-arg typing gives no runtime validation), and it is atomic: any structurally broken entry — non-dict, bad/duplicate name, a name colliding with a JS `Object.prototype` member (`__proto__`/`constructor`), or exceeding the caps (16 fields / 24 options per field / 200 chars per text / `MAX_FORM_SERIALIZED_BYTES` = 16KB UTF-8, the per-item caps alone admitting forms whose IM text fallback overruns channel limits) — degrades the whole form to the legacy option/free-text modes, so a card never renders "complete" while missing a field. Benign issues degrade locally (unknown types — incl. unhashable JSON like `type: []` — and option-less selects become `text`); options are trimmed/deduped with blanks dropped (form- and top-level) since the frontend rejects blank labels. XML-to-dict option payloads are recursively flattened from dict/list containers in source order, scalar leaves kept, residual XML tags stripped before that trimming. Checkboxes are booleans defaulting to "no"; `required` on one means consent semantics. The response protocol is unchanged (v1 `text`/`option`): form cards submit a text summary as `response_kind: "text"`, so journal persistence needs no new allowlist entries. `RunJournal` reconciles visible, unpersisted `ToolMessage`s for current-run lead-agent calls at the next lead-agent `on_chat_model_start` after consumption, or at successful root `on_chain_end`. This preserves any middleware short-circuit, including blocked writes, across later model errors and checkpoint compaction (#4666). `_remember_current_run_tool_calls` excludes subagents; their results stay in `subagent.step`. Human Input Card replies are `hide_from_ui` `HumanMessage`s with `additional_kwargs.human_input_response`; `RunJournal` persists only allowlisted hidden sources (currently `ask_clarification`) as `llm.human.input`. Inline `skill_references` accepts 1–16 names; empty lists retain slash/plain flow. diff --git a/backend/packages/harness/deerflow/agents/middlewares/safety_finish_reason_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/safety_finish_reason_middleware.py index 59ea548915..65782bd977 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/safety_finish_reason_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/safety_finish_reason_middleware.py @@ -190,7 +190,7 @@ class SafetyFinishReasonMiddleware(AgentMiddleware[AgentState]): message: AIMessage, termination: SafetyTermination, ) -> AIMessage: - tool_calls = message.tool_calls or [] + tool_calls = [*(message.tool_calls or []), *(message.invalid_tool_calls or [])] suppressed_names = [tc.get("name") or "unknown" for tc in tool_calls] template = _USER_FACING_MESSAGE if tool_calls else _USER_FACING_EMPTY_MESSAGE explanation = template.format( @@ -206,7 +206,10 @@ class SafetyFinishReasonMiddleware(AgentMiddleware[AgentState]): # It only rewrites finish_reason when the old value was "tool_calls", # which is not our case — content_filter / refusal / SAFETY stay put # so downstream SSE / converters keep seeing the real provider reason. - cleared = clone_ai_message_with_tool_calls(message, [], content=new_content) + # Invalid calls must also be suppressed: the shared helper otherwise + # retains their native content blocks for dangling-call recovery. + source = message.model_copy(update={"invalid_tool_calls": []}) + cleared = clone_ai_message_with_tool_calls(source, [], content=new_content) # Re-clone additional_kwargs so we don't accidentally mutate the # dict returned by clone_ai_message_with_tool_calls (which already @@ -373,7 +376,7 @@ class SafetyFinishReasonMiddleware(AgentMiddleware[AgentState]): # 'assistant' must not be empty", #4393), which poisons the whole # thread until a new chat is started. Backfill an explanation so # the persisted message is non-empty. - tool_calls = list(last.tool_calls or []) + tool_calls = [*(last.tool_calls or []), *(last.invalid_tool_calls or [])] # Keep local falsey-content normalization as a defensive guard; the # shared helper also handles None from validation-skipping rewrites. # The trailing strip() makes whitespace-only content blank as well. diff --git a/backend/packages/harness/deerflow/agents/middlewares/safety_termination_detectors.py b/backend/packages/harness/deerflow/agents/middlewares/safety_termination_detectors.py index b98e9f4d7e..d11efaf4f2 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/safety_termination_detectors.py +++ b/backend/packages/harness/deerflow/agents/middlewares/safety_termination_detectors.py @@ -2,7 +2,7 @@ Different LLM providers signal "I stopped this response for safety reasons" through different fields with different values. This module defines a small -strategy interface and three built-in detectors that cover the major +strategy interface and built-in detectors that cover the major providers DeerFlow supports today. New providers (Wenxin, Hunyuan, Bedrock adapters, in-house gateways, ...) can be added by implementing ``SafetyTerminationDetector`` and wiring it through @@ -119,6 +119,25 @@ class OpenAICompatibleContentFilterDetector: ) +class OpenAIResponsesContentFilterDetector: + """Native Responses safety termination, distinct from token exhaustion.""" + + name = "openai_responses_content_filter" + + def detect(self, message: AIMessage) -> SafetyTermination | None: + for metadata in (message.response_metadata, message.additional_kwargs): + if not isinstance(metadata, dict) or metadata.get("status") != "incomplete": + continue + details = metadata.get("incomplete_details") + if isinstance(details, dict) and details.get("reason") == "content_filter": + return SafetyTermination( + detector=self.name, + reason_field="incomplete_details.reason", + reason_value="content_filter", + ) + return None + + class AnthropicRefusalDetector: """Anthropic ``stop_reason == "refusal"`` signal. @@ -222,6 +241,7 @@ def default_detectors() -> list[SafetyTerminationDetector]: """Built-in detector set used when no custom detectors are configured.""" return [ OpenAICompatibleContentFilterDetector(), + OpenAIResponsesContentFilterDetector(), AnthropicRefusalDetector(), GeminiSafetyDetector(), ] @@ -231,6 +251,7 @@ __all__ = [ "AnthropicRefusalDetector", "GeminiSafetyDetector", "OpenAICompatibleContentFilterDetector", + "OpenAIResponsesContentFilterDetector", "SafetyTermination", "SafetyTerminationDetector", "default_detectors", diff --git a/backend/tests/test_safety_finish_reason_middleware.py b/backend/tests/test_safety_finish_reason_middleware.py index 622b372af7..eb36e72643 100644 --- a/backend/tests/test_safety_finish_reason_middleware.py +++ b/backend/tests/test_safety_finish_reason_middleware.py @@ -525,9 +525,9 @@ class TestDetectorWiring: class TestFromConfig: def test_default_config_uses_builtin_detectors(self): mw = SafetyFinishReasonMiddleware.from_config(SafetyFinishReasonConfig()) - assert len(mw._detectors) == 3 + assert len(mw._detectors) == 4 names = {d.name for d in mw._detectors} - assert names == {"openai_compatible_content_filter", "anthropic_refusal", "gemini_safety"} + assert names == {"openai_compatible_content_filter", "openai_responses_content_filter", "anthropic_refusal", "gemini_safety"} def test_custom_detectors_loaded_via_reflection(self): cfg = SafetyFinishReasonConfig( diff --git a/backend/tests/test_safety_responses.py b/backend/tests/test_safety_responses.py new file mode 100644 index 0000000000..47159c9643 --- /dev/null +++ b/backend/tests/test_safety_responses.py @@ -0,0 +1,187 @@ +"""Offline Responses SDK regressions for provider safety termination. + +Synthetic HTTP/SSE packets exercise the real adapter; no provider is contacted. +""" + +import json +from types import SimpleNamespace +from unittest.mock import MagicMock + +import httpx +import pytest +from langchain.agents import create_agent +from langchain_core.messages import AIMessage, HumanMessage +from langchain_core.tools import tool +from langchain_openai import ChatOpenAI + +from deerflow.agents.middlewares.safety_finish_reason_middleware import SafetyFinishReasonMiddleware +from deerflow.agents.middlewares.safety_termination_detectors import default_detectors + + +def _packet(reason="content_filter", arguments="{}", final=False): + output = ( + [{"type": "message", "id": "msg_final", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": "Done.", "annotations": []}]}] + if final + else [{"type": "function_call", "id": "fc_probe", "call_id": "call_probe", "name": "record_probe", "arguments": arguments, "status": "incomplete" if reason else "completed"}] + ) + return { + "id": "resp_final" if final else "resp_probe", + "object": "response", + "created_at": 1700000000, + "model": "gpt-4.1-mini", + "status": "incomplete" if reason and not final else "completed", + "incomplete_details": {"reason": reason} if reason and not final else None, + "output": output, + "parallel_tool_calls": True, + "tools": [], + "tool_choice": "auto", + } + + +@pytest.fixture +def sdk_model(): + clients = [] + + def make(*, streaming=False, reason="content_filter", arguments="{}"): + requests = [] + + def transport(request): + requests.append(json.loads(request.content)) + response = _packet(reason, arguments, final=len(requests) > 1) + if not streaming: + return httpx.Response(200, json=response) + created = {**response, "output": [], "status": "in_progress", "incomplete_details": None} + events = [{"type": "response.created", "response": created, "sequence_number": 0}] + if len(requests) == 1: + events.append({"type": "response.output_item.added", "output_index": 0, "item": response["output"][0], "sequence_number": 1}) + else: + events.append({"type": "response.output_text.delta", "output_index": 0, "content_index": 0, "item_id": "msg_final", "delta": "Done.", "sequence_number": 1}) + events.append({"type": "response.incomplete" if response["status"] == "incomplete" else "response.completed", "response": response, "sequence_number": 2}) + body = "".join(f"event: {event['type']}\ndata: {json.dumps(event)}\n\n" for event in events) + return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=body.encode()) + + client = httpx.Client(transport=httpx.MockTransport(transport)) + async_client = httpx.AsyncClient(transport=httpx.MockTransport(transport)) + clients.append((client, async_client)) + return ChatOpenAI(model="gpt-4.1-mini", api_key="offline-placeholder", base_url="https://offline.invalid/v1", use_responses_api=True, http_client=client, http_async_client=async_client, max_retries=0, streaming=streaming), requests + + yield make + import asyncio + + for client, async_client in clients: + client.close() + asyncio.run(async_client.aclose()) + + +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize("arguments", ["{}", '{"unfinished":']) +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.asyncio +async def test_sdk_safety_calls_removed_from_replay(sdk_model, streaming, arguments, asynchronous): + model, requests = sdk_model(streaming=streaming, arguments=arguments) + original = await model.ainvoke("probe") if asynchronous else model.invoke("probe") + assert original.response_metadata["incomplete_details"] == {"reason": "content_filter"} + assert original.tool_calls or original.invalid_tool_calls + runtime = SimpleNamespace(context={}) + guard = SafetyFinishReasonMiddleware() + state = {"messages": [original]} + update = await guard.aafter_model(state, runtime) if asynchronous else guard.after_model(state, runtime) + assert update is not None + cleaned = update["messages"][0] + assert not cleaned.tool_calls and not cleaned.invalid_tool_calls + assert cleaned.response_metadata == original.response_metadata + assert original.tool_calls or original.invalid_tool_calls # no mutation + assert runtime.context["stop_reason"] == "safety_capped" + assert cleaned.additional_kwargs["safety_termination"]["suppressed_tool_call_count"] == 1 + history = [HumanMessage(content="probe"), cleaned, HumanMessage(content="continue")] + if asynchronous: + await model.ainvoke(history) + else: + model.invoke(history) + assert all(item.get("type") not in ("function_call", "function_call_output") for item in requests[-1]["input"]) + + +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.parametrize("reason", ["content_filter", None]) +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.asyncio +async def test_compiled_agent_suppresses_only_filtered_tool(sdk_model, streaming, reason, asynchronous): + effects = [] + + @tool + def record_probe() -> str: + """Record a harmless in-memory marker.""" + effects.append("ran") + return "recorded" + + model, requests = sdk_model(streaming=streaming, reason=reason) + graph = create_agent(model, tools=[record_probe], middleware=[SafetyFinishReasonMiddleware()]) + state = {"messages": [HumanMessage(content="probe")]} + result = await graph.ainvoke(state) if asynchronous else graph.invoke(state) + assert effects == ([] if reason else ["ran"]) + assert len(requests) == (1 if reason else 2) + assert not result["messages"][-1].tool_calls + + +@pytest.mark.parametrize( + "metadata", + [ + {}, + {"status": "completed", "incomplete_details": {"reason": "content_filter"}}, + {"status": "incomplete", "incomplete_details": {"reason": "max_output_tokens"}}, + {"status": "incomplete", "incomplete_details": None}, + {"status": "incomplete", "incomplete_details": "content_filter"}, + {"status": "incomplete", "incomplete_details": {"reason": []}}, + ], +) +def test_non_safety_responses_metadata_ignored(metadata): + message = AIMessage(content="", response_metadata=metadata) + assert all(detector.detect(message) is None for detector in default_detectors()) + + +@pytest.mark.parametrize("container", ["response_metadata", "additional_kwargs"]) +def test_responses_signal_detected_in_one_container(container): + message = AIMessage(content="", **{container: {"status": "incomplete", "incomplete_details": {"reason": "content_filter"}}}) + hits = [hit for detector in default_detectors() if (hit := detector.detect(message)) is not None] + assert len(hits) == 1 + assert hits[0].reason_field == "incomplete_details.reason" + + +def test_responses_signal_not_combined_across_containers(): + message = AIMessage(content="", response_metadata={"status": "incomplete"}, additional_kwargs={"incomplete_details": {"reason": "content_filter"}}) + assert all(detector.detect(message) is None for detector in default_detectors()) + + +@pytest.mark.parametrize("content", ["", "Partial answer"]) +def test_filtered_text_keeps_existing_empty_and_visible_behavior(content): + message = AIMessage(content=content, response_metadata={"status": "incomplete", "incomplete_details": {"reason": "content_filter"}}) + update = SafetyFinishReasonMiddleware().after_model({"messages": [message]}, SimpleNamespace(context={})) + if content: + assert update is None + else: + assert update is not None + assert "returned no content" in update["messages"][0].content + + +@pytest.mark.parametrize("native", [False, True]) +def test_invalid_safety_call_with_partial_text_is_suppressed_and_audited(native): + metadata = {"status": "incomplete", "incomplete_details": {"reason": "content_filter"}} if native else {"finish_reason": "content_filter"} + message = AIMessage( + id="msg_probe", + content=[{"type": "text", "text": "Partial answer"}, {"type": "function_call", "call_id": "call_probe", "name": "record_probe", "arguments": "{"}], + invalid_tool_calls=[{"name": "record_probe", "id": "call_probe", "args": "{", "error": "truncated"}], + response_metadata=metadata, + ) + journal = MagicMock() + runtime = SimpleNamespace(context={"__run_journal": journal}) + update = SafetyFinishReasonMiddleware().after_model({"messages": [message]}, runtime) + assert update is not None + cleaned = update["messages"][0] + assert cleaned.id == message.id + assert cleaned.content[0] == message.content[0] + assert all(block["type"] != "function_call" for block in cleaned.content) + assert not cleaned.invalid_tool_calls + changes = journal.record_middleware.call_args.kwargs["changes"] + assert changes["suppressed_tool_call_count"] == 1 + assert changes["suppressed_tool_call_ids"] == ["call_probe"] + assert "args" not in json.dumps(changes) diff --git a/backend/tests/test_safety_termination_detectors.py b/backend/tests/test_safety_termination_detectors.py index 0679aed0e9..acbf8e4e40 100644 --- a/backend/tests/test_safety_termination_detectors.py +++ b/backend/tests/test_safety_termination_detectors.py @@ -149,17 +149,17 @@ class TestGeminiSafetyDetector: class TestDefaultDetectorSet: - def test_default_set_returns_three_detectors(self): + def test_default_set_returns_four_detectors(self): dets = default_detectors() names = {d.name for d in dets} - assert names == {"openai_compatible_content_filter", "anthropic_refusal", "gemini_safety"} + assert names == {"openai_compatible_content_filter", "openai_responses_content_filter", "anthropic_refusal", "gemini_safety"} def test_default_set_returns_fresh_list(self): # Caller mutation must not affect later calls. first = default_detectors() first.clear() second = default_detectors() - assert len(second) == 3 + assert len(second) == 4 class TestProtocolConformance: