Merge pull request #494 from conductor-oss/fix/wmq-examples-2

Fix WMQ examples (78-84)

Author: mp-orkes

examples/agents/78_approval_workflow.py

@@ -1,20 +1,24 @@
 """Approval Workflow — agent dynamically decides which tasks need human sign-off.
 
 Demonstrates:
-    - wait_for_message_tool as a dynamic approval gate driven by LLM reasoning
+    - wait_for_message_tool as the task intake; flag_for_approval (a @tool) as a
+      dynamic approval gate driven by LLM reasoning
     - The agent itself decides mid-loop whether a task is risky, rather than
       the workflow being designed with an explicit approval step upfront
     - flag_for_approval blocks until the operator decides, returning "approve"
       or "reject" directly — no second wait_for_message needed for the decision,
       which prevents the agent from pulling the next task while approval is pending
     - Filesystem-based IPC between the main process and worker processes:
       tool workers run as separate OS processes (different PIDs, same filesystem),
-      so @tool functions use sentinel files to communicate with the main thread
-    - Clean shutdown: the agent responds with no tool calls on the stop signal,
-      which lets the DoWhile loop exit naturally (workflow ends COMPLETED)
-
-How this differs from examples 09a–09d (HITL):
-    In 09a–09d the approval pause is a WaitTask node baked into the workflow
+      so @tool functions use sentinel files to talk to the main process.  The
+      shared directory crosses process boundaries via APPROVAL_WORKFLOW_IPC_DIR —
+      a per-import mkdtemp() would give every worker its own dir.
+    - Deterministic stop: handle.stop() ends the loop once every task has been
+      accounted for, without any stop-handling instructions in the prompt
+      (workflow ends COMPLETED)
+
+How this differs from examples 09–09d (HITL):
+    In 09–09d the approval pause is a WaitTask node baked into the workflow
     definition at compile time — the workflow always pauses at that point
     regardless of the input.  Here, the LLM inspects each incoming task and
     decides dynamically whether it is safe to execute immediately or requires
@@ -31,7 +35,7 @@
     blocks on flag_for_approval until the operator responds.
 
 Requirements:
-    - Conductor server running at http://localhost:8080
+    - Conductor server with WMQ support (conductor.workflow-message-queue.enabled=true)
     - CONDUCTOR_SERVER_URL=http://localhost:8080/api as environment variable
     - CONDUCTOR_AGENT_LLM_MODEL=openai/gpt-4o-mini as environment variable
 """
@@ -49,12 +53,19 @@
 from settings import settings
 
 # Shared directory for IPC between main process and worker processes.
-# Workers run as separate OS processes (different PIDs, same filesystem).
-_ipc_dir = Path(tempfile.mkdtemp(prefix="approval_workflow_"))
+# Workers run as separate OS processes (different PIDs, same filesystem) that
+# re-import this module, so the directory is passed down via an env var — a
+# per-import mkdtemp() would give every worker its own dir and break the IPC.
+_IPC_DIR_ENV = "APPROVAL_WORKFLOW_IPC_DIR"
+if _IPC_DIR_ENV in os.environ:
+    _ipc_dir = Path(os.environ[_IPC_DIR_ENV])
+else:
+    _ipc_dir = Path(tempfile.mkdtemp(prefix="approval_workflow_"))
+    os.environ[_IPC_DIR_ENV] = str(_ipc_dir)
 _APPROVAL_DIR = _ipc_dir / "approvals"
 _DONE_DIR = _ipc_dir / "done"
-_APPROVAL_DIR.mkdir()
-_DONE_DIR.mkdir()
+_APPROVAL_DIR.mkdir(exist_ok=True)
+_DONE_DIR.mkdir(exist_ok=True)
 
 
 @tool
@@ -94,7 +105,7 @@ def log_rejection(task: str) -> str:
 
 receive_message = wait_for_message_tool(
     name="wait_for_message",
-    description="Dequeue the next task or stop signal ({stop: true}).",
+    description="Dequeue the next task to process.",
 )
 
 agent = Agent(
@@ -127,35 +138,43 @@ def log_rejection(task: str) -> str:
     "Grant admin access to user@example.com",
 ]
 
