refactor(tools): todo/tool_search — fold validate/normalize guards, dedupe key expression, arguments-null default
This commit is contained in:
@@ -82,14 +82,13 @@ class TodoStore:
|
||||
self._items = self._normalize_order(rebuilt)
|
||||
|
||||
def read(self) -> List[Dict[str, str]]:
|
||||
"""Return a copy of the current list."""
|
||||
return [item.copy() for item in self._items]
|
||||
|
||||
def has_items(self) -> bool:
|
||||
return bool(self._items)
|
||||
|
||||
def snapshot(self) -> Dict[str, Any]:
|
||||
"""Return the full state clients can reconcile atomically."""
|
||||
"""Full state clients can reconcile atomically."""
|
||||
return {"todos": self.read(), "revision": self._revision}
|
||||
|
||||
def restore(self, todos: List[Dict[str, Any]], *, revision: Any = 0) -> List[Dict[str, str]]:
|
||||
@@ -148,11 +147,10 @@ class TodoStore:
|
||||
return {"id": "?", "content": "(invalid item)", "status": "pending"}
|
||||
item_id = str(item.get("id", "")).strip() or "?"
|
||||
content = str(item.get("content", "")).strip()
|
||||
content = TodoStore._cap_content(content) if content else "(no description)"
|
||||
status = str(item.get("status", "pending")).strip().lower()
|
||||
if status not in VALID_STATUSES:
|
||||
status = "pending"
|
||||
result = {"id": item_id, "content": content, "status": status}
|
||||
result = {"id": item_id,
|
||||
"content": TodoStore._cap_content(content) if content else "(no description)",
|
||||
"status": status if status in VALID_STATUSES else "pending"}
|
||||
parent = str(item.get("parent") or "").strip()
|
||||
if parent and parent != item_id:
|
||||
result["parent"] = parent
|
||||
@@ -166,8 +164,7 @@ class TodoStore:
|
||||
if item.get("parent") and item["parent"] not in by_id:
|
||||
item.pop("parent", None)
|
||||
for item in items:
|
||||
seen = {item["id"]}
|
||||
node = item
|
||||
seen, node = {item["id"]}, item
|
||||
while node.get("parent"):
|
||||
if node["parent"] in seen:
|
||||
item.pop("parent", None)
|
||||
@@ -179,21 +176,17 @@ class TodoStore:
|
||||
def _dedupe_by_id(todos: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
||||
"""Collapse duplicate ids, keeping the last occurrence in its position."""
|
||||
last_index: Dict[str, int] = {}
|
||||
for i, item in enumerate(todos):
|
||||
if not isinstance(item, dict): # synthetic key so _validate can handle them
|
||||
last_index[f"__invalid_{i}"] = i
|
||||
continue
|
||||
last_index[str(item.get("id", "")).strip() or "?"] = i
|
||||
for i, item in enumerate(todos): # non-dicts get a synthetic key; _validate handles them
|
||||
key = str(item.get("id", "")).strip() if isinstance(item, dict) else f"__invalid_{i}"
|
||||
last_index[key or "?"] = i
|
||||
return [todos[i] for i in sorted(last_index.values())]
|
||||
|
||||
@staticmethod
|
||||
def _normalize_order(items: List[Dict[str, str]]) -> List[Dict[str, str]]:
|
||||
"""Lift the in_progress step ahead of any earlier pending placeholder. Nested lists
|
||||
keep authored order — reordering would tear a subtask from its siblings."""
|
||||
if any(item.get("parent") for item in items):
|
||||
return items
|
||||
statuses = [item["status"] for item in items]
|
||||
if "in_progress" not in statuses:
|
||||
if any(item.get("parent") for item in items) or "in_progress" not in statuses:
|
||||
return items
|
||||
active_index = statuses.index("in_progress")
|
||||
if "pending" not in statuses[:active_index]:
|
||||
|
||||
@@ -73,8 +73,8 @@ _TRI_STATE_ALIASES = {"true": "on", "1": "on", "yes": "on", "false": "off", "0":
|
||||
|
||||
def _tri_state(value: Any) -> str:
|
||||
"""Normalize an ``auto``/``on``/``off`` setting (bool-ish aliases accepted)."""
|
||||
text = _TRI_STATE_ALIASES.get(str(value).strip().lower(), str(value).strip().lower())
|
||||
return text if text in ("auto", "on", "off") else "auto"
|
||||
text = str(value).strip().lower()
|
||||
return _TRI_STATE_ALIASES.get(text, text if text in ("auto", "on", "off") else "auto")
|
||||
|
||||
|
||||
def _clamped_int(value: Any, fallback: int, lo: int, hi: int) -> int:
|
||||
@@ -109,8 +109,7 @@ def load_config() -> ToolSearchConfig:
|
||||
return _config_from_loader("load_config")
|
||||
|
||||
|
||||
def load_config_readonly() -> ToolSearchConfig:
|
||||
"""Same as ``load_config`` without copying the cached full config."""
|
||||
def load_config_readonly() -> ToolSearchConfig: # no copy of the cached full config
|
||||
return _config_from_loader("load_config_readonly")
|
||||
|
||||
|
||||
@@ -497,13 +496,12 @@ def resolve_underlying_call(args: Dict[str, Any]) -> Tuple[Optional[str], Dict[s
|
||||
if name in BRIDGE_TOOL_NAMES:
|
||||
return None, {}, f"tool_call cannot invoke '{name}' (it is itself a bridge tool)"
|
||||
raw_args = args.get("arguments")
|
||||
if raw_args is None:
|
||||
raw_args = {}
|
||||
if isinstance(raw_args, str):
|
||||
try:
|
||||
raw_args = json.loads(raw_args)
|
||||
except json.JSONDecodeError as e:
|
||||
return None, {}, f"tool_call 'arguments' is not valid JSON: {e}"
|
||||
raw_args = {} if raw_args is None else raw_args
|
||||
if not isinstance(raw_args, dict):
|
||||
return None, {}, "tool_call 'arguments' must be an object"
|
||||
if not is_deferrable_tool_name(name, load_config_readonly().effective_defer_tools):
|
||||
|
||||
Reference in New Issue
Block a user