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

Fix WMQ examples (75-77)

Author: mp-orkes

examples/agents/75_wait_for_message.py

@@ -3,11 +3,17 @@
 Demonstrates:
     - wait_for_message_tool: dequeues messages from the WMQ (Conductor PULL_WORKFLOW_MESSAGES task)
     - Mixing a server-side message tool with a local Python action tool
-    - Looping agent that keeps processing messages indefinitely
+    - Open-ended looping agent: it never decides to stop on its own
     - Pushing messages from outside the workflow with runtime.send_message()
+    - handle.stop() ending the loop deterministically
 
-The agent loops forever: each iteration waits for a message, reads the
-"task" field, executes it, and goes back to listening.
+Each iteration waits for a message, reads the "task" field, executes it, and
+goes back to listening.  The agent's instructions tell it to never stop, so the
+loop only ends when the caller calls handle.stop(): that sets the
+``_stop_requested`` workflow variable checked by the DoWhile condition and
+pushes a ``{"_signal": "stop"}`` message to unblock the pending
+PULL_WORKFLOW_MESSAGES.  The workflow ends COMPLETED, not TERMINATED — see
+84_deterministic_stop.py.
 
 Requirements:
     - Conductor server with WMQ support (conductor.workflow-message-queue.enabled=true)
@@ -67,6 +73,7 @@ def main() -> None:
 
         # Let the agent process all messages (~5-6s per message)
         time.sleep(30)
+        # The agent will never stop on its own — end the loop from here.
         handle.stop()
         handle.join(timeout=30)
         print("\nDone.")

examples/agents/76_wait_for_message_streaming.py

@@ -4,10 +4,18 @@
     - wait_for_message_tool with streaming: push messages in and see the agent react
     - Using handle.stream() to observe WAITING → processing → WAITING cycles
     - runtime.send_message() to push payloads into the Workflow Message Queue
+    - handle.stop() ending the loop deterministically
 
-The agent starts, immediately waits for a message, processes whatever it
-receives (by calling wait_for_message again), then waits again.  The caller
-drives the conversation by sending messages and reading streamed events.
+The agent starts, immediately waits for a message, answers it with respond(),
+then loops back to wait_for_message.  The caller drives the conversation from a
+background thread — sending a task every 8 seconds — while the main thread reads
+streamed events.
+
+The agent's instructions tell it to never stop, so the loop only ends when the
+sender calls handle.stop() — after giving the last task time to be answered.
+That sets the ``_stop_requested`` workflow variable
+checked by the DoWhile condition and pushes a ``{"_signal": "stop"}`` message to
+unblock the pending PULL_WORKFLOW_MESSAGES.  stream() then yields DONE.
 
 Requirements:
     - Conductor server running at http://localhost:8080
@@ -66,15 +74,16 @@ def main() -> None:
         print(f"Agent started: {handle.execution_id}\n")
 
         # Push messages from a background thread while we stream events on the main thread.
-        # Wait long enough between sends for the agent to finish processing each message.
-        # No sleep after the last send — handle.stream() on the main thread is already the
-        # barrier: it blocks until DONE, which only fires once the workflow reaches a
-        # terminal state (after stop() sets the flag and the current iteration completes).
+        # Wait long enough between sends for the agent to finish processing each message —
+        # including after the last one.  Calling stop() immediately after the final send
+        # would set _stop_requested while the agent is still mid-turn on that task, and the
+        # DoWhile would exit before it ever answers.
         def sender():
             for task in TASKS:
                 time.sleep(8)
                 print(f"\n  [caller] sending -> {task!r}")
                 runtime.send_message(handle.execution_id, {"task": task})
+            time.sleep(8)
             handle.stop()
 
         threading.Thread(target=sender, daemon=True).start()

examples/agents/77_kafka_consumer_agent.py

@@ -58,36 +58,48 @@ def echo_message(value: str, topic: str, offset: int) -> str:
 )
 
 
-with AgentRuntime() as runtime:
-    handle = runtime.start(agent, "Start consuming messages from Kafka.")
-    print(f"Agent started: {handle.execution_id}")
-
-    consumer = Consumer(
-        {
-            "bootstrap.servers": KAFKA_BOOTSTRAP,
-            "group.id": KAFKA_GROUP,
-            "auto.offset.reset": "latest",
-        }
-    )
-    consumer.subscribe([KAFKA_TOPIC])
-    try:
-        while True:
-            msg = consumer.poll(timeout=1.0)
-            if msg is None:
-                continue
-            if msg.error():
-                if msg.error().code() == KafkaError._PARTITION_EOF:
+def main() -> None:
+    with AgentRuntime() as runtime:
+        handle = runtime.start(agent, "Start consuming messages from Kafka.")
+        print(f"Agent started: {handle.execution_id}")
+
+        consumer = Consumer(
+            {
+                "bootstrap.servers": KAFKA_BOOTSTRAP,
+                "group.id": KAFKA_GROUP,
+                "auto.offset.reset": "latest",
+            }
+        )
+        consumer.subscribe([KAFKA_TOPIC])
+        print(f"Consuming '{KAFKA_TOPIC}' from {KAFKA_BOOTSTRAP} — Ctrl+C to stop.")
+        try:
+            while True:
+                msg = consumer.poll(timeout=1.0)
+                if msg is None:
                     continue
-                raise RuntimeError(f"Kafka error: {msg.error()}")
-            runtime.send_message(
-                handle.execution_id,
-                {
-                    "topic": msg.topic(),
-                    "partition": msg.partition(),
-                    "offset": msg.offset(),
-                    "key": msg.key().decode("utf-8") if msg.key() else None,
-                    "value": msg.value().decode("utf-8") if msg.value() else "",
-                },
-            )
-    finally:
-        consumer.close()
+                if msg.error():
+                    if msg.error().code() == KafkaError._PARTITION_EOF:
+                        continue
+                    raise RuntimeError(f"Kafka error: {msg.error()}")
+                runtime.send_message(
+                    handle.execution_id,
+                    {
+                        "topic": msg.topic(),
+                        "partition": msg.partition(),
+                        "offset": msg.offset(),
+                        "key": msg.key().decode("utf-8") if msg.key() else None,
+                        "value": msg.value().decode("utf-8") if msg.value() else "",
+                    },
+                )
+        except KeyboardInterrupt:
+            print("\nStopping agent...")
+            handle.stop()
+        finally:
+            consumer.close()
+
+
+# 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()