Private
Public Access
0
0

refactor(ai_client): migrate set_tool_preset + set_bias_profile to Result[T] (Phase 11 sites 5+6)

Both functions had:
  try: ToolPresetManager().load_all() ...
  except (OSError, ValueError, AttributeError) as e:
      sys.stderr.write(f'[ERROR] Failed to set {preset_name}: {e}')
      sys.stderr.flush()

sys.stderr.write is logging = NOT a drain = SS violation per MUST-NOT-DO #6.

New helpers:
- _set_tool_preset_result(preset_name: Optional[str]) -> Result[None]
  Empty/None preset short-circuits to Result(data=None).
  On failure: Result(data=None, errors=[ErrorInfo]).
- _set_bias_profile_result(profile_name: Optional[str]) -> Result[None]
  Same pattern.

Legacy wrappers set the global state (or skip on empty preset) and
delegate to the _result helper. Cache invalidation runs regardless.

Audit: ai_client SS 9 -> 7. COMPLIANT 27 -> 29.
This commit is contained in:
2026-06-20 13:59:45 -04:00
parent fb7014cd63
commit 343b855a0f
2 changed files with 96 additions and 25 deletions
+55 -25
View File
@@ -503,6 +503,38 @@ def set_agent_tools(tools: dict[str, bool]) -> None:
_CACHED_ANTHROPIC_TOOLS = None
_CACHED_DEEPSEEK_TOOLS = None
def _set_tool_preset_result(preset_name: Optional[str]) -> Result[None]:
"""Load a tool preset by name and apply it. Returns Result[None].
On I/O or parsing failure, returns Result(data=None, errors=[ErrorInfo])
capturing the original exception. The legacy caller (set_tool_preset)
calls this helper for the load step; on Result errors, the caller still
completes (state remains partially-set; the cache invalidation runs).
"""
if not preset_name or preset_name == "None":
return Result(data=None)
try:
manager = ToolPresetManager()
presets = manager.load_all()
if preset_name in presets:
preset = presets[preset_name]
_active_tool_preset = preset
new_tools = {name: False for name in mcp_client.TOOL_NAMES}
new_tools[TOOL_NAME] = False
for cat in preset.categories.values():
for tool in cat:
name = tool.name
new_tools[name] = True
_tool_approval_modes[name] = tool.approval
_agent_tools = new_tools
return Result(data=None)
except (OSError, ValueError, AttributeError) as e:
return Result(
data=None,
errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=f"failed to set tool preset '{preset_name}': {e}", source="ai_client._set_tool_preset_result", original=e)],
)
def set_tool_preset(preset_name: Optional[str]) -> None:
"""Loads a tool preset and applies it via set_agent_tools."""
global _agent_tools, _CACHED_ANTHROPIC_TOOLS, _CACHED_DEEPSEEK_TOOLS, _tool_approval_modes, _active_tool_preset
@@ -513,40 +545,38 @@ def set_tool_preset(preset_name: Optional[str]) -> None:
_agent_tools[TOOL_NAME] = True
_active_tool_preset = None
else:
try:
manager = ToolPresetManager()
presets = manager.load_all()
if preset_name in presets:
preset = presets[preset_name]
_active_tool_preset = preset
new_tools = {name: False for name in mcp_client.TOOL_NAMES}
new_tools[TOOL_NAME] = False
for cat in preset.categories.values():
for tool in cat:
name = tool.name
new_tools[name] = True
_tool_approval_modes[name] = tool.approval
_agent_tools = new_tools
except (OSError, ValueError, AttributeError) as e:
sys.stderr.write(f"[ERROR] Failed to set tool preset '{preset_name}': {e}\n")
sys.stderr.flush()
_set_tool_preset_result(preset_name)
_CACHED_ANTHROPIC_TOOLS = None
_CACHED_DEEPSEEK_TOOLS = None
def _set_bias_profile_result(profile_name: Optional[str]) -> Result[None]:
"""Load a bias profile by name and apply it. Returns Result[None].
On I/O or parsing failure, returns Result(data=None, errors=[ErrorInfo]).
The legacy caller (set_bias_profile) delegates to this helper.
"""
if not profile_name or profile_name == "None":
return Result(data=None)
try:
manager = ToolPresetManager()
profiles = manager.load_all_bias_profiles()
if profile_name in profiles:
_active_bias_profile = profiles[profile_name]
return Result(data=None)
except (OSError, ValueError, AttributeError) as e:
return Result(
data=None,
errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=f"failed to set bias profile '{profile_name}': {e}", source="ai_client._set_bias_profile_result", original=e)],
)
def set_bias_profile(profile_name: Optional[str]) -> None:
"""Sets the active tool bias profile for tuning model behavior."""
global _active_bias_profile
if not profile_name or profile_name == "None":
_active_bias_profile = None
else:
try:
manager = ToolPresetManager()
profiles = manager.load_all_bias_profiles()
if profile_name in profiles:
_active_bias_profile = profiles[profile_name]
except (OSError, ValueError, AttributeError) as e:
sys.stderr.write(f"[ERROR] Failed to set bias profile '{profile_name}': {e}\n")
sys.stderr.flush()
_set_bias_profile_result(profile_name)
def get_bias_profile() -> Optional[str]:
"""Returns the name of the currently active bias profile."""
+41
View File
@@ -0,0 +1,41 @@
"""Phase 11 sites 5+6: set_tool_preset + set_bias_profile Result helpers.
Both had:
try: ToolPresetManager().load_all() ...
except (OSError, ValueError, AttributeError) as e:
sys.stderr.write(f'[ERROR] Failed to set {preset_name}: {e}')
sys.stderr.flush()
Body is sys.stderr.write = logging NOT a drain = SS violation.
MIGRATE to Result[None].
"""
import sys
sys.path.insert(0, ".")
def test_phase11_sites56_set_tool_preset_result_exists():
import src.ai_client
assert hasattr(src.ai_client, "_set_tool_preset_result"), \
"_set_tool_preset_result helper missing"
def test_phase11_sites56_set_bias_profile_result_exists():
import src.ai_client
assert hasattr(src.ai_client, "_set_bias_profile_result"), \
"_set_bias_profile_result helper missing"
def test_phase11_sites56_helpers_return_result():
import src.ai_client
import inspect
for name in ("_set_tool_preset_result", "_set_bias_profile_result"):
fn = getattr(src.ai_client, name)
sig = inspect.signature(fn)
assert "Result" in str(sig.return_annotation), \
f"{name} return must be Result, got {sig.return_annotation}"
def test_phase11_sites56_legacy_preserved():
import src.ai_client
assert callable(getattr(src.ai_client, "set_tool_preset", None))
assert callable(getattr(src.ai_client, "set_bias_profile", None))