From 3e2489429ff3061489537d7303240fef3f199f83 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=80=E7=89=87=E4=B8=80=E7=89=87?= <3037678834@qq.com> Date: Sat, 14 Nov 2026 20:22:43 +0800 Subject: [PATCH] fix(gateway): reject an unusable promoted record at the input boundary (#8170) * fix(gateway): reject an unusable promoted record at the input boundary * fix(gateway): refuse an Overwrite-marker promotion record * test(gateway): pin the server-owned metadata strip on promoted --------- Co-authored-by: Willem Jiang --- backend/app/gateway/services.py | 51 +++++++- .../harness/deerflow/agents/thread_state.py | 7 + backend/tests/test_gateway_services.py | 120 ++++++++++++++++++ backend/tests/test_thread_state_promoted.py | 14 ++ 4 files changed, 190 insertions(+), 2 deletions(-) diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index 1c0fd7f10e..ba588e37a1 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -702,6 +702,48 @@ def _normalize_external_skill_context(value: Any, *, location: str) -> list[dict return normalized +def _reject_overwrite_marker(value: Mapping[str, Any], *, location: str) -> None: + """Refuse a value LangGraph would apply with ``Overwrite`` semantics. + + ``Overwrite`` reaches the validator in two dict spellings: the sentinel form + ``{"__overwrite__": value}`` and the dataclass-erased form an orjson round + trip produces, ``{"type": "__overwrite__", "value": value}``. Either one + bypasses the channel reducer and stores the caller's value directly on an + initialized channel, so a shape check alone cannot catch it. + """ + if "__overwrite__" in value or value.get("type") == "__overwrite__": + raise HTTPException(status_code=400, detail=f"{location} cannot contain an Overwrite marker") + + +def _normalize_external_promoted(value: Any, *, location: str) -> dict[str, Any] | None: + """Reject deferred-tool promotion records the reducer cannot merge. + + ``promoted`` is a checkpointed reducer channel a caller may Overwrite even + though only the server writes it. ``merge_promoted`` calls ``new.get(...)`` + and indexes ``new["catalog_hash"]``/``new["names"]``, so a string, a missing + key or a non-list ``names`` raises while the state update merges — failing + the run with an unhandled ``AttributeError``/``KeyError``/``TypeError`` + instead of a 400 at this boundary. On the thread-state route the bad shape + would also be persisted, so every later promotion write on that thread would + keep failing on it. + """ + if value is None: + return None + if not isinstance(value, Mapping): + raise HTTPException(status_code=400, detail=f"{location} must be an object with a catalog_hash and a names list") + _reject_overwrite_marker(value, location=location) + catalog_hash = value.get("catalog_hash") + if not isinstance(catalog_hash, str) or not catalog_hash: + raise HTTPException(status_code=400, detail=f"{location}.catalog_hash must be a non-empty string") + names = value.get("names") + if not isinstance(names, list) or not all(isinstance(name, str) for name in names): + raise HTTPException(status_code=400, detail=f"{location}.names must be a list of strings") + turn_id = value.get("turn_id") + if turn_id is not None and not isinstance(turn_id, str): + raise HTTPException(status_code=400, detail=f"{location}.turn_id must be a string") + return {**value, "catalog_hash": catalog_hash, "names": names} + + def _normalize_external_tool_artifacts(value: Any, *, location: str) -> list[dict[str, Any]] | None: """Reject artifact-registry entries the reducer cannot key. @@ -735,8 +777,7 @@ def _normalize_external_goal(value: Any, *, location: str) -> dict[str, Any] | N return None if not isinstance(value, Mapping) or not isinstance(value.get("objective"), str): raise HTTPException(status_code=400, detail=f"{location} must be an object with a string objective") - if "__overwrite__" in value or value.get("type") == "__overwrite__": - raise HTTPException(status_code=400, detail=f"{location} cannot contain an Overwrite marker") + _reject_overwrite_marker(value, location=location) try: objective = normalize_goal_objective(value["objective"]) except ValueError as exc: @@ -786,6 +827,10 @@ def strip_server_owned_state_metadata(values: Mapping[str, Any]) -> dict[str, An elif channel == "skill_context": normalized_skill_context = _normalize_external_skill_context(value, location="values.skill_context") stripped[channel] = None if normalized_skill_context is None else [_strip_external_metadata_from_message_like(item) for item in normalized_skill_context] + elif channel == "promoted": + # A dedicated validator must not bypass the metadata strip the + # generic branches apply to caller-supplied channel values. + stripped[channel] = _strip_external_metadata_from_message_like(_normalize_external_promoted(value, location="values.promoted")) elif channel == "tool_artifacts": normalized_tool_artifacts = _normalize_external_tool_artifacts(value, location="values.tool_artifacts") # Registry entries are rich dicts, so keep the per-entry metadata @@ -862,6 +907,8 @@ def normalize_input(raw_input: dict[str, Any] | None, *, trusted_internal: bool result = {**result, "delegations": cleaned} if "skill_context" in result: result = {**result, "skill_context": _normalize_external_skill_context(result["skill_context"], location="input.skill_context")} + if "promoted" in result: + result = {**result, "promoted": _normalize_external_promoted(result["promoted"], location="input.promoted")} if "tool_artifacts" in result: result = {**result, "tool_artifacts": _normalize_external_tool_artifacts(result["tool_artifacts"], location="input.tool_artifacts")} # merge_goal ignores None, so only a goal value replaces the goal. diff --git a/backend/packages/harness/deerflow/agents/thread_state.py b/backend/packages/harness/deerflow/agents/thread_state.py index d213e54485..dd08454b72 100644 --- a/backend/packages/harness/deerflow/agents/thread_state.py +++ b/backend/packages/harness/deerflow/agents/thread_state.py @@ -171,9 +171,16 @@ def merge_promoted(existing: PromotedTools | None, new: PromotedTools | None) -> - same scope -> union names, dedupe, preserve order. - a missing new turn_id inherits the existing scope. ``tool_search`` writes this shape after the filter establishes the current turn before the model. + - a non-object value from either side is unusable and carries nothing to + merge: an Overwrite-poisoned checkpoint stores one, and without this the + thread's later genuine promotion writes would keep failing on it. """ + if not isinstance(existing, Mapping): + existing = None if not new: return existing + if not isinstance(new, Mapping): + return existing new_turn_id = new.get("turn_id") replace = existing is None or existing.get("catalog_hash") != new["catalog_hash"] or (new_turn_id is not None and existing.get("turn_id") != new_turn_id) existing_names = existing.get("names") if existing is not None else None diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py index d77a1252aa..1eb0dc521d 100644 --- a/backend/tests/test_gateway_services.py +++ b/backend/tests/test_gateway_services.py @@ -974,6 +974,126 @@ def test_state_values_reject_unusable_skill_context_with_400(): assert "values.skill_context" in excinfo.value.detail +@pytest.mark.parametrize( + "promoted", + [ + "oops", + ["oops"], + {"names": ["search"]}, + {"catalog_hash": "abc"}, + {"catalog_hash": "", "names": []}, + {"catalog_hash": "abc", "names": "search"}, + {"catalog_hash": "abc", "names": [1]}, + {"catalog_hash": "abc", "names": [], "turn_id": 3}, + {"catalog_hash": "abc", "names": [], "type": "__overwrite__", "value": "oops"}, + {"catalog_hash": "abc", "names": [], "__overwrite__": "oops"}, + ], + ids=[ + "string-channel", + "list-channel", + "missing-catalog-hash", + "missing-names", + "empty-catalog-hash", + "names-not-a-list", + "names-not-strings", + "turn-id-not-a-string", + "erased-overwrite-marker", + "sentinel-overwrite-marker", + ], +) +def test_normalize_input_rejects_unusable_promoted_with_400(promoted): + """A promotion record the deferred-tool reducer cannot merge must be a 400. + + ``merge_promoted`` calls ``new.get(...)`` and indexes ``new["catalog_hash"]`` + and ``new["names"]``, so each shape below raises ``AttributeError``, + ``KeyError`` or ``TypeError`` while the state update merges — a 500 for the + caller instead of a refusal at the boundary. + """ + from fastapi import HTTPException + + from app.gateway.services import normalize_input + + with pytest.raises(HTTPException) as excinfo: + normalize_input({"promoted": promoted}) + + assert excinfo.value.status_code == 400 + assert "input.promoted" in excinfo.value.detail + + +def test_normalize_input_keeps_usable_promoted_record(): + from app.gateway.services import normalize_input + + record = {"catalog_hash": "abc", "names": ["search", "fetch"], "turn_id": "user-1"} + + assert normalize_input({"promoted": record})["promoted"] == record + assert normalize_input({"promoted": None})["promoted"] is None + + +def test_state_values_reject_unusable_promoted_with_400(): + """The thread-state update path persists this channel, so it shares the boundary.""" + from fastapi import HTTPException + + from app.gateway.services import strip_server_owned_state_metadata + + with pytest.raises(HTTPException) as excinfo: + strip_server_owned_state_metadata({"promoted": "oops"}) + + assert excinfo.value.status_code == 400 + assert "values.promoted" in excinfo.value.detail + + +def test_state_values_reject_an_overwrite_marker_in_promoted(): + """An ``Overwrite`` written as a plain dict bypasses the channel reducer. + + LangGraph accepts the dataclass-erased form ``{"type": "__overwrite__", + "value": ...}`` on an initialized channel and stores the value directly, so + a record that carries it alongside valid fields would persist a non-object + promotion value that the shape checks above cannot see. + """ + from fastapi import HTTPException + + from app.gateway.services import strip_server_owned_state_metadata + + with pytest.raises(HTTPException) as excinfo: + strip_server_owned_state_metadata({"promoted": {"catalog_hash": "abc", "names": [], "type": "__overwrite__", "value": "oops"}}) + + assert excinfo.value.status_code == 400 + assert "values.promoted" in excinfo.value.detail + assert "Overwrite" in excinfo.value.detail + + +def test_state_values_keep_usable_promoted_record(): + from app.gateway.services import strip_server_owned_state_metadata + + record = {"catalog_hash": "abc", "names": ["search"], "turn_id": "user-1"} + + assert strip_server_owned_state_metadata({"promoted": record})["promoted"] == record + + +def test_state_values_strip_forged_metadata_from_promoted_record(): + """The dedicated ``promoted`` branch must not bypass the generic metadata strip. + + A caller-supplied record reaches the checkpoint through this branch, so a + preserved ``additional_kwargs`` would seed server-owned provenance + (``origin_task_id``) that only the server may write. The sibling + implementation of this boundary returns the accepted record untouched, so + this case pins the difference and fails on that shape. + """ + from app.gateway.services import strip_server_owned_state_metadata + + stripped = strip_server_owned_state_metadata( + { + "promoted": { + "catalog_hash": "abc", + "names": [], + "additional_kwargs": {"origin_task_id": "forged", "keep": "me"}, + } + } + ) + + assert stripped["promoted"]["additional_kwargs"] == {"keep": "me"} + + @pytest.mark.parametrize( "message", [ diff --git a/backend/tests/test_thread_state_promoted.py b/backend/tests/test_thread_state_promoted.py index 0ff2ea4690..7c94f39362 100644 --- a/backend/tests/test_thread_state_promoted.py +++ b/backend/tests/test_thread_state_promoted.py @@ -73,3 +73,17 @@ def test_merge_promoted_recovers_from_malformed_existing_names(): result = merge_promoted(existing, {"catalog_hash": "abc", "names": ["search"]}) assert result == {"catalog_hash": "abc", "names": ["search"], "turn_id": "user-1"} + + +def test_merge_promoted_replaces_non_object_existing(): + # A checkpoint poisoned before the gateway validated this channel stores a + # raw Overwrite value; genuine promotion writes must not keep failing on it. + result = merge_promoted("oops", {"catalog_hash": "abc", "names": ["search"]}) + + assert result == {"catalog_hash": "abc", "names": ["search"]} + + +def test_merge_promoted_ignores_non_object_update(): + existing = {"catalog_hash": "abc", "names": ["search"], "turn_id": "user-1"} + + assert merge_promoted(existing, "oops") is existing