fix(codex): run codex CLI in a Temporal activity, not workflow code

declan-scale · claude · declan-scale · commit c37646ccb47f · 2026-06-22T17:49:48.000-04:00
The temporal tutorial spawned codex exec 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. (Replay was never the issue, contrary to the
old docstring; the sandboxed loop is.)

Move the subprocess + CodexTurn + UnifiedEmitter.auto_send_turn into a new
run_codex_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 codex-thread resume still works. Offline
tests now cover both the workflow's state update and the activity directly.

Co-Authored-By: Claude Opus 4.8 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/examples/tutorials/10_async/10_temporal/harness_codex/project/activities.py b/examples/tutorials/10_async/10_temporal/harness_codex/project/activities.py
@@ -0,0 +1,135 @@
+"""Temporal activity for the Codex harness 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 ``codex exec`` directly in the signal
+handler raises ``NotImplementedError``. This activity runs codex, drives the
+``CodexTurn`` through ``UnifiedEmitter.auto_send_turn`` (the async Redis push
+path), and returns the turn result to the workflow.
+
+The ``_spawn_codex`` / ``_process_stdout`` seams are injectable: offline tests
+replace them with fakes that yield pre-recorded event lines so no real CLI
+runs.
+"""
+
+from __future__ import annotations
+
+import os
+import asyncio
+from typing import Any
+from datetime import datetime
+from collections.abc import AsyncIterator
+
+from temporalio import activity
+
+from agentex.lib.adk import CodexTurn
+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_CODEX_TURN_ACTIVITY = "run_codex_turn"
+
+
+class RunCodexTurnParams(BaseModel):
+    """Arguments for one codex turn run inside an activity."""
+
+    task_id: str
+    prompt: str
+    model: str
+    trace_id: str | None = None
+    parent_span_id: str | None = None
+    thread_id: str | None = None
+    created_at: datetime | None = None
+
+
+class RunCodexTurnResult(BaseModel):
+    """Result returned from the activity to the workflow."""
+
+    final_text: str
+    session_id: str | None = None
+    model: str | None = None
+
+
+async def _spawn_codex(
+    model: str,
+    thread_id: str | None = None,
+) -> asyncio.subprocess.Process:
+    """Spawn ``codex exec --json`` locally and return the live process.
+
+    Injection seam: tests replace this function with a fake that returns a
+    mock process whose stdout yields pre-recorded event lines.
+
+    The caller writes the prompt to stdin after the process starts, then
+    closes stdin so codex knows input is complete.
+    """
+    base_flags = [
+        "--json",
+        "--skip-git-repo-check",
+        "--dangerously-bypass-approvals-and-sandbox",
+        "--model",
+        model,
+    ]
+
+    if thread_id:
+        cmd = ["codex", "exec", *base_flags, "resume", thread_id, "-"]
+    else:
+        cmd = ["codex", "exec", *base_flags, "-"]
+
+    return await asyncio.create_subprocess_exec(
+        *cmd,
+        stdin=asyncio.subprocess.PIPE,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+        env={**os.environ},
+    )
+
+
+async def _process_stdout(process: asyncio.subprocess.Process) -> AsyncIterator[str]:
+    """Yield newline-delimited JSON lines from the process stdout."""
+    assert process.stdout is not None
+    buffer = ""
+    while True:
+        chunk = await process.stdout.read(4096)
+        if not chunk:
+            break
+        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()
+
+
+@activity.defn(name=RUN_CODEX_TURN_ACTIVITY)
+async def run_codex_turn(params: RunCodexTurnParams) -> dict[str, Any]:
+    """Run one codex turn end-to-end and stream events to the task.
+
+    Runs in an activity (real asyncio loop) so subprocess I/O is permitted.
+    """
+    process = await _spawn_codex(params.model, thread_id=params.thread_id)
+
+    assert process.stdin is not None
+    process.stdin.write(params.prompt.encode("utf-8"))
+    await process.stdin.drain()
+    process.stdin.close()
+
+    turn = CodexTurn(events=_process_stdout(process), model=params.model)
+    emitter = UnifiedEmitter(
+        task_id=params.task_id,
+        trace_id=params.trace_id,
+        parent_span_id=params.parent_span_id,
+    )
+    result = await emitter.auto_send_turn(turn, created_at=params.created_at)
+
+    await process.wait()
+
+    return RunCodexTurnResult(
+        final_text=result.final_text,
+        session_id=turn.session_id,
+        model=turn.usage().model,
+    ).model_dump()
diff --git a/examples/tutorials/10_async/10_temporal/harness_codex/project/run_worker.py b/examples/tutorials/10_async/10_temporal/harness_codex/project/run_worker.py
@@ -3,13 +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.
 