-try:
-    with AgentRuntime() as runtime:
-        handle = runtime.start(agent, "Start processing the task queue.")
-        execution_id = handle.execution_id
-        time.sleep(4)
-        print(f"Agent started: {execution_id}\n")
-
-        print("Dispatching all tasks...\n")
-        for task in TASKS:
-            print(f"  → {task!r}")
-            runtime.send_message(execution_id, {"task": task})
-
-        # Poll for approval requests; write decision files to unblock the tool.
-        # Poll for completions to know when to send the stop signal.
-        while len(list(_DONE_DIR.iterdir())) < len(TASKS):
-            for req in sorted(_APPROVAL_DIR.glob("*.json")):
-                data = json.loads(req.read_text())
-                req.unlink()
-                print(f"\n  ⚠ APPROVAL REQUIRED")
-                print(f"    Task:   {data['task']}")
-                print(f"    Reason: {data['reason']}\n")
-                answer = input("  Approve? [Y/N]: ").strip().upper()
-                decision = "approve" if answer == "Y" else "reject"
-                req.with_suffix(".decision").write_text(decision)
-            time.sleep(0.1)
-
-        # Deterministic stop — no stop-handling instructions needed.
-        handle.stop()
-        handle.join(timeout=30)
-        print("\nDone.")
-finally:
-    shutil.rmtree(_ipc_dir, ignore_errors=True)
+def main() -> None:
+    try:
+        with AgentRuntime() as runtime:
+            handle = runtime.start(agent, "Start processing the task queue.")
+            execution_id = handle.execution_id
+            time.sleep(4)
+            print(f"Agent started: {execution_id}\n")
+
+            print("Dispatching all tasks...\n")
+            for task in TASKS:
+                print(f"  → {task!r}")
+                runtime.send_message(execution_id, {"task": task})
+
+            # Poll for approval requests; write decision files to unblock the tool.
+            # Poll for completions to know when to send the stop signal.
+            while len(list(_DONE_DIR.iterdir())) < len(TASKS):
+                for req in sorted(_APPROVAL_DIR.glob("*.json")):
+                    data = json.loads(req.read_text())
+                    req.unlink()
+                    print("\n  ⚠ APPROVAL REQUIRED")
+                    print(f"    Task:   {data['task']}")
+                    print(f"    Reason: {data['reason']}\n")
+                    answer = input("  Approve? [Y/N]: ").strip().upper()
+                    decision = "approve" if answer == "Y" else "reject"
+                    req.with_suffix(".decision").write_text(decision)
+                time.sleep(0.1)
+
+            # Deterministic stop — no stop-handling instructions needed.
+            handle.stop()
+            handle.join(timeout=30)
+            print("\nDone.")
+    finally:
+        shutil.rmtree(_ipc_dir, ignore_errors=True)
+
+
+# Guard the runtime block: spawned tool workers re-import this module, and
+# without the guard they would re-run the orchestration (multiprocessing's
+# "Safe importing of main module" error).
+if __name__ == "__main__":
+    main()

examples/agents/79_agent_message_bus.py

@@ -3,10 +3,16 @@
 Demonstrates:
     - Agent-to-agent messaging: one running agent sending messages directly
       into another running agent's WMQ via runtime.send_message()
-    - A tool that closes over an execution_id to forward results downstream
+    - Module-level tools that pick up runtime values from the environment:
+      forward_to_writer reads the Writer's execution id from
+      MESSAGE_BUS_WRITER_EXECUTION_ID, since a tool that closed over it could not
+      be pickled to its spawned worker process
     - Parallel agent pipelines: researcher → writer running concurrently
-    - Filesystem-based IPC: forward_to_writer writes sentinel files so the main
-      thread knows when all topics have been forwarded
+    - Filesystem-based IPC between the main process and worker processes:
+      forward_to_writer and publish each write sentinel files, so the main process
+      can tell forwarding from publishing.  The barrier waits on publish — the
+      Researcher forwards the last topic while the Writer is still mid-turn on it,
+      so stopping at "all forwarded" would cut the final paragraph.
     - Deterministic stop: handle.stop() exits each agent's loop gracefully
 
 How this differs from 06_sequential_pipeline:
@@ -26,7 +32,7 @@
     Researcher autonomously drives the Writer.
 
 Requirements:
