mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 23:47:54 +08:00
tools/chat: return shape errors for mis-typed ask-family arguments
This commit is contained in:
@@ -2,6 +2,7 @@ from __future__ import annotations
|
|||||||
|
|
||||||
from plyngent.agent import ToolTag, tool
|
from plyngent.agent import ToolTag, tool
|
||||||
from plyngent.prompting import NonInteractiveError, ask_async
|
from plyngent.prompting import NonInteractiveError, ask_async
|
||||||
|
from plyngent.tools.chat.shape import first_error, string_error
|
||||||
|
|
||||||
|
|
||||||
@tool(name="ask_user_line", tags=ToolTag.LOCAL | ToolTag.READ_ONLY)
|
@tool(name="ask_user_line", tags=ToolTag.LOCAL | ToolTag.READ_ONLY)
|
||||||
@@ -12,6 +13,12 @@ async def ask_user(question: str, default: str = "") -> str:
|
|||||||
or any input that is not a fixed menu. Optional ``default`` is used if the
|
or any input that is not a fixed menu. Optional ``default`` is used if the
|
||||||
user submits empty input (and in non-interactive mode when provided).
|
user submits empty input (and in non-interactive mode when provided).
|
||||||
"""
|
"""
|
||||||
|
shape_error = first_error(
|
||||||
|
string_error("ask_user_line", "question", question),
|
||||||
|
string_error("ask_user_line", "default", default),
|
||||||
|
)
|
||||||
|
if shape_error is not None:
|
||||||
|
return f"error: {shape_error}"
|
||||||
try:
|
try:
|
||||||
return await ask_async(question, default=default or None)
|
return await ask_async(question, default=default or None)
|
||||||
except NonInteractiveError as exc:
|
except NonInteractiveError as exc:
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ from typing import cast
|
|||||||
|
|
||||||
from plyngent.agent import ToolTag, tool
|
from plyngent.agent import ToolTag, tool
|
||||||
from plyngent.prompting import ChoiceOption, NonInteractiveError, choose_async
|
from plyngent.prompting import ChoiceOption, NonInteractiveError, choose_async
|
||||||
|
from plyngent.tools.chat.shape import first_error, list_error, string_error
|
||||||
|
|
||||||
|
|
||||||
def parse_options(options: list[object]) -> list[ChoiceOption]:
|
def parse_options(options: list[object]) -> list[ChoiceOption]:
|
||||||
@@ -51,6 +52,19 @@ async def choose_user(
|
|||||||
When ``allow_custom`` is true (default), free-text answers are accepted.
|
When ``allow_custom`` is true (default), free-text answers are accepted.
|
||||||
Returns the chosen option value (or custom text).
|
Returns the chosen option value (or custom text).
|
||||||
"""
|
"""
|
||||||
|
shape_error = first_error(
|
||||||
|
string_error("ask_user_choice", "question", question),
|
||||||
|
list_error(
|
||||||
|
"ask_user_choice",
|
||||||
|
"options",
|
||||||
|
options,
|
||||||
|
items="strings or objects with a label",
|
||||||
|
example='["alpha", "beta"]',
|
||||||
|
),
|
||||||
|
string_error("ask_user_choice", "default", default),
|
||||||
|
)
|
||||||
|
if shape_error is not None:
|
||||||
|
return f"error: {shape_error}"
|
||||||
try:
|
try:
|
||||||
parsed = parse_options([cast("object", o) for o in options])
|
parsed = parse_options([cast("object", o) for o in options])
|
||||||
except (TypeError, ValueError) as exc:
|
except (TypeError, ValueError) as exc:
|
||||||
|
|||||||
@@ -1,21 +1,28 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import json
|
import json
|
||||||
from typing import cast
|
from typing import TYPE_CHECKING, cast
|
||||||
|
|
||||||
from plyngent.agent import ToolTag, tool
|
from plyngent.agent import ToolTag, tool
|
||||||
from plyngent.prompting import FormField, NonInteractiveError, form_async
|
from plyngent.prompting import FormField, NonInteractiveError, form_async
|
||||||
from plyngent.tools.chat.choose import parse_options
|
from plyngent.tools.chat.choose import parse_options
|
||||||
|
from plyngent.tools.chat.shape import first_error, list_error, string_error
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from collections.abc import Sequence
|
||||||
|
|
||||||
|
|
||||||
def parse_fields(fields: list[dict[str, object]]) -> list[FormField]:
|
def parse_fields(fields: Sequence[object]) -> list[FormField]:
|
||||||
"""Normalize a list of field dicts into ``FormField`` list."""
|
"""Normalize a list of field dicts into ``FormField`` list."""
|
||||||
if not fields:
|
if not fields:
|
||||||
msg = "fields must be a non-empty list"
|
msg = "fields must be a non-empty list"
|
||||||
raise ValueError(msg)
|
raise ValueError(msg)
|
||||||
out: list[FormField] = []
|
out: list[FormField] = []
|
||||||
for item_obj in fields:
|
for item_obj in fields:
|
||||||
raw_map = {str(key): value for key, value in item_obj.items()}
|
if not isinstance(item_obj, dict):
|
||||||
|
msg = "each field must be an object with name/prompt"
|
||||||
|
raise TypeError(msg)
|
||||||
|
raw_map = {str(key): value for key, value in cast("dict[object, object]", item_obj).items()}
|
||||||
name = raw_map.get("name")
|
name = raw_map.get("name")
|
||||||
prompt = raw_map.get("prompt")
|
prompt = raw_map.get("prompt")
|
||||||
if not isinstance(name, str) or not name:
|
if not isinstance(name, str) or not name:
|
||||||
@@ -61,6 +68,18 @@ async def form_user(
|
|||||||
optional ``allow_custom`` (default true).
|
optional ``allow_custom`` (default true).
|
||||||
When ``confirm_submit`` is true, the human reviews a summary before submit.
|
When ``confirm_submit`` is true, the human reviews a summary before submit.
|
||||||
"""
|
"""
|
||||||
|
shape_error = first_error(
|
||||||
|
string_error("ask_user_form", "title", title),
|
||||||
|
list_error(
|
||||||
|
"ask_user_form",
|
||||||
|
"fields",
|
||||||
|
fields,
|
||||||
|
items="objects",
|
||||||
|
example='[{"name": "port", "prompt": "Which port?"}]',
|
||||||
|
),
|
||||||
|
)
|
||||||
|
if shape_error is not None:
|
||||||
|
return f"error: {shape_error}"
|
||||||
try:
|
try:
|
||||||
parsed = parse_fields(fields)
|
parsed = parse_fields(fields)
|
||||||
except (TypeError, ValueError) as exc:
|
except (TypeError, ValueError) as exc:
|
||||||
|
|||||||
@@ -0,0 +1,56 @@
|
|||||||
|
"""Shape checks for ask-family tool arguments.
|
||||||
|
|
||||||
|
Tool arguments arrive as decoded JSON, so a parameter declared ``list[str]`` can
|
||||||
|
still be a plain string at runtime. Duck-typing then misleads instead of failing:
|
||||||
|
``ask_user_choice(options="yes/no")`` iterates the characters and shows a menu of
|
||||||
|
letters, and a string ``fields`` dies inside the form parser with an
|
||||||
|
``AttributeError``. Each ask-family tool checks its container arguments up front
|
||||||
|
and answers with a model-facing error that also shows the expected shape (same
|
||||||
|
idea as :func:`plyngent.tools.workspace.argv_shape_error`).
|
||||||
|
|
||||||
|
Every check takes ``object``, so the isinstance calls stay meaningful and the
|
||||||
|
tools themselves stay free of unnecessary-isinstance warnings.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
_SHAPES: dict[type, str] = {
|
||||||
|
str: "a string",
|
||||||
|
list: "an array",
|
||||||
|
dict: "an object",
|
||||||
|
bool: "a boolean",
|
||||||
|
int: "a number",
|
||||||
|
float: "a number",
|
||||||
|
type(None): "null",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def shape_of(value: object) -> str:
|
||||||
|
"""Model-facing name for a decoded JSON value (``a string``, ``an array``…)."""
|
||||||
|
return _SHAPES.get(type(value), type(value).__name__)
|
||||||
|
|
||||||
|
|
||||||
|
def first_error(*errors: str | None) -> str | None:
|
||||||
|
"""First non-``None`` error (``None`` when every shape was acceptable)."""
|
||||||
|
return next((error for error in errors if error), None)
|
||||||
|
|
||||||
|
|
||||||
|
def string_error(tool: str, field: str, value: object) -> str | None:
|
||||||
|
"""Error when *value* must be a string but is not; ``None`` when it is."""
|
||||||
|
if isinstance(value, str):
|
||||||
|
return None
|
||||||
|
return f"`{tool}` expects `{field}` as a string, got {shape_of(value)}"
|
||||||
|
|
||||||
|
|
||||||
|
def list_error(tool: str, field: str, value: object, *, items: str, example: str) -> str | None:
|
||||||
|
"""Error when *value* must be a JSON array of *items* but is not; else ``None``.
|
||||||
|
|
||||||
|
Only the container is checked: item-level validation stays in the tool's own
|
||||||
|
parser (``parse_options`` / ``parse_fields``), which already reports what is
|
||||||
|
wrong with a bad entry.
|
||||||
|
"""
|
||||||
|
if isinstance(value, str):
|
||||||
|
return f"`{tool}` expects `{field}` as a JSON array of {items}, not a string; pass e.g. {example}"
|
||||||
|
if not isinstance(value, list):
|
||||||
|
return f"`{tool}` expects `{field}` as a JSON array of {items}, got {shape_of(value)}; pass e.g. {example}"
|
||||||
|
return None
|
||||||
@@ -192,3 +192,85 @@ def test_wait_prompt_colored_on_tty(monkeypatch) -> None:
|
|||||||
prompt = wait_module._wait_prompt(5, reason="tests")
|
prompt = wait_module._wait_prompt(5, reason="tests")
|
||||||
assert "\x1b[36m" in prompt # cyan status
|
assert "\x1b[36m" in prompt # cyan status
|
||||||
assert "\x1b[33m" in prompt # yellow input prompt
|
assert "\x1b[33m" in prompt # yellow input prompt
|
||||||
|
|
||||||
|
|
||||||
|
def test_shape_of_names_json_types() -> None:
|
||||||
|
from plyngent.tools.chat.shape import first_error, shape_of
|
||||||
|
|
||||||
|
assert shape_of("x") == "a string"
|
||||||
|
assert shape_of([]) == "an array"
|
||||||
|
assert shape_of({}) == "an object"
|
||||||
|
assert shape_of(True) == "a boolean" # bool is not reported as a number
|
||||||
|
assert shape_of(3) == "a number"
|
||||||
|
assert shape_of(3.5) == "a number"
|
||||||
|
assert shape_of(None) == "null"
|
||||||
|
assert first_error(None, "boom", "later") == "boom"
|
||||||
|
assert first_error(None, None) is None
|
||||||
|
|
||||||
|
|
||||||
|
async def test_ask_user_line_rejects_non_string_args() -> None:
|
||||||
|
"""Mis-typed args get explained instead of reaching the human as garbage."""
|
||||||
|
backend = ScriptedBackend(["unused"])
|
||||||
|
with temporary_backend(backend):
|
||||||
|
registry = ToolRegistry([ask_user])
|
||||||
|
bad_question = await registry.execute("ask_user_line", json.dumps({"question": ["a", "b"]}))
|
||||||
|
bad_default = await registry.execute(
|
||||||
|
"ask_user_line",
|
||||||
|
json.dumps({"question": "Which?", "default": 5}),
|
||||||
|
)
|
||||||
|
assert bad_question.startswith("error:")
|
||||||
|
assert "`question`" in bad_question
|
||||||
|
assert "got an array" in bad_question
|
||||||
|
assert bad_default.startswith("error:")
|
||||||
|
assert "`default`" in bad_default
|
||||||
|
assert "got a number" in bad_default
|
||||||
|
assert backend.lines == ["unused"] # the prompt never happened
|
||||||
|
|
||||||
|
|
||||||
|
async def test_choose_user_rejects_string_options() -> None:
|
||||||
|
"""``options="alpha, beta"`` used to build a menu of single characters."""
|
||||||
|
backend = ScriptedBackend(["1"])
|
||||||
|
with temporary_backend(backend):
|
||||||
|
registry = ToolRegistry([choose_user])
|
||||||
|
out = await registry.execute(
|
||||||
|
"ask_user_choice",
|
||||||
|
json.dumps({"question": "Pick", "options": "alpha, beta"}),
|
||||||
|
)
|
||||||
|
assert out.startswith("error:")
|
||||||
|
assert "`options`" in out
|
||||||
|
assert "not a string" in out
|
||||||
|
assert '["alpha", "beta"]' in out
|
||||||
|
assert backend.lines == ["1"]
|
||||||
|
|
||||||
|
|
||||||
|
async def test_choose_user_rejects_object_options() -> None:
|
||||||
|
registry = ToolRegistry([choose_user])
|
||||||
|
out = await registry.execute(
|
||||||
|
"ask_user_choice",
|
||||||
|
json.dumps({"question": "Pick", "options": {"alpha": 1}}),
|
||||||
|
)
|
||||||
|
assert out.startswith("error:")
|
||||||
|
assert "got an object" in out
|
||||||
|
|
||||||
|
|
||||||
|
async def test_form_user_rejects_string_fields() -> None:
|
||||||
|
backend = ScriptedBackend(["ncbm"], confirms=[True])
|
||||||
|
with temporary_backend(backend):
|
||||||
|
registry = ToolRegistry([form_user])
|
||||||
|
out = await registry.execute(
|
||||||
|
"ask_user_form",
|
||||||
|
json.dumps({"title": "Setup", "fields": '[{"name": "user"}]'}),
|
||||||
|
)
|
||||||
|
assert out.startswith("error:")
|
||||||
|
assert "`fields`" in out
|
||||||
|
assert "not a string" in out
|
||||||
|
assert backend.lines == ["ncbm"]
|
||||||
|
|
||||||
|
|
||||||
|
async def test_form_user_rejects_non_object_field() -> None:
|
||||||
|
registry = ToolRegistry([form_user])
|
||||||
|
out = await registry.execute(
|
||||||
|
"ask_user_form",
|
||||||
|
json.dumps({"title": "Setup", "fields": ["user"]}),
|
||||||
|
)
|
||||||
|
assert out == "error: each field must be an object with name/prompt"
|
||||||
|
|||||||
Reference in New Issue
Block a user