Compare commits

...

2 Commits

Author SHA1 Message Date
bearsyankees
d0768a5c54 fix(config): preserve file overrides on startup 2026-08-25 17:34:30 -04:00
alex s
187f41f36f Treat literal 'null'/'none' strings as absent for optional tool args (#1164)
* 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.

* Limit nullish coercion to query tools and keep literal tags

A literal "null"/"none" is only a mistake where the argument is a filter, so
gate the coercion on read-only query tools; a tool that writes keeps the value,
which stops update_note(content="none") from being read as "leave unchanged".

Stop dropping nullish entries from a notes tag filter too: tags are free-form,
so a literal "none" tag stays filterable and mixed tag queries keep every
branch.
2026-08-25 13:21:12 -04:00
10 changed files with 292 additions and 30 deletions

View File

@@ -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)
@@ -189,7 +217,12 @@ def _coerce_argument(value: Any, spec: dict[str, Any]) -> Any:
return value
def _coerce_arguments(raw_input: str, schema: dict[str, Any]) -> str:
# Only query tools get nullish coercion: there a literal "null" is a filter that
# matches nothing, while a tool that writes may well be given it as real content.
_QUERY_TOOL_PREFIXES = ("list_", "search_", "view_", "get_")
def _coerce_arguments(raw_input: str, schema: dict[str, Any], *, nullish: bool = False) -> str:
properties = schema.get("properties")
if not isinstance(properties, dict) or not properties:
return raw_input
@@ -205,7 +238,9 @@ 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=nullish and _is_nullable(key, spec, schema)
)
if coerced is not value:
payload[key] = coerced
changed = True
@@ -220,9 +255,10 @@ def _with_coerced_arguments(tool: FunctionTool) -> FunctionTool:
return tool
invoke_tool = tool.on_invoke_tool
schema = tool.params_json_schema
nullish = tool.name.startswith(_QUERY_TOOL_PREFIXES)
async def invoke(ctx: Any, raw_input: str) -> Any:
return await invoke_tool(ctx, _coerce_arguments(raw_input, schema))
return await invoke_tool(ctx, _coerce_arguments(raw_input, schema, nullish=nullish))
tool.on_invoke_tool = invoke
tool._strix_coerced = True # type: ignore[attr-defined]

View File

@@ -54,21 +54,29 @@ def apply_config_override(path: Path) -> None:
def persist_current() -> None:
"""Write currently-set env vars to the active config file (0o600)."""
"""Merge currently-set env vars into the active config file (0o600)."""
s = load_settings()
target = _override or _DEFAULT_PATH
target.parent.mkdir(parents=True, exist_ok=True)
env_block: dict[str, str] = {}
for sub_name in s.model_fields:
env_block = {
key: value
for key, value in _read_env_block(target).items()
if isinstance(value, str)
}
process_env = {key.upper(): value for key, value in os.environ.items()}
for sub_name in type(s).model_fields:
sub_model = getattr(s, sub_name)
if not isinstance(sub_model, BaseModel):
continue
for finfo in type(sub_model).model_fields.values():
for alias in _aliases_for(finfo):
value = os.environ.get(alias.upper())
aliases = [alias.upper() for alias in _aliases_for(finfo)]
for alias in aliases:
value = process_env.get(alias)
if value:
env_block[alias.upper()] = value
for sibling_alias in aliases:
env_block.pop(sibling_alias, None)
env_block[alias] = value
break
write_secret_text(target, json.dumps({"env": env_block}, indent=2))
@@ -93,17 +101,7 @@ def _read_json_overrides(path: Path) -> dict[str, dict[str, Any]]:
Only includes keys whose env var is NOT already set, so env always
wins over the persisted file.
"""
if not path.exists():
return {}
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
return {}
env_block = data.get("env", {}) if isinstance(data, dict) else {}
if not isinstance(env_block, dict):
return {}
env_block_upper = {str(k).upper(): v for k, v in env_block.items()}
env_block_upper = _read_env_block(path)
env_present = {k.upper() for k in os.environ}
nested: dict[str, dict[str, Any]] = {}
@@ -123,3 +121,17 @@ def _read_json_overrides(path: Path) -> dict[str, dict[str, Any]]:
if sub_data:
nested[sub_name] = sub_data
return nested
def _read_env_block(path: Path) -> dict[str, Any]:
"""Return a config file's environment block with normalized keys."""
if not path.exists():
return {}
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
return {}
env_block = data.get("env", {}) if isinstance(data, dict) else {}
if not isinstance(env_block, dict):
return {}
return {str(key).upper(): value for key, value in env_block.items()}

