mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-11-15 07:41:20 +00:00
fix(gateway): reject unusable tool_artifacts entries at the input boundary (#8207)
* fix(gateway): reject unusable tool_artifacts entries at the input boundary merge_tool_artifacts keys every registry entry by handle, so a caller-supplied entry with a missing or unhashable handle raised while the state update merged; the thread-state route persists reducer values with Overwrite, so the bad shape was stored and every later artifact capture failed on it. Validate the list-of-objects shape with a non-empty string handle on both boundaries, and skip such an entry in the reducer so a poisoned checkpoint heals. * fix(gateway): keep the per-entry metadata strip for tool_artifacts The dedicated registry branch replaced the generic list branch and silently dropped its _strip_external_metadata_from_message_like pass. Registry entries are rich dicts, so a caller-crafted entry could persist server-owned message metadata into the checkpointed registry. Apply the strip to each validated entry; it is a no-op for legitimate registry entries.
This commit is contained in:
parent
ae871431a2
commit
599105e0fb
@ -702,6 +702,33 @@ def _normalize_external_skill_context(value: Any, *, location: str) -> list[dict
|
||||
return normalized
|
||||
|
||||
|
||||
def _normalize_external_tool_artifacts(value: Any, *, location: str) -> list[dict[str, Any]] | None:
|
||||
"""Reject artifact-registry entries the reducer cannot key.
|
||||
|
||||
``tool_artifacts`` is a checkpointed reducer channel a caller may Overwrite.
|
||||
``merge_tool_artifacts`` keys each entry by its ``handle``, so an entry whose
|
||||
handle is missing or unhashable raises while the state update merges — a 500
|
||||
instead of a 400 at this boundary. The thread-state route persists the value
|
||||
with ``Overwrite`` before any reducer sees it, so the thread stays broken for
|
||||
every later artifact capture. Require the registry's list-of-objects shape
|
||||
with a non-empty string handle; the reducer also tolerates an already-poisoned
|
||||
checkpoint.
|
||||
"""
|
||||
if value is None:
|
||||
return None
|
||||
if not isinstance(value, list):
|
||||
raise HTTPException(status_code=400, detail=f"{location} must be a list of artifact objects")
|
||||
normalized: list[dict[str, Any]] = []
|
||||
for index, entry in enumerate(value):
|
||||
if not isinstance(entry, Mapping):
|
||||
raise HTTPException(status_code=400, detail=f"{location}[{index}] must be an object")
|
||||
handle = entry.get("handle")
|
||||
if not isinstance(handle, str) or not handle:
|
||||
raise HTTPException(status_code=400, detail=f"{location}[{index}].handle must be a non-empty string")
|
||||
normalized.append(dict(entry))
|
||||
return normalized
|
||||
|
||||
|
||||
def _normalize_external_goal(value: Any, *, location: str) -> dict[str, Any] | None:
|
||||
"""Require usable goal text without rebuilding caller-supplied goal state."""
|
||||
if value is None:
|
||||
@ -759,6 +786,12 @@ 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 == "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
|
||||
# strip the generic list branch applied: a caller-crafted entry must
|
||||
# not smuggle server-owned message metadata into the checkpoint.
|
||||
stripped[channel] = None if normalized_tool_artifacts is None else [_strip_external_metadata_from_message_like(item) for item in normalized_tool_artifacts]
|
||||
elif isinstance(value, list):
|
||||
stripped[channel] = [_strip_external_metadata_from_message_like(item) for item in value]
|
||||
else:
|
||||
@ -829,6 +862,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 "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.
|
||||
if result.get("goal") is not None:
|
||||
normalized_goal = _normalize_external_goal(result["goal"], location="input.goal")
|
||||
|
||||
@ -398,9 +398,16 @@ def merge_tool_artifacts(existing: list[ArtifactEntry] | None, new: list[Artifac
|
||||
by_handle: dict[str, ArtifactEntry] = {}
|
||||
order: list[str] = []
|
||||
for entry in [*(existing or []), *new]:
|
||||
if not isinstance(entry, dict) or "handle" not in entry:
|
||||
if not isinstance(entry, dict):
|
||||
continue
|
||||
# The state-update route persists reducer values with ``Overwrite``, so
|
||||
# a checkpoint can carry an entry whose handle is missing or unhashable
|
||||
# without this reducer ever seeing it. Skip such an entry instead of
|
||||
# raising on the dict key: one bad write must not fail every later
|
||||
# artifact capture on the thread.
|
||||
handle = entry.get("handle")
|
||||
if not isinstance(handle, str) or not handle:
|
||||
continue
|
||||
handle = entry["handle"]
|
||||
if handle not in by_handle:
|
||||
order.append(handle)
|
||||
by_handle[handle] = entry
|
||||
|
||||
@ -629,6 +629,19 @@ def test_merge_tool_artifacts_empty_new_preserves_existing():
|
||||
assert merged2 == existing
|
||||
|
||||
|
||||
def test_merge_tool_artifacts_skips_an_unhashable_handle():
|
||||
"""A checkpoint poisoned through the state route must not fail later captures."""
|
||||
merged = merge_tool_artifacts([{"handle": ["poisoned"]}], [{"handle": "art_1", "display_name": "a.txt"}])
|
||||
|
||||
assert [entry["handle"] for entry in merged] == ["art_1"]
|
||||
|
||||
|
||||
def test_merge_tool_artifacts_skips_missing_or_empty_handles():
|
||||
merged = merge_tool_artifacts(None, [{"display_name": "no handle"}, {"handle": ""}, {"handle": "art_2"}])
|
||||
|
||||
assert [entry["handle"] for entry in merged] == ["art_2"]
|
||||
|
||||
|
||||
def test_merge_tool_artifacts_trim_directive_slides_window():
|
||||
"""A trailing trim directive makes the configured cap a sliding window."""
|
||||
existing = [_entry(f"art_{i:08x}", i) for i in range(20)]
|
||||
|
||||
@ -861,6 +861,69 @@ def test_normalize_input_rejects_malformed_message_with_400():
|
||||
assert "input.messages[1]" in excinfo.value.detail
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"tool_artifacts",
|
||||
[
|
||||
"oops",
|
||||
["oops"],
|
||||
[{"display_name": "report.md"}],
|
||||
[{"handle": 3}],
|
||||
[{"handle": ["h"]}],
|
||||
{"type": "__overwrite__", "value": []},
|
||||
],
|
||||
ids=["non-list", "non-object-entry", "missing-handle", "non-string-handle", "unhashable-handle", "overwrite-shaped"],
|
||||
)
|
||||
def test_normalize_input_rejects_unusable_tool_artifacts_with_400(tool_artifacts):
|
||||
"""An entry the registry reducer cannot key must be a 400 here.
|
||||
|
||||
``merge_tool_artifacts`` keys each entry by ``handle``; a missing or
|
||||
unhashable handle raised while the state update merged, and the state route
|
||||
persists reducer values with ``Overwrite``, so the thread stayed broken for
|
||||
every later artifact capture.
|
||||
"""
|
||||
from fastapi import HTTPException
|
||||
|
||||
from app.gateway.services import normalize_input
|
||||
|
||||
with pytest.raises(HTTPException) as excinfo:
|
||||
normalize_input({"tool_artifacts": tool_artifacts})
|
||||
|
||||
assert excinfo.value.status_code == 400
|
||||
assert "input.tool_artifacts" in excinfo.value.detail
|
||||
|
||||
|
||||
def test_normalize_input_keeps_usable_tool_artifacts():
|
||||
from app.gateway.services import normalize_input
|
||||
|
||||
entry = {"handle": "art_00000001", "display_name": "report.md", "artifact_type": "file"}
|
||||
|
||||
assert normalize_input({"tool_artifacts": [entry]})["tool_artifacts"] == [entry]
|
||||
|
||||
|
||||
def test_state_values_reject_unusable_tool_artifacts_with_400():
|
||||
"""The thread-state update route shares the artifact-registry 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({"tool_artifacts": [{"handle": ["h"]}]})
|
||||
|
||||
assert excinfo.value.status_code == 400
|
||||
assert "values.tool_artifacts" in excinfo.value.detail
|
||||
|
||||
|
||||
def test_state_values_strip_server_owned_metadata_from_tool_artifacts():
|
||||
"""The registry branch keeps the per-entry strip the generic list branch applied."""
|
||||
from app.gateway.services import strip_server_owned_state_metadata
|
||||
|
||||
entry = {"handle": "art_1", "additional_kwargs": {"original_user_content": "forged", "title": "kept"}}
|
||||
|
||||
stripped = strip_server_owned_state_metadata({"tool_artifacts": [entry]})["tool_artifacts"][0]
|
||||
|
||||
assert stripped["additional_kwargs"] == {"title": "kept"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"skill_context",
|
||||
[
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user