From 7258f975b1071aa6734803ca60b35ad047f7c1b4 Mon Sep 17 00:00:00 2001 From: Ling-Sen Peng Date: Sat, 18 Jul 2026 14:39:47 -0700 Subject: [PATCH] fix(examples): fix 9 broken examples found by catalog verification; fix worker-tool registry lookup after re-registration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Companion to the verification-report PR (#438) — every fix verified end-to-end against a live conductor-oss 3.32.0-rc.8 server. Agents suite: 91 PASS / 12 ERROR / 3 HUNG before, 96 PASS / 2 ERROR / 2 HUNG after. Example fixes: - greetings_worker.py: duplicate `def greet` shadowed the registered worker and broke spawn pickling on macOS/Windows (workers_e2e, helloworld, task_listener_example) - worker_example.py: hardcoded /Users/viren metrics dir -> tempfile - user_workers.py: import via examples.* package identity crashed spawn children that load the module as user_example.* - kitchen_sink.py: @agent classifier functions must take zero args - 16i/16j: tools defined inside factory functions are not importable by spawn workers (SpawnSafetyError) — hoisted to module level - 94_openai_runner_tools.py: keep the plain function at module level and apply function_tool() at Agent construction so the module global is not rebound to an unpicklable FunctionTool - llm_chat.py: server requires a non-empty user message (two tasks sent system-only lists); also exit non-zero when the workflow ends FAILED instead of reporting success on any terminal state - run_all_examples.py: classify 6 env-dependent examples as SKIP with reasons (dg skill repo; workflow messages API absent in rc.8) SDK fix: - tool.py _try_worker_task: registry identity check failed once ToolRegistry.register_tool_workers re-registered the task name with a spawn-safe ToolWorkerEntry wrapper; lookup now walks __wrapped__ / fn_direct / fn_ref carriers (broke 14_existing_workers.py; 2 regression tests added, tests/unit/ai: 1744 passed) Co-Authored-By: Claude Fable 5 --- examples/agentic_workflows/llm_chat.py | 31 +++++++-- examples/agents/16i_credentials_langchain.py | 21 +++--- examples/agents/16j_credentials_openai_sdk.py | 23 ++++--- examples/agents/94_openai_runner_tools.py | 7 +- examples/agents/kitchen_sink.py | 9 +-- examples/agents/run_all_examples.py | 6 ++ examples/helloworld/greetings_worker.py | 2 +- examples/user_example/user_workers.py | 2 +- examples/worker_example.py | 3 +- src/conductor/ai/agents/tool.py | 32 ++++++++- tests/unit/ai/test_tool.py | 69 +++++++++++++++++++ 11 files changed, 171 insertions(+), 34 deletions(-) diff --git a/examples/agentic_workflows/llm_chat.py b/examples/agentic_workflows/llm_chat.py index 7805dc1a3..1205a31ee 100644 --- a/examples/agentic_workflows/llm_chat.py +++ b/examples/agentic_workflows/llm_chat.py @@ -87,8 +87,14 @@ def create_chat_workflow(executor) -> ConductorWorkflow: messages=[ ChatMessage( role="system", - message="You are an expert in science. Think of a random scientific " - "discovery and create a short, interesting question about it.", + message="You are an expert in science.", + ), + # The server requires a non-empty user message — a system-only + # conversation fails validation before the first task runs. + ChatMessage( + role="user", + message="Think of a random scientific discovery and create a " + "short, interesting question about it.", ), ], temperature=0.7, @@ -121,9 +127,16 @@ def create_chat_workflow(executor) -> ConductorWorkflow: ChatMessage( role="system", message=( - "You are an expert in science. Given the context below, " + "You are an expert in science. Given the context provided, " "generate a follow-up question to dive deeper into the topic. " - "Do not repeat previous questions.\n\n" + "Do not repeat previous questions." + ), + ), + # The server requires a non-empty user message — a system-only + # conversation fails validation before the task runs. + ChatMessage( + role="user", + message=( "Context:\n${chat_complete_ref.output.result}\n\n" "Previous questions:\n" "${collect_history_ref.input.history}" @@ -198,6 +211,16 @@ def main(): printed_tasks.add(ref) time.sleep(2) + # is_completed() is true for any terminal state — verify the workflow + # actually COMPLETED rather than FAILED/TERMINATED/TIMED_OUT. + result = workflow_client.get_workflow(workflow_id=workflow_id, include_tasks=False) + if result.status != "COMPLETED": + print("=" * 70) + print(f"Workflow ended {result.status}: {result.reason_for_incompletion}") + print(f"Full execution: {api_config.ui_host}/execution/{workflow_id}") + print("=" * 70) + raise SystemExit(1) + print("=" * 70) print("Conversation complete.") print(f"Full execution: {api_config.ui_host}/execution/{workflow_id}") diff --git a/examples/agents/16i_credentials_langchain.py b/examples/agents/16i_credentials_langchain.py index 656486430..da92b982d 100644 --- a/examples/agents/16i_credentials_langchain.py +++ b/examples/agents/16i_credentials_langchain.py @@ -23,19 +23,22 @@ from settings import settings +# Tool callables must live at module level: worker processes are spawned on +# macOS/Windows and re-import this module, so a tool defined inside a factory +# function ("") cannot be resolved by qualified name (SpawnSafetyError). +def check_github_token() -> str: + """Check if GitHub token is available in the environment.""" + token = os.environ.get("GITHUB_TOKEN", "") + if token: + return f"GitHub token available (starts with {token[:4]}...)" + return "GitHub token is NOT available" + + def create_langchain_agent(): """Create a LangChain agent with a tool that uses GITHUB_TOKEN.""" from langchain.agents import create_agent from langchain_core.tools import tool as lc_tool - @lc_tool - def check_github_token() -> str: - """Check if GitHub token is available in the environment.""" - token = os.environ.get("GITHUB_TOKEN", "") - if token: - return f"GitHub token available (starts with {token[:4]}...)" - return "GitHub token is NOT available" - model_str = settings.llm_model # create_agent accepts "provider:model" format (e.g. "openai:gpt-4o") if "/" in model_str: @@ -44,7 +47,7 @@ def check_github_token() -> str: agent = create_agent( model_str, - tools=[check_github_token], + tools=[lc_tool(check_github_token)], system_prompt="You are a helpful assistant. Use tools when asked.", ) return agent diff --git a/examples/agents/16j_credentials_openai_sdk.py b/examples/agents/16j_credentials_openai_sdk.py index d35382bb6..e1b15e05a 100644 --- a/examples/agents/16j_credentials_openai_sdk.py +++ b/examples/agents/16j_credentials_openai_sdk.py @@ -22,22 +22,27 @@ from conductor.ai.agents import AgentRuntime +# Tool callables must live at module level: worker processes are spawned on +# macOS/Windows and re-import this module, so a tool defined inside a factory +# function ("") cannot be resolved by qualified name (SpawnSafetyError). +# Keep the plain function importable and apply @function_tool at agent +# construction, so the module global is not rebound to a FunctionTool. +def check_github_auth() -> str: + """Check if GitHub authentication is available.""" + token = os.environ.get("GITHUB_TOKEN", "") + if token: + return f"GitHub token is set (starts with {token[:4]}...)" + return "GitHub token is NOT set" + + def create_openai_agent(): """Create an OpenAI Agent SDK agent with a credential-aware tool.""" from agents import Agent, function_tool - @function_tool - def check_github_auth() -> str: - """Check if GitHub authentication is available.""" - token = os.environ.get("GITHUB_TOKEN", "") - if token: - return f"GitHub token is set (starts with {token[:4]}...)" - return "GitHub token is NOT set" - agent = Agent( name="github_checker", instructions="You check GitHub authentication status. Use the tool when asked.", - tools=[check_github_auth], + tools=[function_tool(check_github_auth)], ) return agent diff --git a/examples/agents/94_openai_runner_tools.py b/examples/agents/94_openai_runner_tools.py index 27b4999f0..977aaf9e0 100644 --- a/examples/agents/94_openai_runner_tools.py +++ b/examples/agents/94_openai_runner_tools.py @@ -49,7 +49,10 @@ class Weather(BaseModel): conditions: str = Field(description="The weather conditions") -@function_tool +# Keep the plain function importable at module level and apply +# @function_tool at Agent construction: worker processes are spawned on +# macOS/Windows and re-import this module by qualified name, which fails +# when the decorator has rebound the module global to a FunctionTool. def get_weather(city: Annotated[str, "The city to get the weather for"]) -> Weather: """Get the current weather information for a specified city.""" print("[debug] get_weather called") @@ -59,7 +62,7 @@ def get_weather(city: Annotated[str, "The city to get the weather for"]) -> Weat agent = Agent( name="weather_agent", instructions="You are a helpful agent.", - tools=[get_weather], + tools=[function_tool(get_weather)], ) diff --git a/examples/agents/kitchen_sink.py b/examples/agents/kitchen_sink.py index 8eda28e85..56248be65 100644 --- a/examples/agents/kitchen_sink.py +++ b/examples/agents/kitchen_sink.py @@ -157,21 +157,18 @@ @agent(name="tech_classifier", model=settings.llm_model) -def tech_classifier(prompt: str) -> str: +def tech_classifier() -> str: """Classifies tech articles.""" - pass @agent(name="business_classifier", model=settings.llm_model) -def business_classifier(prompt: str) -> str: +def business_classifier() -> str: """Classifies business articles.""" - pass @agent(name="creative_classifier", model=settings.llm_model) -def creative_classifier(prompt: str) -> str: +def creative_classifier() -> str: """Classifies creative articles.""" - pass intake_router = Agent( diff --git a/examples/agents/run_all_examples.py b/examples/agents/run_all_examples.py index e25d9b260..74bc84b0b 100644 --- a/examples/agents/run_all_examples.py +++ b/examples/agents/run_all_examples.py @@ -64,6 +64,12 @@ } # Require external infra / providers not available in this environment. INFRA_SKIP = { + "30_skills_dg_review.py": "needs the dg skill cloned to ~/.claude/skills/dg", + "32_skills_multi_agent.py": "needs the dg skill cloned to ~/.claude/skills/dg", + "75_wait_for_message.py": "needs the workflow messages API (not in conductor-oss 3.32.0-rc.8)", + "82_fan_out_fan_in.py": "needs the workflow messages API (not in conductor-oss 3.32.0-rc.8)", + "83_stateful_resume.py": "needs the workflow messages API (not in conductor-oss 3.32.0-rc.8)", + "84_deterministic_stop.py": "needs the workflow messages API (not in conductor-oss 3.32.0-rc.8)", "04_mcp_weather.py": "needs an MCP server", "04_http_and_mcp_tools.py": "needs an MCP server", "16f_credentials_mcp_tool.py": "needs an MCP server", diff --git a/examples/helloworld/greetings_worker.py b/examples/helloworld/greetings_worker.py index 44d8b5b61..5ce808ce0 100644 --- a/examples/helloworld/greetings_worker.py +++ b/examples/helloworld/greetings_worker.py @@ -21,7 +21,7 @@ def greet(name: str) -> str: poll_timeout=100, # Default poll timeout (ms) lease_extend_enabled=False # Fast tasks don't need lease extension ) -def greet(name: str) -> str: +def greet_sync(name: str) -> str: """ Synchronous worker - automatically runs in thread pool to avoid blocking. Good for legacy code or simple CPU-bound tasks. diff --git a/examples/user_example/user_workers.py b/examples/user_example/user_workers.py index 89af91592..147594679 100644 --- a/examples/user_example/user_workers.py +++ b/examples/user_example/user_workers.py @@ -8,7 +8,7 @@ from conductor.client.context import get_task_context from conductor.client.worker.worker_task import worker_task -from examples.user_example.models import User +from user_example.models import User @worker_task( diff --git a/examples/worker_example.py b/examples/worker_example.py index 7242cf6fe..e42499783 100644 --- a/examples/worker_example.py +++ b/examples/worker_example.py @@ -30,6 +30,7 @@ import logging import os import shutil +import tempfile import time from typing import Union @@ -313,7 +314,7 @@ def main(): api_config = Configuration() # Metrics configuration - HTTP mode (recommended) - metrics_dir = os.path.join('/Users/viren/', 'conductor_metrics') + metrics_dir = os.path.join(tempfile.gettempdir(), 'conductor_metrics') # Clean up any stale metrics data from previous runs if os.path.exists(metrics_dir): diff --git a/src/conductor/ai/agents/tool.py b/src/conductor/ai/agents/tool.py index 14c91d904..cde9565f2 100644 --- a/src/conductor/ai/agents/tool.py +++ b/src/conductor/ai/agents/tool.py @@ -1159,6 +1159,36 @@ def agent_tool( # ── Utilities ─────────────────────────────────────────────────────────── +def _entry_resolves_to(entry_func: Any, original: Callable[..., Any], + func: Callable[..., Any]) -> bool: + """True if a ``_decorated_functions`` entry refers to *original*/*func*. + + A plain identity check is not enough: once a ``@worker_task`` function is + used as an agent tool, ``ToolRegistry.register_tool_workers`` re-registers + the same task name with a spawn-safe ``ToolWorkerEntry`` wrapper, which + carries the original either directly (``fn_direct``) or by qualified name + (``fn_ref``). Walk the ``__wrapped__`` chain and those carriers. + """ + obj = entry_func + for _ in range(8): # bounded — wrapper chains are shallow + if obj is None: + return False + if obj is original or obj is func: + return True + fn_direct = getattr(obj, "fn_direct", None) + if fn_direct is not None and (fn_direct is original or fn_direct is func): + return True + fn_ref = getattr(obj, "fn_ref", None) + if ( + fn_ref is not None + and getattr(fn_ref, "module", None) == getattr(original, "__module__", None) + and getattr(fn_ref, "qualname", None) == getattr(original, "__qualname__", None) + ): + return True + obj = getattr(obj, "__wrapped__", None) + return False + + def _try_worker_task(func: Callable[..., Any]) -> Optional[ToolDef]: """Try to build a :class:`ToolDef` from a ``@worker_task``-decorated function. @@ -1174,7 +1204,7 @@ def _try_worker_task(func: Callable[..., Any]) -> Optional[ToolDef]: original = getattr(func, "__wrapped__", func) for (task_name, _domain), entry in _decorated_functions.items(): - if entry["func"] is original or entry["func"] is func: + if _entry_resolves_to(entry["func"], original, func): from conductor.ai.agents._internal.schema_utils import schema_from_function description = inspect.getdoc(original) or "" diff --git a/tests/unit/ai/test_tool.py b/tests/unit/ai/test_tool.py index 74173b4c5..36cd5d5d1 100644 --- a/tests/unit/ai/test_tool.py +++ b/tests/unit/ai/test_tool.py @@ -360,6 +360,75 @@ def some_func(x: str) -> str: with pytest.raises(TypeError): get_tool_def(some_func) + def test_worker_task_detected_after_tool_registry_reregistration(self): + """Detection must survive ToolRegistry re-registering the task name. + + When a @worker_task function is used as an agent tool, + register_tool_workers() overwrites the _decorated_functions entry + with a spawn-safe ToolWorkerEntry wrapper (wrapped again by + @worker_task). get_tool_def() on the original function must still + resolve — a second runtime.run() with the same agent hits this path. + """ + import functools + + registry = {} + wrapper, original = self._make_worker_task_func(registry) + + class FakeToolWorkerEntry: + """Stands in for ToolWorkerEntry: carries the original via fn_direct.""" + + def __init__(self, fn): + self.fn_direct = fn + self.fn_ref = None + + def __call__(self, *args, **kwargs): + return self.fn_direct(*args, **kwargs) + + entry = FakeToolWorkerEntry(original) + + @functools.wraps(original) + def reregistered(*args, **kwargs): + return entry(*args, **kwargs) + + reregistered.__wrapped__ = entry # @worker_task's wraps() points at the entry + registry[("get_customer_data", None)] = {"func": reregistered} + + with mock.patch( + "conductor.client.automator.task_handler._decorated_functions", + registry, + ): + td = get_tool_def(wrapper) + + assert td.name == "get_customer_data" + assert td.tool_type == "worker" + assert td.func is original + + def test_worker_task_detected_via_fn_ref_qualname(self): + """Detection falls back to fn_ref module+qualname when identity is lost.""" + registry = {} + wrapper, original = self._make_worker_task_func(registry) + + class FakeRef: + def __init__(self, fn): + self.module = fn.__module__ + self.qualname = fn.__qualname__ + + class FakeToolWorkerEntry: + def __init__(self, fn): + self.fn_direct = None + self.fn_ref = FakeRef(fn) + + registry[("get_customer_data", None)] = {"func": FakeToolWorkerEntry(original)} + + with mock.patch( + "conductor.client.automator.task_handler._decorated_functions", + registry, + ): + td = get_tool_def(wrapper) + + assert td.name == "get_customer_data" + assert td.func is original + class TestExternalTool: """Test @tool(external=True) for referencing external workers."""