fix(claude-code): run Claude Code CLI in a Temporal activity, not workflow code

declan-scale · claude · declan-scale · commit eee33201e613 · 2026-06-22T17:49:34.000-04:00
The temporal tutorial spawned the CLI subprocess directly in the workflow
signal handler. Temporal runs signal-handler bodies on its deterministic
sandbox event loop, which does not implement asyncio.create_subprocess_exec —
so the worker crashed with NotImplementedError and no tool_request/response was
ever produced. (Replay was never the issue; the sandboxed loop is.)

Move the subprocess + ClaudeCodeTurn + UnifiedEmitter.auto_send_turn into a new
run_claude_code_turn activity (project/activities.py), register it on the
worker, and have the signal handler delegate via workflow.execute_activity. The
activity returns final_text + session_id so multi-turn resume still works.

Co-Authored-By: Claude Opus 4.8 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/examples/tutorials/10_async/10_temporal/140_claude_code/project/activities.py b/examples/tutorials/10_async/10_temporal/140_claude_code/project/activities.py
@@ -0,0 +1,143 @@
+"""Temporal activity for the Claude Code tutorial.
+
+Subprocess spawning (and any other I/O) must run inside a Temporal *activity*,
+not in workflow code. Temporal runs workflow + signal-handler bodies on a
+deterministic sandbox event loop that does not implement ``subprocess_exec``
+(or threads / sockets), so spawning the CLI directly in the signal handler
+raises ``NotImplementedError``. This activity runs the Claude Code CLI, drives
+the ``ClaudeCodeTurn`` through ``UnifiedEmitter.auto_send_turn`` (the async
+Redis push path), and returns the turn result to the workflow.
+
+The ``_spawn_claude`` async generator is an injectable seam: offline tests
+provide a fake that yields pre-recorded stdout lines so no real CLI runs.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any, AsyncIterator
+from datetime import datetime
+
+from temporalio import activity
+
+from agentex.lib.adk import ClaudeCodeTurn
+from agentex.lib.core.harness import UnifiedEmitter
+from agentex.lib.utils.logging import make_logger
+from agentex.lib.utils.model_utils import BaseModel
+
+logger = make_logger(__name__)
+
+RUN_CLAUDE_CODE_TURN_ACTIVITY = "run_claude_code_turn"
+
+
+class RunClaudeCodeTurnParams(BaseModel):
+    """Arguments for one Claude Code turn run inside an activity."""
+
+    task_id: str
+    prompt: str
+    trace_id: str | None = None
+    parent_span_id: str | None = None
+    session_id: str | None = None
+    created_at: datetime | None = None
+
+
+class RunClaudeCodeTurnResult(BaseModel):
+    """Result returned from the activity to the workflow."""
+
+    final_text: str
+    session_id: str | None = None
+
+
+async def _spawn_claude(prompt: str, session_id: str | None = None) -> AsyncIterator[str]:
+    """Spawn ``claude -p --output-format stream-json`` locally and yield stdout lines.
+
+    Pass ``session_id`` to resume a previous Claude Code session (multi-turn
+    memory via ``-r <session_id>``).
+
+    Injectable seam: tests monkeypatch this with a fake async iterator so no
+    real CLI invocation is needed offline.
+    """
+    cmd = [
+        "claude",
+        "-p",
+        "--output-format",
+        "stream-json",
+        "--verbose",
+    ]
+    if session_id:
+        cmd.extend(["-r", session_id])
+
+    proc = await asyncio.create_subprocess_exec(
+        *cmd,
+        stdin=asyncio.subprocess.PIPE,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    assert proc.stdout is not None
+    assert proc.stdin is not None
+
+    proc.stdin.write(prompt.encode())
+    proc.stdin.close()
+
+    # Drain stderr concurrently. With --verbose, Claude Code can write enough to
+    # stderr to fill the OS pipe buffer; if we only read stdout, the CLI blocks
+    # on its stderr write while we block reading stdout — a deadlock. A
+    # background task keeps stderr flowing so stdout never stalls.
+    async def _drain_stderr() -> None:
+        assert proc.stderr is not None
+        async for _ in proc.stderr:
+            pass
+
+    stderr_task = asyncio.create_task(_drain_stderr())
+
+    try:
+        buffer = ""
+        async for chunk in proc.stdout:
+            buffer += chunk.decode("utf-8", errors="replace")
+            while "\n" in buffer:
+                line, buffer = buffer.split("\n", 1)
+                line = line.strip()
+                if line:
+                    yield line
+
+        if buffer.strip():
+            yield buffer.strip()
+
+        await proc.wait()
+    finally:
+        # Release the subprocess and stderr drain task even if the consumer
+        # abandons the generator early (task cancellation / client disconnect):
+        # cancel the drain task and terminate+reap the process if it is still
+        # running, so neither is leaked.
+        stderr_task.cancel()
+        try:
+            await stderr_task
+        except asyncio.CancelledError:
+            pass
+        if proc.returncode is None:
+            try:
+                proc.terminate()
+            except ProcessLookupError:
+                pass
+            await proc.wait()
+
+
+@activity.defn(name=RUN_CLAUDE_CODE_TURN_ACTIVITY)
+async def run_claude_code_turn(params: RunClaudeCodeTurnParams) -> dict[str, Any]:
+    """Run one Claude Code turn end-to-end and stream events to the task.
+
+    Runs in an activity (real asyncio loop) so subprocess I/O is permitted.
+    """
+    emitter = UnifiedEmitter(
+        task_id=params.task_id,
+        trace_id=params.trace_id,
+        parent_span_id=params.parent_span_id,
+    )
+    turn = ClaudeCodeTurn(_spawn_claude(params.prompt, session_id=params.session_id))
+    result = await emitter.auto_send_turn(turn, created_at=params.created_at)
+
+    session_id: str | None = None
+    if getattr(turn, "_result_envelope", None):
+        session_id = turn._result_envelope.get("session_id")
+
+    return RunClaudeCodeTurnResult(final_text=result.final_text, session_id=session_id).model_dump()
diff --git a/examples/tutorials/10_async/10_temporal/140_claude_code/project/run_worker.py b/examples/tutorials/10_async/10_temporal/140_claude_code/project/run_worker.py
@@ -3,18 +3,15 @@
 Run as a separate long-lived process alongside the ACP HTTP server. The
 worker polls Temporal for workflow + activity tasks and executes them.
 
-Claude Code does not register custom activities here -- subprocess spawning
-happens directly in the workflow signal handler (workflow code) to keep
-the tutorial minimal. The built-in Agentex activities (state, messages,
-streaming, tracing) are registered via ``get_all_activities()``.
-
-For production use, move the subprocess spawn into a custom activity so
-it gets Temporal's retry + timeout guarantees.
+The Claude Code CLI subprocess runs in the ``run_claude_code_turn`` activity
+(registered below alongside the built-in Agentex activities), because
+subprocess I/O is not permitted on the Temporal workflow event loop.
 """
 
 import asyncio
 
 from project.workflow import At140ClaudeCodeWorkflow
