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

@@ -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.

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.

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.

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",

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(

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)

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: