From 6e06c86d04fcc0c4f7e8d8a08de8672a70ab8c28 Mon Sep 17 00:00:00 2001 From: Miguel Prieto Date: Fri, 14 Aug 2026 12:58:53 -0300 Subject: [PATCH 1/9] Fix example 78 crashing on worker spawn Co-Authored-By: Claude Opus 5 (1M context) --- examples/agents/78_approval_workflow.py | 87 +++++++++++++++---------- 1 file changed, 51 insertions(+), 36 deletions(-) diff --git a/examples/agents/78_approval_workflow.py b/examples/agents/78_approval_workflow.py index 91d9d7fb..a6270a7a 100644 --- a/examples/agents/78_approval_workflow.py +++ b/examples/agents/78_approval_workflow.py @@ -49,12 +49,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 @@ -127,35 +134,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() From 0ba5adbcab6b3623fc7e870b1f14e1613e60ce5e Mon Sep 17 00:00:00 2001 From: Miguel Prieto Date: Fri, 14 Aug 2026 13:03:54 -0300 Subject: [PATCH 2/9] Fix stale docstring in example 78 Co-Authored-By: Claude Opus 5 (1M context) --- examples/agents/78_approval_workflow.py | 22 +++++++++++++--------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/examples/agents/78_approval_workflow.py b/examples/agents/78_approval_workflow.py index a6270a7a..71b8bc1a 100644 --- a/examples/agents/78_approval_workflow.py +++ b/examples/agents/78_approval_workflow.py @@ -1,7 +1,8 @@ """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" @@ -9,12 +10,15 @@ 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 """ @@ -101,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( From 2d67db94805b3965382db923461428f7bd0c6a74 Mon Sep 17 00:00:00 2001 From: Miguel Prieto Date: Fri, 14 Aug 2026 13:09:05 -0300 Subject: [PATCH 3/9] Fix example 79 dropping the last paragraph Co-Authored-By: Claude Opus 5 (1M context) --- examples/agents/79_agent_message_bus.py | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/examples/agents/79_agent_message_bus.py b/examples/agents/79_agent_message_bus.py index f2dde389..40855a76 100644 --- a/examples/agents/79_agent_message_bus.py +++ b/examples/agents/79_agent_message_bus.py @@ -48,8 +48,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 +103,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 +155,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. From c960a5f4a45141bfe1454df948c4ea860e89f217 Mon Sep 17 00:00:00 2001 From: Miguel Prieto Date: Fri, 14 Aug 2026 13:22:43 -0300 Subject: [PATCH 4/9] Fix stale docstring in example 79 Co-Authored-By: Claude Opus 5 (1M context) --- examples/agents/79_agent_message_bus.py | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/examples/agents/79_agent_message_bus.py b/examples/agents/79_agent_message_bus.py index 40855a76..55051c13 100644 --- a/examples/agents/79_agent_message_bus.py +++ b/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 """ From 5dbbf10023233f67f3a6a0adc111af55b3ce5288 Mon Sep 17 00:00:00 2001 From: Miguel Prieto Date: Fri, 14 Aug 2026 14:19:06 -0300 Subject: [PATCH 5/9] Document /disconnect and /tools in example 81 Co-Authored-By: Claude Opus 5 (1M context) --- examples/agents/81_chat_repl.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/examples/agents/81_chat_repl.py b/examples/agents/81_chat_repl.py index 8c54170a..8c0c4a83 100644 --- a/examples/agents/81_chat_repl.py +++ b/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 ): + Built-in tasks (activate with /tool , list the active ones with /tools): word_count — count words in input char_count — count characters in input reverse — reverse the input string @@ -41,7 +43,7 @@ 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 """ From f88cbfb07e10c9ced031cdfd616dcd72140ae7ed Mon Sep 17 00:00:00 2001 From: Miguel Prieto Date: Fri, 14 Aug 2026 15:35:44 -0300 Subject: [PATCH 6/9] Require reply_to_user before looping in example 82 Co-Authored-By: Claude Opus 5 (1M context) --- examples/agents/82_coding_agent.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/examples/agents/82_coding_agent.py b/examples/agents/82_coding_agent.py index 9ce43de3..2955081e 100644 --- a/examples/agents/82_coding_agent.py +++ b/examples/agents/82_coding_agent.py @@ -17,7 +17,7 @@ Requirements: - Conductor server running at http://localhost:8080 - 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. """, ) From c10590e116e4d043cf2b8227dda5ac105b8df566 Mon Sep 17 00:00:00 2001 From: Miguel Prieto Date: Fri, 14 Aug 2026 15:35:44 -0300 Subject: [PATCH 7/9] Replace retired model id in examples 80 and 81 Co-Authored-By: Claude Opus 5 (1M context) --- examples/agents/80_live_dashboard.py | 2 +- examples/agents/81_chat_repl.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/examples/agents/80_live_dashboard.py b/examples/agents/80_live_dashboard.py index cdabaefb..e6e525a1 100644 --- a/examples/agents/80_live_dashboard.py +++ b/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 diff --git a/examples/agents/81_chat_repl.py b/examples/agents/81_chat_repl.py index 8c0c4a83..244d9f15 100644 --- a/examples/agents/81_chat_repl.py +++ b/examples/agents/81_chat_repl.py @@ -45,7 +45,7 @@ Requirements: - 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 From d81bdef2a7eb09d94c3d5475e5b5e7d35a287876 Mon Sep 17 00:00:00 2001 From: Miguel Prieto Date: Fri, 14 Aug 2026 16:13:25 -0300 Subject: [PATCH 8/9] Fix stale docstring in example 82_fan_out_fan_in Co-Authored-By: Claude Opus 5 (1M context) --- examples/agents/82_fan_out_fan_in.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/examples/agents/82_fan_out_fan_in.py b/examples/agents/82_fan_out_fan_in.py index 3fe65431..dc32726c 100644 --- a/examples/agents/82_fan_out_fan_in.py +++ b/examples/agents/82_fan_out_fan_in.py @@ -21,7 +21,9 @@ - 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 @@ -29,8 +31,9 @@ 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 From 714b289bfbffe438bfddc5a3bc5f4c8c01f9e320 Mon Sep 17 00:00:00 2001 From: Miguel Prieto Date: Fri, 14 Aug 2026 16:39:44 -0300 Subject: [PATCH 9/9] Fix stale docstrings in examples 82 and 84 Co-Authored-By: Claude Opus 5 (1M context) --- examples/agents/82_coding_agent.py | 2 +- examples/agents/84_deterministic_stop.py | 7 ++++--- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/examples/agents/82_coding_agent.py b/examples/agents/82_coding_agent.py index 2955081e..0dd4fe53 100644 --- a/examples/agents/82_coding_agent.py +++ b/examples/agents/82_coding_agent.py @@ -15,7 +15,7 @@ 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-5 """ diff --git a/examples/agents/84_deterministic_stop.py b/examples/agents/84_deterministic_stop.py index be3c78a9..a5bcb8c3 100644 --- a/examples/agents/84_deterministic_stop.py +++ b/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 """