+from project.activities import run_claude_code_turn
 from agentex.lib.utils.debug import setup_debug_if_enabled
 from agentex.lib.utils.logging import make_logger
 from agentex.lib.environment_variables import EnvironmentVariables
@@ -35,7 +32,7 @@ async def main():
     worker = AgentexWorker(task_queue=task_queue_name)
 
     await worker.run(
-        activities=get_all_activities(),
+        activities=[run_claude_code_turn, *get_all_activities()],
         workflow=At140ClaudeCodeWorkflow,
     )
 
diff --git a/examples/tutorials/10_async/10_temporal/140_claude_code/project/workflow.py b/examples/tutorials/10_async/10_temporal/140_claude_code/project/workflow.py
@@ -1,34 +1,32 @@
 """Temporal workflow for the Claude Code tutorial.
 
 Holds conversation state (session_id for multi-turn resume) durably across
-crashes. Each user message triggers ``on_task_event_send``, which spawns the
-Claude Code CLI locally as an asyncio subprocess, wraps the stdout line
-stream in ``ClaudeCodeTurn``, and delivers the turn via
+crashes. Each user message triggers ``on_task_event_send``, which delegates the
+turn to the ``run_claude_code_turn`` activity. The activity spawns the Claude
+Code CLI, wraps its stdout in ``ClaudeCodeTurn``, and delivers the turn via
 ``UnifiedEmitter.auto_send_turn`` (the async Redis push path).
 
 Note on subprocess inside Temporal
 ------------------------------------
-Temporal activities, not workflow code, should do I/O. However, this tutorial
-executes the subprocess directly in the signal handler (workflow code) to keep
-the example minimal. For production use, move the subprocess spawn into a
-dedicated activity so it benefits from Temporal's retry and timeout guarantees.
-See ``examples/tutorials/10_async/10_temporal/030_custom_activities/`` for
-the activity pattern.
+Subprocess (and all other) I/O must run in a Temporal *activity*, never in
+workflow code. Temporal runs workflow + signal-handler bodies on a
+deterministic sandbox event loop that does not implement ``subprocess_exec``
+(spawning the CLI there raises ``NotImplementedError``). The activity also gets
+Temporal's retry + timeout guarantees. See
+``examples/tutorials/10_async/10_temporal/030_custom_activities/`` for the
+activity pattern.
 """
 
 from __future__ import annotations
 
 import os
 import json
-import asyncio
-from typing import AsyncIterator
+from datetime import timedelta
 
 from temporalio import workflow
 
 from agentex.lib import adk
-from agentex.lib.adk import ClaudeCodeTurn
 from agentex.lib.types.acp import SendEventParams, CreateTaskParams
-from agentex.lib.core.harness import UnifiedEmitter
 from agentex.lib.types.tracing import SGPTracingProcessorConfig
 from agentex.lib.utils.logging import make_logger
 from agentex.types.text_content import TextContent
@@ -37,6 +35,9 @@
 from agentex.lib.core.temporal.workflows.workflow import BaseWorkflow
 from agentex.lib.core.tracing.tracing_processor_manager import add_tracing_processor_config
 
