diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index a371c7b899..46baa35c3c 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -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") diff --git a/backend/packages/harness/deerflow/agents/thread_state.py b/backend/packages/harness/deerflow/agents/thread_state.py index d4656bd923..d213e54485 100644 --- a/backend/packages/harness/deerflow/agents/thread_state.py +++ b/backend/packages/harness/deerflow/agents/thread_state.py @@ -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 diff --git a/backend/tests/test_artifact_registry.py b/backend/tests/test_artifact_registry.py index b41438761f..cfe14c5587 100644 --- a/backend/tests/test_artifact_registry.py +++ b/backend/tests/test_artifact_registry.py @@ -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)] diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py index 6a1e5c4889..d77a1252aa 100644 --- a/backend/tests/test_gateway_services.py +++ b/backend/tests/test_gateway_services.py @@ -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", [