mirror of
https://hubproxy.babadafafafafa.cn/https://github.com/usestrix/strix.git
synced 2026-09-20 16:13:44 +08:00
Compare commits
2 Commits
devin/null
...
agent/pres
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d0768a5c54 | ||
|
|
187f41f36f |
@@ -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]
|
||||
|
||||
@@ -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()}
|
||||
|
||||
@@ -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
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,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)
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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,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"]
|
||||
|
||||
Reference in New Issue
Block a user