-Codex subprocess calls happen inside signal handler bodies (not activities),
-so no extra activity registrations are needed beyond the standard Agentex set.
+The codex CLI subprocess runs in the ``run_codex_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 AtHarnessCodexWorkflow
+from project.activities import run_codex_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
@@ -30,7 +32,7 @@ async def main():
     worker = AgentexWorker(task_queue=task_queue_name)
 
     await worker.run(
-        activities=get_all_activities(),
+        activities=[run_codex_turn, *get_all_activities()],
         workflow=AtHarnessCodexWorkflow,
     )
 
diff --git a/examples/tutorials/10_async/10_temporal/harness_codex/project/workflow.py b/examples/tutorials/10_async/10_temporal/harness_codex/project/workflow.py
@@ -4,36 +4,29 @@
 ``UnifiedEmitter`` for a Temporal-durable ACP agent.
 
 KEY CONCEPTS DEMONSTRATED:
-- Spawning ``codex exec --json`` as a local asyncio subprocess inside a
-  Temporal workflow signal handler. The subprocess itself is NOT a Temporal
-  activity — for a tutorial that is fine. Production agents would wrap the
-  spawn in a Temporal activity for durability + observability.
-- Wrapping the stdout line stream in a ``CodexTurn``.
+- Running ``codex exec --json`` in the ``run_codex_turn`` activity. Subprocess
+  I/O is not permitted on the Temporal workflow event loop (the deterministic
+  sandbox loop does not implement ``subprocess_exec``), so the signal handler
+  delegates the turn to an activity, which also gets Temporal's retry + timeout
+  guarantees.
+- Wrapping the stdout line stream in a ``CodexTurn`` (inside the activity).
 - Delivering events via ``UnifiedEmitter.auto_send_turn``, which pushes
   ``StreamTaskMessage*`` events to Redis so the UI sees tokens in real time.
 - Passing ``created_at=workflow.now()`` for deterministic timestamps under
   Temporal replay (required for Temporal-safe delivery).
 - Persisting the codex thread ID on the workflow instance itself — Temporal's
   workflow state is durable, so no external ``adk.state`` round-trip is needed.
