Merge pull request #439 from conductor-oss/examples/verification-fixes

fix(examples): fix 9 broken examples + worker-tool registry lookup (companion to #438)

Author: ling-senpeng13

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}")

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 ("<locals>") 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

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 ("<locals>") 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
 

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)],
 )
 
 

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(

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",

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.

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(

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):

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 ""