-    - Conductor server running at http://localhost:8080
+    - Conductor server with WMQ support (conductor.workflow-message-queue.enabled=true)
     - CONDUCTOR_SERVER_URL=http://localhost:8080/api as environment variable
     - CONDUCTOR_AGENT_LLM_MODEL=openai/gpt-4o-mini as environment variable
 """
@@ -48,8 +54,10 @@
 else:
     _ipc_dir = Path(tempfile.mkdtemp(prefix="message_bus_"))
     os.environ[_IPC_DIR_ENV] = str(_ipc_dir)
-_FORWARDED_DIR = _ipc_dir / "forwarded"  # one file per forwarded topic
+_FORWARDED_DIR = _ipc_dir / "forwarded"  # one file per topic forwarded by the Researcher
 _FORWARDED_DIR.mkdir(exist_ok=True)
+_PUBLISHED_DIR = _ipc_dir / "published"  # one file per paragraph published by the Writer
+_PUBLISHED_DIR.mkdir(exist_ok=True)
 
 TOPICS = [
     "the impact of edge computing on cloud infrastructure",
@@ -101,6 +109,7 @@ def publish(topic: str, paragraph: str) -> str:
     """Publish the finished paragraph."""
     print(f"\n  [writer] ── {topic} ──")
     print(f"  {paragraph}\n")
+    (_PUBLISHED_DIR / f"{time.time_ns()}.done").touch()
     return "published"
 
 
@@ -152,8 +161,18 @@ def main() -> None:
                 print(f"  → {topic!r}")
                 runtime.send_message(researcher_id, {"topic": topic})
 
-            # Wait until all topics have been forwarded to the Writer
-            while len(list(_FORWARDED_DIR.iterdir())) < len(TOPICS):
+            # Wait until the Writer has published every paragraph.  Gating on
+            # _FORWARDED_DIR is not enough: the Researcher forwards the last topic
+            # while the Writer is still mid-turn on it, so stopping there would cut
+            # the final paragraph and can leave the Researcher's stop() racing an
+            # in-flight iteration.
+            deadline = time.monotonic() + 180
+            while len(list(_PUBLISHED_DIR.iterdir())) < len(TOPICS):
+                if time.monotonic() > deadline:
+                    raise TimeoutError(
+                        f"Writer published {len(list(_PUBLISHED_DIR.iterdir()))} of "
+                        f"{len(TOPICS)} paragraphs before the deadline."
+                    )
                 time.sleep(0.1)
 
             # Deterministic stop — no stop-handling instructions needed.

examples/agents/80_live_dashboard.py

@@ -37,7 +37,7 @@
 Requirements:
     - Conductor server running at http://localhost:8080
     - CONDUCTOR_SERVER_URL=http://localhost:8080/api as environment variable
-    - CONDUCTOR_AGENT_LLM_MODEL=anthropic/claude-sonnet-4-20250514 as environment variable
+    - CONDUCTOR_AGENT_LLM_MODEL=anthropic/claude-sonnet-5 as environment variable
 """
 
 import json

examples/agents/81_chat_repl.py

@@ -15,8 +15,10 @@
     communicated via the shared filesystem rather than an in-process queue.
 
 Resume support:
-    The REPL saves the execution_id to a session file on start.  On subsequent
-    runs, pass ``--resume`` to reconnect to the same workflow.  ``resume()``
+    The REPL saves the execution_id to a session file on start.  Leave with
+    ``/disconnect`` to exit the console without stopping the agent (``quit`` /
+    ``exit`` stop it), then pass ``--resume`` on a later run to reconnect to the
+    same workflow.  ``resume()``
     fetches the workflow from the server, extracts the worker domain from
     ``taskToDomain``, and re-registers tools under that domain — so stateful
     agents resume correctly.  Conversation history is not restored in the
@@ -30,7 +32,7 @@
     activate predefined text-processing tasks at runtime.  The agent is notified
     via a WMQ message and can start using the new capability immediately.
 
-    Built-in tasks (activate with /tool <name>):
+    Built-in tasks (activate with /tool <name>, list the active ones with /tools):
         word_count   — count words in input
         char_count   — count characters in input
         reverse      — reverse the input string
@@ -41,9 +43,9 @@
         bullet_split — split input into one bullet point per sentence
 
 Requirements:
-    - Conductor server running at http://localhost:8080
+    - Conductor server with WMQ support (conductor.workflow-message-queue.enabled=true)
     - CONDUCTOR_SERVER_URL=http://localhost:8080/api as environment variable