-
-NOTE: Subprocess spawning is safe inside a Temporal signal handler because
-Temporal does NOT replay signal handler bodies (only ``@workflow.run`` is
-subject to replay determinism constraints). Signal handlers run in the live
-process after the initial replay is complete.
 """
 
 from __future__ import annotations
 
 import os
-import asyncio
-from collections.abc import AsyncIterator
+from datetime import timedelta
 
 from temporalio import workflow
 
 from agentex.lib import adk
-from agentex.lib.adk import CodexTurn
 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
@@ -42,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 RunCodexTurnParams, run_codex_turn
+
 add_tracing_processor_config(
     SGPTracingProcessorConfig(
         sgp_api_key=os.environ.get("SGP_API_KEY", ""),
@@ -62,62 +58,6 @@
 MODEL = os.environ.get("CODEX_MODEL", "o4-mini")
 
 
-async def _spawn_codex(
-    model: str,
-    thread_id: str | None = None,
-) -> asyncio.subprocess.Process:
-    """Spawn ``codex exec --json`` locally and return the live process.
-
-    Injection seam: tests replace this function with a fake that returns a
-    mock process whose stdout yields pre-recorded event lines.
-
-    NOTE: This function must NOT be called during Temporal workflow replay
-    (Temporal's determinism guard would flag it as a non-deterministic side
-    effect). Signal handler bodies are safe: Temporal does not replay them.
-
-    The caller writes the prompt to stdin after the process starts, then
-    closes stdin so codex knows input is complete.
-    """
-    base_flags = [
-        "--json",
-        "--skip-git-repo-check",
-        "--dangerously-bypass-approvals-and-sandbox",
-        "--model",
-        model,
-    ]
-
-    if thread_id:
-        cmd = ["codex", "exec", *base_flags, "resume", thread_id, "-"]
-    else:
-        cmd = ["codex", "exec", *base_flags, "-"]
-
-    return await asyncio.create_subprocess_exec(
-        *cmd,
-        stdin=asyncio.subprocess.PIPE,
-        stdout=asyncio.subprocess.PIPE,
-        stderr=asyncio.subprocess.PIPE,
-        env={**os.environ},
-    )
-
-
-async def _process_stdout(process: asyncio.subprocess.Process) -> AsyncIterator[str]:
-    """Yield newline-delimited JSON lines from the process stdout."""
-    assert process.stdout is not None
-    buffer = ""
-    while True:
-        chunk = await process.stdout.read(4096)
-        if not chunk:
-            break
-        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()
-
-
 @workflow.defn(name=environment_variables.WORKFLOW_NAME)
 class AtHarnessCodexWorkflow(BaseWorkflow):
     """Long-running Temporal workflow that runs codex exec for each turn.
@@ -149,37 +89,33 @@ async def on_task_event_send(self, params: SendEventParams) -> None:
             name=f"Turn {self._turn_number}",
             input={"message": user_message},
         ) as span:
-            process = await _spawn_codex(MODEL, thread_id=self._codex_thread_id)
-
-            assert process.stdin is not None
-            process.stdin.write(user_message.encode("utf-8"))
-            await process.stdin.drain()
-            process.stdin.close()
-
-            turn = CodexTurn(
-                events=_process_stdout(process),
-                model=MODEL,
-            )
-
-            emitter = UnifiedEmitter(
-                task_id=params.task.id,
-                trace_id=params.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 + codex thread id.
+            # workflow.now() gives a deterministic timestamp under replay.
+            result = await workflow.execute_activity(
+                run_codex_turn,
+                RunCodexTurnParams(
+                    task_id=params.task.id,
+                    prompt=user_message,
+                    model=MODEL,
+                    trace_id=params.task.id,
+                    parent_span_id=span.id if span else None,
+                    thread_id=self._codex_thread_id,
+                    created_at=workflow.now(),
+                ),
+                start_to_close_timeout=timedelta(minutes=5),
             )
 
-            # Pass workflow.now() so Temporal replay stamps messages with a
-            # deterministic timestamp from the workflow clock, not wall time.
-            result = await emitter.auto_send_turn(turn, created_at=workflow.now())
-
-            await process.wait()
-
-            if turn.session_id:
-                self._codex_thread_id = turn.session_id
+            # Persist the codex thread id so the next turn resumes the session.
+            session_id = result.get("session_id")
+            if session_id:
+                self._codex_thread_id = session_id
 
             if span:
                 span.output = {
-                    "final_text": result.final_text,
-                    "model": turn.usage().model,
+                    "final_text": result.get("final_text"),
+                    "model": result.get("model"),
                 }
 
     @workflow.run
diff --git a/examples/tutorials/10_async/10_temporal/harness_codex/tests/test_agent.py b/examples/tutorials/10_async/10_temporal/harness_codex/tests/test_agent.py