View File

@@ -14,6 +14,8 @@ from typing import Any
from agents import RunContextWrapper, function_tool
from strix.tools.nullish import clean_optional
logger = logging.getLogger(__name__)
@@ -111,6 +113,9 @@ 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)
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
View 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

View File

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

View File

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

View File

@@ -9,24 +9,30 @@ 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:
def _capturing_tool(
captured: dict[str, str], schema: dict[str, Any], name: str = "probe"
) -> FunctionTool:
async def invoke(_ctx: Any, raw_input: str) -> str:
captured["raw_input"] = raw_input
return "ok"
return FunctionTool(
name="probe",
name=name,
description="test tool",
params_json_schema={"type": "object", "properties": schema},
on_invoke_tool=invoke,
)
async def _roundtrip(schema: dict[str, Any], payload: dict[str, Any]) -> dict[str, Any]:
async def _roundtrip(
schema: dict[str, Any], payload: dict[str, Any], name: str = "probe"
) -> dict[str, Any]:
captured: dict[str, str] = {}
wrapped = factory._with_coerced_arguments(_capturing_tool(captured, schema))
wrapped = factory._with_coerced_arguments(_capturing_tool(captured, schema, name))
assert await wrapped.on_invoke_tool(cast("Any", None), json.dumps(payload)) == "ok"
return cast("dict[str, Any]", json.loads(captured["raw_input"]))
@@ -144,3 +150,88 @@ 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_CONTENT = {"content": {"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}, "list_probes")
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, "list_probes")
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"}}, "list_probes")
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"}, "list_probes")
assert parsed["todos"] == "none"
@pytest.mark.asyncio
@pytest.mark.parametrize("name", ["update_note", "create_note", "record_coverage"])
@pytest.mark.parametrize("value", ["null", "none"])
async def test_a_nullish_value_survives_on_a_tool_that_writes(name: str, value: str) -> None:
parsed = await _roundtrip(_NULLABLE_CONTENT, {"content": value}, name)
assert parsed["content"] == value
@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"}, "list_probes"
)
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"}, "list_probes")
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)

View File

@@ -208,6 +208,40 @@ def test_persist_current_writes_env_block(tmp_path: Path, monkeypatch: pytest.Mo
}
def test_persist_current_preserves_file_only_values(tmp_path: Path) -> None:
target = tmp_path / "cli-config.json"
expected = {
"env": {
"STRIX_LLM": "openrouter/z-ai/glm-5.3",
"LLM_API_KEY": "sk-from-file",
}
}
target.write_text(json.dumps(expected), encoding="utf-8")
loader.apply_config_override(target)
loader.persist_current()
assert json.loads(target.read_text(encoding="utf-8")) == expected
def test_persist_current_replaces_stale_alias_from_environment(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
target = tmp_path / "cli-config.json"
target.write_text(
json.dumps({"env": {"LLM_API_KEY": "sk-from-file"}}),
encoding="utf-8",
)
monkeypatch.setenv("OPENAI_API_KEY", "sk-from-env")
loader.apply_config_override(target)
loader.persist_current()
assert json.loads(target.read_text(encoding="utf-8")) == {
"env": {"OPENAI_API_KEY": "sk-from-env"}
}
def test_persist_current_sets_0600_mode(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("STRIX_LLM", "persisted-model")
target = tmp_path / "cli-config.json"

View File

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

View File

@@ -101,3 +101,33 @@ 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
@pytest.mark.parametrize("tag", ["null", "none"])
def test_list_notes_filters_on_a_literal_nullish_tag(tag: str) -> None:
notes_tools._create_note_impl("tagged", "content", tags=[tag])
notes_tools._create_note_impl("other", "content", tags=["auth"])
assert [n["title"] for n in notes_tools._list_notes_impl(tags=[tag])["notes"]] == ["tagged"]
mixed = notes_tools._list_notes_impl(tags=[tag, "auth"])
assert sorted(n["title"] for n in mixed["notes"]) == ["other", "tagged"]
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"]