+with workflow.unsafe.imports_passed_through():
+    from project.activities import RunClaudeCodeTurnParams, run_claude_code_turn
+
 add_tracing_processor_config(
     SGPTracingProcessorConfig(
         sgp_api_key=os.environ.get("SGP_API_KEY", ""),
@@ -55,80 +56,6 @@
 logger = make_logger(__name__)
 
 
-async def _spawn_claude(prompt: str, session_id: str | None = None) -> AsyncIterator[str]:
-    """Spawn ``claude -p --output-format stream-json`` locally and yield stdout lines.
-
-    Pass ``session_id`` to resume a previous Claude Code session (multi-turn
-    memory via ``-r <session_id>``).
-
-    Injectable seam: tests monkeypatch this with a fake async iterator so no
-    real CLI invocation is needed offline.
-    """
-    cmd = [
-        "claude",
-        "-p",
-        "--output-format",
-        "stream-json",
-        "--verbose",
-    ]
-    if session_id:
-        cmd.extend(["-r", session_id])
-
-    proc = await asyncio.create_subprocess_exec(
-        *cmd,
-        stdin=asyncio.subprocess.PIPE,
-        stdout=asyncio.subprocess.PIPE,
-        stderr=asyncio.subprocess.PIPE,
-    )
-    assert proc.stdout is not None
-    assert proc.stdin is not None
-
-    proc.stdin.write(prompt.encode())
-    proc.stdin.close()
-
-    # Drain stderr concurrently. With --verbose, Claude Code can write enough to
-    # stderr to fill the OS pipe buffer; if we only read stdout, the CLI blocks
-    # on its stderr write while we block reading stdout — a deadlock. A
-    # background task keeps stderr flowing so stdout never stalls.
-    async def _drain_stderr() -> None:
-        assert proc.stderr is not None
-        async for _ in proc.stderr:
-            pass
-
-    stderr_task = asyncio.create_task(_drain_stderr())
-
-    try:
-        buffer = ""
-        async for chunk in proc.stdout:
-            buffer += chunk.decode("utf-8", errors="replace")
-            while "\n" in buffer:
-                line, buffer = buffer.split("\n", 1)
-                line = line.strip()
-                if line:
-                    yield line
-
-        if buffer.strip():
-            yield buffer.strip()
-
-        await proc.wait()
-    finally:
-        # Release the subprocess and stderr drain task even if the consumer
-        # abandons the generator early (task cancellation / client disconnect):
-        # cancel the drain task and terminate+reap the process if it is still
-        # running, so neither is leaked.
-        stderr_task.cancel()
-        try:
-            await stderr_task
-        except asyncio.CancelledError:
-            pass
-        if proc.returncode is None:
-            try:
-                proc.terminate()
-            except ProcessLookupError:
-                pass
-            await proc.wait()
-
-
 @workflow.defn(name=environment_variables.WORKFLOW_NAME)
 class At140ClaudeCodeWorkflow(BaseWorkflow):
     """Temporal workflow that runs Claude Code locally for each user message.
@@ -161,29 +88,30 @@ async def on_task_event_send(self, params: SendEventParams) -> None:
             name=f"Turn {self._turn_number}",
             input={"message": prompt},
         ) as span:
-            emitter = UnifiedEmitter(
-                task_id=task_id,
-                trace_id=task_id,
-                parent_span_id=span.id if span else None,
+            # Delegate the subprocess turn to an activity: subprocess I/O is not
+            # permitted on the Temporal workflow event loop. The activity streams
+            # events to the task and returns the final text + session_id.
+            # workflow.now() gives a deterministic timestamp under replay.
+            result = await workflow.execute_activity(
+                run_claude_code_turn,
+                RunClaudeCodeTurnParams(
+                    task_id=task_id,
+                    prompt=prompt,
+                    trace_id=task_id,
+                    parent_span_id=span.id if span else None,
+                    session_id=self._session_id,
+                    created_at=workflow.now(),
+                ),
+                start_to_close_timeout=timedelta(minutes=5),
             )
 
-            # Use workflow.now() for deterministic timestamps under Temporal replay.
-            created_at = workflow.now()
-
-            turn = ClaudeCodeTurn(_spawn_claude(prompt, session_id=self._session_id))
-            result = await emitter.auto_send_turn(turn, created_at=created_at)
-
-            # Capture session_id from result envelope to enable resume on next turn.
-            # ClaudeCodeTurn.usage() gives us access to the raw result envelope via
-            # TurnUsage -- but session_id is not part of TurnUsage. We extract it
-            # separately by looking at the turn's internal state post-exhaust.
-            if hasattr(turn, "_result_envelope") and turn._result_envelope:
-                sid = turn._result_envelope.get("session_id")
-                if sid:
-                    self._session_id = sid
+            # Capture session_id to enable Claude Code resume on the next turn.
+            sid = result.get("session_id")
+            if sid:
+                self._session_id = sid
 
             if span:
-                span.output = {"final_text": result.final_text}
+                span.output = {"final_text": result.get("final_text")}
 
     @workflow.run
     async def on_task_create(self, params: CreateTaskParams) -> str:
