Coverage for core / labeled_autogen_function.py: 23.1%

13 statements  

« prev     ^ index     » next       coverage.py v7.14.0, created at 2026-05-12 04:49 +0000

1"""Mandatory-UI-label autogen registration chokepoint (#508/#509). 

2 

3`autogen.register_function(func, caller=..., executor=..., ...)` stores 

4the raw Python callable in the executor's `function_map`. When autogen 

5later invokes the tool, it bypasses LangChain's `_with_tool_logging` 

6wrapper, so per-tool UI status events (`publish_chat_stage('tool_call', 

7…)`) never fire and the chat spinner stays on the generic "Thinking…" 

8verb during autogen turns. 

9 

10This module closes that gap: 

11 

12 1. Requires a `ui_label` kwarg at registration time (TypeError if 

13 omitted, ValueError if empty) — compile-time enforcement replaces 

14 an AST drift-guard. 

15 

16 2. Registers the label into the canonical `TOOL_LABELS` dict via 

17 `core.constants.register_tool_label`. 

18 

19 3. Wraps the function with the canonical 

20 `core.tool_logging.log_tool_execution` decorator, which: 

21 - Branches sync vs async via `inspect.iscoroutinefunction(func)`. 

22 - Emits the per-tool UI status BEFORE invocation. 

23 - Logs entry, args, success result, errors (with traceback). 

24 - Coerces non-string returns to str. 

25 - Returns a structured JSON error envelope on exception (autogen 

26 friendly — the LLM sees the failure as a normal tool output). 

27 

28 4. Passes the wrapped function to `autogen.register_function` with 

29 identical `caller`/`executor`/`name`/`description` kwargs. 

30 

31Per CLAUDE.md Gate 4 (no parallel paths) this is THIN by design: 

32all the wrap-and-log logic lives in `core.tool_logging`, shared with 

33the ~65 decorator sites in `create_recipe.py`, `reuse_recipe.py`, 

34`core/agent_tools.py`, and `integrations/channels/agent_tools.py`. 

35""" 

36from __future__ import annotations 

37 

38from typing import Any, Callable 

39 

40from autogen import register_function as _autogen_register_function 

41 

42from core.constants import register_tool_label 

43from core.tool_logging import log_tool_execution 

44 

45 

46def register_labeled_function( 

47 func: Callable[..., Any], 

48 *, 

49 caller: Any, 

50 executor: Any, 

51 description: str | None = None, 

52 ui_label: str, 

53 name: str | None = None, 

54) -> Any: 

55 """Register `func` with autogen, wrapping it via the canonical 

56 `core.tool_logging.log_tool_execution` decorator so the UI spinner 

57 shows tool-specific status text and the tool gets full structured 

58 logging. 

59 

60 Args: 

61 func: The autogen tool function (sync or async). 

62 caller: autogen agent that asks for the tool to be invoked. 

63 executor: autogen agent that runs the function (its function_map 

64 holds the wrapped func). 

65 description: LLM-facing description; defaults to func.__doc__. 

66 ui_label: REQUIRED user-facing spinner text shown when this tool 

67 fires. Same enforcement as `core.labeled_tool.labeled_tool`. 

68 name: Optional explicit tool name (defaults to func.__name__). 

69 

70 Raises: 

71 TypeError: if `ui_label` is omitted (Python kwarg enforcement). 

72 ValueError: if `ui_label` is empty or non-string. 

73 

74 Returns: 

75 autogen.register_function's return value. 

76 """ 

77 if not isinstance(ui_label, str) or not ui_label.strip(): 

78 raise ValueError( 

79 f"register_labeled_function({func.__name__!r}): ui_label must " 

80 f"be a non-empty string; pass generic_autogen_label({func.__name__!r}) " 

81 f"to opt into the 'Running …' fallback" 

82 ) 

83 

84 tool_name = name or func.__name__ 

85 register_tool_label(tool_name, ui_label) 

86 

87 # Canonical chokepoint — preserves the sync/async branch, structured 

88 # error envelope, str-coercion, and publish_chat_stage emit. All 

89 # behavior lives in one place (CLAUDE.md Gate 4: no parallel paths). 

90 wrapped = log_tool_execution(func) 

91 

92 return _autogen_register_function( 

93 wrapped, 

94 caller=caller, 

95 executor=executor, 

96 name=tool_name, 

97 description=description or func.__doc__, 

98 ) 

99 

100 

101def generic_autogen_label(name: str) -> str: 

102 """Explicit opt-in to the 'Running {name}…' fallback for autogen 

103 tools whose name is already self-descriptive. Use sparingly — a 

104 real verb phrase is more polished.""" 

105 return f'Running {name}…'