-    - CONDUCTOR_AGENT_LLM_MODEL=anthropic/claude-sonnet-4-20250514 as environment variable
+    - CONDUCTOR_AGENT_LLM_MODEL=anthropic/claude-sonnet-5 as environment variable
 """
 
 import argparse

examples/agents/82_coding_agent.py

@@ -15,9 +15,9 @@
     python 82_coding_agent.py --resume             # resume last session
 
 Requirements:
-    - Conductor server running at http://localhost:8080
+    - Conductor server with WMQ support (conductor.workflow-message-queue.enabled=true)
     - CONDUCTOR_SERVER_URL=http://localhost:8080/api
-    - CONDUCTOR_AGENT_LLM_MODEL=anthropic/claude-sonnet-4-20250514
+    - CONDUCTOR_AGENT_LLM_MODEL=anthropic/claude-sonnet-5
 """
 
 import argparse
@@ -461,9 +461,12 @@ def build_agent(working_dir: str, shell_timeout: int = _DEFAULT_SHELL_TIMEOUT) -
 - run_shell(command)                           run a shell command (cwd: {working_dir}, timeout: {shell_timeout}s)
 - find_files(pattern, path=".")               find files by glob, e.g. "**/*.py"
 - search_in_files(regex, path=".", file_glob) grep files by regex
-- reply_to_user(message)                       send your response to the user
+- reply_to_user(message)                       REQUIRED — how the user sees your answer
 
 Rules:
+- You MUST call reply_to_user before calling wait_for_message again. It is the only
+  channel the user can see — plain text replies are discarded.
+- Never call wait_for_message twice in a row. Every task ends with reply_to_user.
 - Work autonomously. Do not ask for permission before reading files, running commands, or writing.
 - Make as many tool calls as needed to fully complete the task before replying.
 - Keep replies concise: what was done, what changed, key output. No lengthy explanations.
@@ -474,8 +477,8 @@ def build_agent(working_dir: str, shell_timeout: int = _DEFAULT_SHELL_TIMEOUT) -
 1. Call wait_for_message to receive the next task.
 2. Think through the task. Explore, read, search, modify, and run as needed.
 3. Complete the task fully.
-4. Call reply_to_user with a concise summary.
-5. Return to step 1 immediately.
+4. Call reply_to_user with a concise summary. Never skip this step.
+5. Only then return to step 1.
 """,
     )
 

examples/agents/82_fan_out_fan_in.py

@@ -21,16 +21,19 @@
     - Filesystem IPC:
         * Workers write sentinels after submit_answer so main counts completions
         * Collector writes reports to files; main thread reads and prints them
-    - No time.sleep() to assume message delivery — synchronisation via files
+    - Result delivery is never assumed from elapsed time — every wait is on a
+      sentinel file.  The one bare sleep is a startup grace period, letting the
+      agents reach their first wait call before the first message is sent.
 
 Scenario:
     A research Orchestrator fans out each question to three Worker agents
     (alpha, beta, gamma) that produce independent short answers.  The Collector
     aggregates the three answers into a side-by-side comparison report.
 
 Requirements:
-    - Conductor server running (CONDUCTOR_SERVER_URL / CONDUCTOR_SERVER_URL)
-    - CONDUCTOR_AGENT_LLM_MODEL set to a working model
+    - Conductor server with WMQ support (conductor.workflow-message-queue.enabled=true)
+    - CONDUCTOR_SERVER_URL=http://localhost:8080/api as environment variable
+    - CONDUCTOR_AGENT_LLM_MODEL=openai/gpt-4o-mini as environment variable
 """
 
 import json

examples/agents/84_deterministic_stop.py

@@ -8,8 +8,8 @@
 
 How it works:
     The server compiles every agent's DoWhile loop with a ``_stop_requested``
-    workflow variable in its condition.  When ``handle.stop()`` is called, the
-    SDK sets this variable to ``true`` via Conductor's ``updateVariables`` API.
+    workflow variable in its condition.  ``handle.stop()`` POSTs to
+    ``/agent/{execution_id}/stop`` and the server sets that variable to ``true``.
     The loop condition evaluates to ``false`` on the next check, and the loop
     exits.  The LLM cannot override this — it's checked by Conductor, not the
     LLM.
@@ -28,7 +28,8 @@
     The LLM could ignore this.  handle.stop() makes this unnecessary.
 
 Requirements:
-    - Conductor server (with _stop_requested support in compiler)
+    - Conductor server with WMQ support (conductor.workflow-message-queue.enabled=true)
+      and _stop_requested support in the compiler
     - CONDUCTOR_SERVER_URL=http://localhost:8080/api as environment variable
     - CONDUCTOR_AGENT_LLM_MODEL=openai/gpt-4o-mini as environment variable
 """