mirror of
https://github.com/usestrix/strix.git
synced 2026-08-28 05:25:00 +00:00
Treat literal 'null'/'none' strings as absent for optional tool args
Models routinely pass the literal string "null" or "none" instead of omitting an optional argument. Taken at face value it becomes a filter that matches nothing, so tools like list_notes / list_reports / list_requests silently return no results. Coerce such values to None in the central argument-coercion layer, but only for parameters the schema allows to be null (or that are absent from a declared "required" list), so required strings keep the literal value. The list/filter helpers normalize the same values too, so a direct call can't regress.
This commit is contained in:
parent
f4ef8867f6
commit
dd7ee42336
8 changed files with 191 additions and 7 deletions
|
|
@ -36,6 +36,7 @@ from strix.tools.notes.tools import (
|
|||
list_notes,
|
||||
update_note,
|
||||
)
|
||||
from strix.tools.nullish import is_nullish
|
||||
from strix.tools.output_store import bound_and_store, bound_text
|
||||
from strix.tools.proxy.tools import (
|
||||
list_requests,
|
||||
|
|
@ -164,6 +165,28 @@ def _schema_types(spec: dict[str, Any]) -> set[str]:
|
|||
return types
|
||||
|
||||
|
||||
def _allows_null(spec: dict[str, Any]) -> bool:
|
||||
raw = spec.get("type")
|
||||
if raw == "null" or (isinstance(raw, list) and "null" in raw):
|
||||
return True
|
||||
return any(
|
||||
isinstance(variant, dict) and _allows_null(variant) for variant in spec.get("anyOf") or ()
|
||||
)
|
||||
|
||||
|
||||
def _is_nullable(key: str, spec: dict[str, Any], schema: dict[str, Any]) -> bool:
|
||||
"""Whether ``key`` may be ``None``.
|
||||
|
||||
Strict schemas list every property as required, so nullability shows up as a
|
||||
``null`` type variant; without a declared one, fall back to the property
|
||||
being absent from a declared ``required`` list.
|
||||
"""
|
||||
if _allows_null(spec):
|
||||
return True
|
||||
required = schema.get("required")
|
||||
return isinstance(required, list) and key not in required
|
||||
|
||||
|
||||
def _decode_structured(value: str, types: set[str]) -> Any:
|
||||
stripped = value.strip()
|
||||
if not stripped:
|
||||
|
|
@ -178,9 +201,14 @@ def _decode_structured(value: str, types: set[str]) -> Any:
|
|||
return decoded if isinstance(decoded, wanted) else value
|
||||
|
||||
|
||||
def _coerce_argument(value: Any, spec: dict[str, Any]) -> Any:
|
||||
def _coerce_argument(value: Any, spec: dict[str, Any], *, nullable: bool = False) -> Any:
|
||||
if value is None:
|
||||
return value
|
||||
if nullable and is_nullish(value):
|
||||
# The model's stand-in for "no value"; as a filter it matches nothing.
|
||||
return None
|
||||
types = _schema_types(spec)
|
||||
if not types or value is None:
|
||||
if not types:
|
||||
return value
|
||||
if isinstance(value, list | dict) and "string" in types and not types & {"array", "object"}:
|
||||
return json.dumps(value, ensure_ascii=False)
|
||||
|
|
@ -205,7 +233,7 @@ def _coerce_arguments(raw_input: str, schema: dict[str, Any]) -> str:
|
|||
spec = properties.get(key)
|
||||
if not isinstance(spec, dict):
|
||||
continue
|
||||
coerced = _coerce_argument(value, spec)
|
||||
coerced = _coerce_argument(value, spec, nullable=_is_nullable(key, spec, schema))
|
||||
if coerced is not value:
|
||||
payload[key] = coerced
|
||||
changed = True
|
||||
|
|
|
|||
|
|
@ -14,6 +14,8 @@ from typing import Any
|
|||
|
||||
from agents import RunContextWrapper, function_tool
|
||||
|
||||
from strix.tools.nullish import clean_optional, is_nullish
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -111,6 +113,10 @@ def _filter_notes(
|
|||
tags: list[str] | None = None,
|
||||
search_query: str | None = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
category = clean_optional(category)
|
||||
search_query = clean_optional(search_query)
|
||||
tags = [tag for tag in tags if not is_nullish(tag)] if tags else None
|
||||
|
||||
filtered: list[dict[str, Any]] = []
|
||||
for note_id, note in _notes_storage.items():
|
||||
if category and note.get("category") != category:
|
||||
|
|
|
|||
23
strix/tools/nullish.py
Normal file
23
strix/tools/nullish.py
Normal file
|
|
@ -0,0 +1,23 @@
|
|||
"""Nullish argument values passed by models in place of omitting an argument.
|
||||
|
||||
Models frequently send the literal string ``"null"`` / ``"none"`` for an
|
||||
optional filter argument instead of leaving it out. Taken at face value it is
|
||||
a filter that matches nothing, so the call quietly returns no results.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
||||
NULLISH_STRINGS = frozenset({"null", "none", "nil", "undefined"})
|
||||
|
||||
|
||||
def is_nullish(value: object) -> bool:
|
||||
"""Whether ``value`` is a string standing in for "no value"."""
|
||||
return isinstance(value, str) and value.strip().lower() in NULLISH_STRINGS
|
||||
|
||||
|
||||
def clean_optional(value: str | None) -> str | None:
|
||||
"""Normalize an optional filter argument: nullish or blank becomes ``None``."""
|
||||
if value is None or is_nullish(value):
|
||||
return None
|
||||
return value.strip() or None
|
||||
|
|
@ -14,6 +14,7 @@ from typing import TYPE_CHECKING, Any, Literal
|
|||
from agents import RunContextWrapper, function_tool
|
||||
|
||||
from strix.runtime.caido_handle import CaidoBootstrapHandle
|
||||
from strix.tools.nullish import clean_optional
|
||||
from strix.tools.proxy import caido_api
|
||||
|
||||
|
||||
|
|
@ -167,6 +168,10 @@ async def list_requests(
|
|||
if client is None:
|
||||
return _no_client()
|
||||
|
||||
httpql_filter = clean_optional(httpql_filter)
|
||||
after = clean_optional(after)
|
||||
scope_id = clean_optional(scope_id)
|
||||
|
||||
try:
|
||||
connection = await _call(
|
||||
client,
|
||||
|
|
@ -472,6 +477,8 @@ async def list_sitemap(
|
|||
client = await _ctx_client(ctx)
|
||||
if client is None:
|
||||
return _no_client()
|
||||
scope_id = clean_optional(scope_id)
|
||||
parent_id = clean_optional(parent_id)
|
||||
try:
|
||||
payload = await _call(
|
||||
client,
|
||||
|
|
|
|||
|
|
@ -16,6 +16,8 @@ from typing import Any
|
|||
|
||||
from agents import RunContextWrapper, function_tool
|
||||
|
||||
from strix.tools.nullish import clean_optional
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -1605,12 +1607,12 @@ def _do_list_reports(
|
|||
caller_agent_id: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
errors: list[str] = []
|
||||
severity = (severity or "").strip().lower() or None
|
||||
severity = (clean_optional(severity) or "").lower() or None
|
||||
if severity and severity not in _VALID_SEVERITIES:
|
||||
errors.append(
|
||||
f"Invalid severity: {severity!r}. Must be one of: {sorted(_VALID_SEVERITIES)}"
|
||||
)
|
||||
finding_class = (finding_class or "").strip().lower() or None
|
||||
finding_class = (clean_optional(finding_class) or "").lower() or None
|
||||
if finding_class and finding_class not in _VALID_FINDING_CLASSES:
|
||||
errors.append(
|
||||
f"Invalid finding_class: {finding_class!r}. "
|
||||
|
|
@ -1640,8 +1642,8 @@ def _do_list_reports(
|
|||
r,
|
||||
severity=severity,
|
||||
finding_class=finding_class,
|
||||
target=(target or "").strip() or None,
|
||||
search=(search or "").strip() or None,
|
||||
target=clean_optional(target),
|
||||
search=clean_optional(search),
|
||||
)
|
||||
]
|
||||
matched.sort(key=lambda r: (_report_severity_rank(r), str(r.get("id", ""))))
|
||||
|
|
|
|||
|
|
@ -9,6 +9,8 @@ import pytest
|
|||
from agents.tool import FunctionTool
|
||||
|
||||
from strix.agents import factory
|
||||
from strix.tools.notes.tools import list_notes
|
||||
from strix.tools.reporting.tool import list_reports
|
||||
|
||||
|
||||
def _capturing_tool(captured: dict[str, str], schema: dict[str, Any]) -> FunctionTool:
|
||||
|
|
@ -144,3 +146,76 @@ async def test_coercion_is_applied_once_per_tool() -> None:
|
|||
tool = factory._with_coerced_arguments(_capturing_tool(captured, _ARRAY))
|
||||
|
||||
assert factory._with_coerced_arguments(tool) is tool
|
||||
|
||||
|
||||
_NULLABLE_STRING = {"category": {"anyOf": [{"type": "string"}, {"type": "null"}]}}
|
||||
_NULLABLE_STRING_TYPE_LIST = {"category": {"type": ["string", "null"]}}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("schema", [_NULLABLE_STRING, _NULLABLE_STRING_TYPE_LIST])
|
||||
@pytest.mark.parametrize("value", ["null", "none", "NULL", " None ", "nil", "undefined"])
|
||||
async def test_nullish_string_on_a_nullable_parameter_becomes_none(
|
||||
schema: dict[str, Any], value: str
|
||||
) -> None:
|
||||
parsed = await _roundtrip(schema, {"category": value})
|
||||
|
||||
assert parsed["category"] is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_nullish_string_on_a_required_parameter_is_untouched() -> None:
|
||||
schema = {"content": {"type": "string"}}
|
||||
captured: dict[str, str] = {}
|
||||
tool = _capturing_tool(captured, schema)
|
||||
tool.params_json_schema["required"] = ["content"]
|
||||
wrapped = factory._with_coerced_arguments(tool)
|
||||
|
||||
assert await wrapped.on_invoke_tool(cast("Any", None), json.dumps({"content": "none"})) == "ok"
|
||||
assert json.loads(captured["raw_input"])["content"] == "none"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_parameter_absent_from_required_is_treated_as_nullable() -> None:
|
||||
captured: dict[str, str] = {}
|
||||
tool = _capturing_tool(captured, {"category": {"type": "string"}})
|
||||
tool.params_json_schema["required"] = []
|
||||
wrapped = factory._with_coerced_arguments(tool)
|
||||
|
||||
assert await wrapped.on_invoke_tool(cast("Any", None), json.dumps({"category": "null"})) == "ok"
|
||||
assert json.loads(captured["raw_input"])["category"] is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_nullish_string_without_a_required_list_is_untouched() -> None:
|
||||
parsed = await _roundtrip(_STRING, {"todos": "none"})
|
||||
|
||||
assert parsed["todos"] == "none"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_nullish_looking_content_is_not_coerced() -> None:
|
||||
parsed = await _roundtrip(_NULLABLE_STRING, {"category": "none of the endpoints reflect input"})
|
||||
|
||||
assert parsed["category"] == "none of the endpoints reflect input"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_empty_string_on_a_nullable_string_parameter_is_untouched() -> None:
|
||||
parsed = await _roundtrip(_NULLABLE_STRING, {"category": ""})
|
||||
|
||||
assert parsed["category"] == ""
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_nullish_string_on_a_nullable_array_parameter_becomes_none() -> None:
|
||||
parsed = await _roundtrip(_NULLABLE_ARRAY, {"tags": "null"})
|
||||
|
||||
assert parsed["tags"] is None
|
||||
|
||||
|
||||
def test_real_tool_schemas_declare_optional_filters_as_nullable() -> None:
|
||||
for tool, params in ((list_notes, ("category", "search")), (list_reports, ("target",))):
|
||||
schema = tool.params_json_schema
|
||||
for param in params:
|
||||
assert factory._is_nullable(param, schema["properties"][param], schema)
|
||||
|
|
|
|||
|
|
@ -357,3 +357,25 @@ def test_get_report_no_state_returns_error(monkeypatch: pytest.MonkeyPatch) -> N
|
|||
result = _do_get_report("vuln-0001")
|
||||
assert result["success"] is False
|
||||
assert result["report"] is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("nullish", ["null", "none", "NULL", " None ", "undefined", "nil"])
|
||||
def test_list_reports_ignores_nullish_filter_strings(
|
||||
report_state: ReportState, nullish: str
|
||||
) -> None:
|
||||
_seed(report_state)
|
||||
unfiltered = _do_list_reports(
|
||||
severity=None, finding_class=None, target=None, search=None, include_details=False
|
||||
)
|
||||
assert unfiltered["filtered_count"] == 3
|
||||
|
||||
assert (
|
||||
_do_list_reports(
|
||||
severity=nullish,
|
||||
finding_class=nullish,
|
||||
target=nullish,
|
||||
search=nullish,
|
||||
include_details=False,
|
||||
)
|
||||
== unfiltered
|
||||
)
|
||||
|
|
|
|||
|
|
@ -101,3 +101,24 @@ def test_get_note_flags_caller_ownership() -> None:
|
|||
assert mine["note"]["agent_name"] == "Agent One"
|
||||
theirs = notes_tools._get_note_impl(note_id, caller_agent_id="agent-9")
|
||||
assert "by_you" not in theirs["note"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("nullish", ["null", "none", "NULL", " None ", "undefined", "nil"])
|
||||
def test_list_notes_ignores_nullish_filter_strings(nullish: str) -> None:
|
||||
notes_tools._create_note_impl("recon", "content", category="findings", tags=["auth"])
|
||||
notes_tools._create_note_impl("other", "content", category="general")
|
||||
|
||||
unfiltered = notes_tools._list_notes_impl()
|
||||
assert unfiltered["filtered_count"] == 2
|
||||
|
||||
assert notes_tools._list_notes_impl(category=nullish) == unfiltered
|
||||
assert notes_tools._list_notes_impl(search=nullish) == unfiltered
|
||||
assert notes_tools._list_notes_impl(tags=[nullish]) == unfiltered
|
||||
|
||||
|
||||
def test_list_notes_still_filters_on_real_values() -> None:
|
||||
notes_tools._create_note_impl("recon", "content", category="findings")
|
||||
notes_tools._create_note_impl("other", "content", category="general")
|
||||
|
||||
result = notes_tools._list_notes_impl(category="findings")
|
||||
assert [n["title"] for n in result["notes"]] == ["recon"]
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue