From 442f4f4b27e6bb38a1cb0dfd8df98a22a669109a Mon Sep 17 00:00:00 2001
From: barmoshe <1barmoshe1@gmail.com>
Date: Tue, 11 Mar 2025 13:06:51 +0200
Subject: [PATCH 1/4] Add Request/Response sample with activity-based responses
 (fixes #6)

---
 reqrespactivity/readme.md        | 64 ++++++++++++++++++++++++++
 reqrespactivity/requester.py     | 77 ++++++++++++++++++++++++++++++++
 reqrespactivity/requester_run.py | 23 ++++++++++
 reqrespactivity/starter.py       | 17 +++++++
 reqrespactivity/worker.py        | 19 ++++++++
 reqrespactivity/workflow.py      | 60 +++++++++++++++++++++++++
 6 files changed, 260 insertions(+)
 create mode 100644 reqrespactivity/readme.md
 create mode 100644 reqrespactivity/requester.py
 create mode 100644 reqrespactivity/requester_run.py
 create mode 100644 reqrespactivity/starter.py
 create mode 100644 reqrespactivity/worker.py
 create mode 100644 reqrespactivity/workflow.py

diff --git a/reqrespactivity/readme.md b/reqrespactivity/readme.md
new file mode 100644
index 000000000..eb5d3872a
--- /dev/null
+++ b/reqrespactivity/readme.md
@@ -0,0 +1,64 @@
+
+# Request/Response Sample with Activity-Based Responses 
+
+This sample demonstrates how to send a request and get a response from a Temporal workflow via a response activity.
+
+In this example, the workflow accepts requests (signals) to uppercase a string and then provides the response via a callback response activity. Because the response is delivered by an activity execution, the requester must have its own worker running.
+
+## Running
+
+Follow these steps to run the sample:
+
+1. **Run a [Temporal service](https://github.com/temporalio/samples-go/tree/main/#how-to-use):**  
+
+2. **Run the Worker:**  
+   In one terminal, run the worker that executes the workflow and activity:
+   ```bash
+   python worker.py
+   ```
+
+3. **Start the Workflow:**  
+   In another terminal, start the workflow instance:
+   ```bash
+   python starter.py
+   ```
+
+4. **Run the Requester:**  
+   In a third terminal, run the requester that sends a request every second:
+   ```bash
+   python requester_run.py
+   ```
+   This will send requests like `foo0`, `foo1`, etc., to be uppercased by the workflow. You should see output similar to:
+   ```
+   Requested uppercase for 'foo0', got: 'FOO0'
+   Requested uppercase for 'foo1', got: 'FOO1'
+   ...
+   ```
+
+Multiple instances of these processes can be run in separate terminals to confirm that they work independently.
+
+## Comparison with Query-Based Responses
+
+In the [reqrespquery](../reqrespquery) sample, responses are fetched by periodically polling the workflow using queries. This sample, however, uses activity-based responses, which has the following pros and cons:
+
+**Pros:**
+
+* Activity-based responses are often faster due to pushing rather than polling.
+* The workflow does not need to explicitly store the response state.
+* The workflow can detect whether a response was actually received.
+
+**Cons:**
+
+* Activity-based responses require a worker on the caller (requester) side.
+* They record the response in history as an activity execution.
+* They can only occur while the workflow is running.
+
+## Explanation of Continue-As-New
+
+Workflows have a limit on history size. When the event count grows too large, a workflow can return a `ContinueAsNew` error to atomically start a new workflow execution. To prevent data loss, signals must be drained and any pending futures completed before a new execution starts. 
+
+In this sample, which is designed to run long-term, a `ContinueAsNew` is performed once the request count reaches a specified limit, provided there are no in-flight signal requests or executing activities. (If signals are processed too quickly or activities take too long, the workflow might never idle long enough for a `ContinueAsNew` to be triggered.) Careful tuning of signal and activity handling (including setting appropriate retry policies) is essential to ensure that the workflow can transition smoothly to a new execution when needed.
+
+## License
+
+This sample is released under the MIT License.
diff --git a/reqrespactivity/requester.py b/reqrespactivity/requester.py
new file mode 100644
index 000000000..daf63a28b
--- /dev/null
+++ b/reqrespactivity/requester.py
@@ -0,0 +1,77 @@
+# requester.py
+import asyncio
+import uuid
+from dataclasses import dataclass
+from temporalio import activity
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+# Global variable to hold the current Requester instance.
+global_requester_instance = None
+
+@dataclass
+class Request:
+    id: str
+    input: str
+    response_activity: str
+    response_task_queue: str
+
+@dataclass
+class Response:
+    id: str
+    output: str
+    error: str = ""
+
+# Define the response activity as a top-level function with the decorator.
+@activity.defn
+async def response_activity(response: Response):
+    global global_requester_instance
+    if global_requester_instance:
+        fut = global_requester_instance.pending.pop(response.id, None)
+        if fut:
+            fut.set_result(response)
+    else:
+        raise Exception("No requester instance available")
+
+class Requester:
+    def __init__(self, client: Client, target_workflow_id: str):
+        self.client = client
+        self.target_workflow_id = target_workflow_id
+        self.task_queue = "requester-" + str(uuid.uuid4())
+        self.pending = {}  # Maps request IDs to asyncio.Future objects
+
+    async def start_worker(self):
+        global global_requester_instance
+        global_requester_instance = self  # Set the global reference
+        self.worker = Worker(
+            self.client,
+            task_queue=self.task_queue,
+            activities=[response_activity],
+        )
+        # Run the worker in the background.
+        asyncio.create_task(self.worker.run())
+
+    async def close(self):
+        await self.worker.shutdown()
+
+    async def request_uppercase(self, text: str) -> str:
+        req_id = str(uuid.uuid4())
+        req = Request(
+            id=req_id,
+            input=text,
+            response_activity="response_activity",  # Must match the name of the decorated function
+            response_task_queue=self.task_queue,
+        )
+        loop = asyncio.get_running_loop()
+        fut = loop.create_future()
+        self.pending[req_id] = fut
+
+        # Get a handle to the workflow and send the signal.
+        handle = self.client.get_workflow_handle(self.target_workflow_id)
+        await handle.signal("request", req)
+
+        # Wait for the callback to return the response.
+        response: Response = await fut
+        if response.error:
+            raise Exception(response.error)
+        return response.output
diff --git a/reqrespactivity/requester_run.py b/reqrespactivity/requester_run.py
new file mode 100644
index 000000000..3a797c209
--- /dev/null
+++ b/reqrespactivity/requester_run.py
@@ -0,0 +1,23 @@
+# requester_run.py
+import asyncio
+from temporalio.client import Client
+from requester import Requester
+
+async def main():
+    client = await Client.connect("localhost:7233")
+    workflow_id = "reqrespactivity_workflow"
+    requester = Requester(client, workflow_id)
+    await requester.start_worker()
+    try:
+        i = 0
+        while True:
+            text = f"foo{i}"  # Create request similar to the Go sample: foo0, foo1, etc.
+            result = await requester.request_uppercase(text)
+            print(f"Requested uppercase for '{text}', got: '{result}'")
+            await asyncio.sleep(1)
+            i += 1
+    finally:
+        await requester.close()
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/reqrespactivity/starter.py b/reqrespactivity/starter.py
new file mode 100644
index 000000000..24e0e4b9e
--- /dev/null
+++ b/reqrespactivity/starter.py
@@ -0,0 +1,17 @@
+# starter.py
+import asyncio
+from temporalio.client import Client
+from workflow import UppercaseWorkflow
+
+async def main():
+    client = await Client.connect("localhost:7233")
+    workflow_id = "reqrespactivity_workflow"
+    handle = await client.start_workflow(
+        UppercaseWorkflow.run,
+        id=workflow_id,
+        task_queue="reqrespactivity",
+    )
+    print(f"Started workflow with ID: {handle.id}")
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/reqrespactivity/worker.py b/reqrespactivity/worker.py
new file mode 100644
index 000000000..6c864b239
--- /dev/null
+++ b/reqrespactivity/worker.py
@@ -0,0 +1,19 @@
+# worker.py
+import asyncio
+from temporalio.client import Client
+from temporalio.worker import Worker
+from workflow import UppercaseWorkflow, uppercase_activity
+
+async def main():
+    client = await Client.connect("localhost:7233")
+    worker = Worker(
+        client,
+        task_queue="reqrespactivity",
+        workflows=[UppercaseWorkflow],
+        activities=[uppercase_activity],
+    )
+    print("Worker started on task queue 'reqrespactivity'")
+    await worker.run()
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/reqrespactivity/workflow.py b/reqrespactivity/workflow.py
new file mode 100644
index 000000000..83e8a58f6
--- /dev/null
+++ b/reqrespactivity/workflow.py
@@ -0,0 +1,60 @@
+# workflow.py
+import asyncio
+from datetime import timedelta
+from dataclasses import dataclass
+from temporalio import workflow, activity
+
+# Define data models similar to the Go structs.
+@dataclass
+class Request:
+    id: str
+    input: str
+    response_activity: str
+    response_task_queue: str
+
+@dataclass
+class Response:
+    id: str
+    output: str
+    error: str = ""
+
+# Activity to convert text to uppercase.
+@activity.defn
+async def uppercase_activity(text: str) -> str:
+    return text.upper()
+
+# Workflow that listens for "request" signals.
+@workflow.defn
+class UppercaseWorkflow:
+    def __init__(self):
+        self.requests = []  # Buffer for incoming requests
+
+    @workflow.signal
+    def request(self, req: Request):
+        self.requests.append(req)
+
+    @workflow.run
+    async def run(self):
+        # Continuously process incoming requests.
+        while True:
+            if self.requests:
+                req = self.requests.pop(0)
+                try:
+                    # Execute the uppercase activity.
+                    result = await workflow.execute_activity(
+                        uppercase_activity,
+                        req.input,
+                        start_to_close_timeout=timedelta(seconds=5),
+                    )
+                    resp = Response(id=req.id, output=result)
+                except Exception as e:
+                    resp = Response(id=req.id, output="", error=str(e))
+                # Call back the requester via the designated response activity.
+                await workflow.execute_activity(
+                    req.response_activity,
+                    resp,
+                    task_queue=req.response_task_queue,
+                    start_to_close_timeout=timedelta(seconds=10),
+                )
+            else:
+                await workflow.sleep(1)

From 662fa808ba2bea76fb038b6aa53b92cd3574533c Mon Sep 17 00:00:00 2001
From: Bar Moshe <1barmoshe1@gmail.com>
Date: Thu, 30 Jul 2026 13:11:42 +0300
Subject: [PATCH 2/4] Rework Request/Response sample around Workflow Update
 (fixes #6)

Issue #6 and the Go sample it was ported from predate Workflow Update. Per
maintainer feedback, replace the response-activity port with an update-based
sample: the update handler runs the uppercase activity and returns the result
to the caller, removing the response task queue, the callback activity and the
request IDs.

Ports the durability parts of samples-go/reqrespupdate: a request counter,
continue-as-new to bound history, and an update validator that rejects requests
while a run drains so the requester backs off and retries against the fresh run.

Uses workflow.all_handlers_finished() for the drain rather than Go's hand-rolled
pending counter, matching message_passing/safe_message_handlers.
message_passing/waiting_for_handlers notes the continue-as-new drain case is not
illustrated there, so this sample covers it.

Adds tests, which the original sample did not have.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 README.md                            |   1 +
 pyproject.toml                       |   1 +
 reqrespactivity/readme.md            |  64 -----------------
 reqrespactivity/requester.py         |  77 ---------------------
 reqrespactivity/requester_run.py     |  23 ------
 reqrespactivity/starter.py           |  17 -----
 reqrespactivity/worker.py            |  19 -----
 reqrespactivity/workflow.py          |  60 ----------------
 reqrespupdate/README.md              |  56 +++++++++++++++
 reqrespupdate/__init__.py            |   0
 reqrespupdate/activities.py          |   6 ++
 reqrespupdate/requester.py           |  42 +++++++++++
 reqrespupdate/starter.py             |  27 ++++++++
 reqrespupdate/worker.py              |  37 ++++++++++
 reqrespupdate/workflow.py            | 100 +++++++++++++++++++++++++++
 tests/reqrespupdate/__init__.py      |   0
 tests/reqrespupdate/workflow_test.py |  92 ++++++++++++++++++++++++
 17 files changed, 362 insertions(+), 260 deletions(-)
 delete mode 100644 reqrespactivity/readme.md
 delete mode 100644 reqrespactivity/requester.py
 delete mode 100644 reqrespactivity/requester_run.py
 delete mode 100644 reqrespactivity/starter.py
 delete mode 100644 reqrespactivity/worker.py
 delete mode 100644 reqrespactivity/workflow.py
 create mode 100644 reqrespupdate/README.md
 create mode 100644 reqrespupdate/__init__.py
 create mode 100644 reqrespupdate/activities.py
 create mode 100644 reqrespupdate/requester.py
 create mode 100644 reqrespupdate/starter.py
 create mode 100644 reqrespupdate/worker.py
 create mode 100644 reqrespupdate/workflow.py
 create mode 100644 tests/reqrespupdate/__init__.py
 create mode 100644 tests/reqrespupdate/workflow_test.py

diff --git a/README.md b/README.md
index d39428dc7..e8800843b 100644
--- a/README.md
+++ b/README.md
@@ -97,6 +97,7 @@ without wrapping them in a workflow.
 * [pydantic_converter](pydantic_converter) - Data converter for using Pydantic models.
 * [pydantic_converter_v1](pydantic_converter_v1) - Data converter for Pydantic v1 models (prefer pydantic_converter for v2).
 * [replay](replay) - Verify that workflow code changes are compatible with existing histories.
+* [reqrespupdate](reqrespupdate) - Send a request to a long-running workflow and get a response back via an update, with continue-as-new to bound history.
 * [resource_pool](resource_pool) - Allocate a pool of shared resources across workflows.
 * [schedules](schedules) - Demonstrates a Workflow Execution that occurs according to a schedule.
 * [sentry](sentry) - Report errors to Sentry.
diff --git a/pyproject.toml b/pyproject.toml
index e8f3ebd87..076dd2ec6 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -120,6 +120,7 @@ packages = [
     "pydantic_converter_v1",
     "pyproject.toml",
     "replay",
+    "reqrespupdate",
     "schedules",
     "sentry",
     "sleep_for_days",
diff --git a/reqrespactivity/readme.md b/reqrespactivity/readme.md
deleted file mode 100644
index eb5d3872a..000000000
--- a/reqrespactivity/readme.md
+++ /dev/null
@@ -1,64 +0,0 @@
-
-# Request/Response Sample with Activity-Based Responses 
-
-This sample demonstrates how to send a request and get a response from a Temporal workflow via a response activity.
-
-In this example, the workflow accepts requests (signals) to uppercase a string and then provides the response via a callback response activity. Because the response is delivered by an activity execution, the requester must have its own worker running.
-
-## Running
-
-Follow these steps to run the sample:
-
-1. **Run a [Temporal service](https://github.com/temporalio/samples-go/tree/main/#how-to-use):**  
-
-2. **Run the Worker:**  
-   In one terminal, run the worker that executes the workflow and activity:
-   ```bash
-   python worker.py
-   ```
-
-3. **Start the Workflow:**  
-   In another terminal, start the workflow instance:
-   ```bash
-   python starter.py
-   ```
-
-4. **Run the Requester:**  
-   In a third terminal, run the requester that sends a request every second:
-   ```bash
-   python requester_run.py
-   ```
-   This will send requests like `foo0`, `foo1`, etc., to be uppercased by the workflow. You should see output similar to:
-   ```
-   Requested uppercase for 'foo0', got: 'FOO0'
-   Requested uppercase for 'foo1', got: 'FOO1'
-   ...
-   ```
-
-Multiple instances of these processes can be run in separate terminals to confirm that they work independently.
-
-## Comparison with Query-Based Responses
-
-In the [reqrespquery](../reqrespquery) sample, responses are fetched by periodically polling the workflow using queries. This sample, however, uses activity-based responses, which has the following pros and cons:
-
-**Pros:**
-
-* Activity-based responses are often faster due to pushing rather than polling.
-* The workflow does not need to explicitly store the response state.
-* The workflow can detect whether a response was actually received.
-
-**Cons:**
-
-* Activity-based responses require a worker on the caller (requester) side.
-* They record the response in history as an activity execution.
-* They can only occur while the workflow is running.
-
-## Explanation of Continue-As-New
-
-Workflows have a limit on history size. When the event count grows too large, a workflow can return a `ContinueAsNew` error to atomically start a new workflow execution. To prevent data loss, signals must be drained and any pending futures completed before a new execution starts. 
-
-In this sample, which is designed to run long-term, a `ContinueAsNew` is performed once the request count reaches a specified limit, provided there are no in-flight signal requests or executing activities. (If signals are processed too quickly or activities take too long, the workflow might never idle long enough for a `ContinueAsNew` to be triggered.) Careful tuning of signal and activity handling (including setting appropriate retry policies) is essential to ensure that the workflow can transition smoothly to a new execution when needed.
-
-## License
-
-This sample is released under the MIT License.
diff --git a/reqrespactivity/requester.py b/reqrespactivity/requester.py
deleted file mode 100644
index daf63a28b..000000000
--- a/reqrespactivity/requester.py
+++ /dev/null
@@ -1,77 +0,0 @@
-# requester.py
-import asyncio
-import uuid
-from dataclasses import dataclass
-from temporalio import activity
-from temporalio.client import Client
-from temporalio.worker import Worker
-
-# Global variable to hold the current Requester instance.
-global_requester_instance = None
-
-@dataclass
-class Request:
-    id: str
-    input: str
-    response_activity: str
-    response_task_queue: str
-
-@dataclass
-class Response:
-    id: str
-    output: str
-    error: str = ""
-
-# Define the response activity as a top-level function with the decorator.
-@activity.defn
-async def response_activity(response: Response):
-    global global_requester_instance
-    if global_requester_instance:
-        fut = global_requester_instance.pending.pop(response.id, None)
-        if fut:
-            fut.set_result(response)
-    else:
-        raise Exception("No requester instance available")
-
-class Requester:
-    def __init__(self, client: Client, target_workflow_id: str):
-        self.client = client
-        self.target_workflow_id = target_workflow_id
-        self.task_queue = "requester-" + str(uuid.uuid4())
-        self.pending = {}  # Maps request IDs to asyncio.Future objects
-
-    async def start_worker(self):
-        global global_requester_instance
-        global_requester_instance = self  # Set the global reference
-        self.worker = Worker(
-            self.client,
-            task_queue=self.task_queue,
-            activities=[response_activity],
-        )
-        # Run the worker in the background.
-        asyncio.create_task(self.worker.run())
-
-    async def close(self):
-        await self.worker.shutdown()
-
-    async def request_uppercase(self, text: str) -> str:
-        req_id = str(uuid.uuid4())
-        req = Request(
-            id=req_id,
-            input=text,
-            response_activity="response_activity",  # Must match the name of the decorated function
-            response_task_queue=self.task_queue,
-        )
-        loop = asyncio.get_running_loop()
-        fut = loop.create_future()
-        self.pending[req_id] = fut
-
-        # Get a handle to the workflow and send the signal.
-        handle = self.client.get_workflow_handle(self.target_workflow_id)
-        await handle.signal("request", req)
-
-        # Wait for the callback to return the response.
-        response: Response = await fut
-        if response.error:
-            raise Exception(response.error)
-        return response.output
diff --git a/reqrespactivity/requester_run.py b/reqrespactivity/requester_run.py
deleted file mode 100644
index 3a797c209..000000000
--- a/reqrespactivity/requester_run.py
+++ /dev/null
@@ -1,23 +0,0 @@
-# requester_run.py
-import asyncio
-from temporalio.client import Client
-from requester import Requester
-
-async def main():
-    client = await Client.connect("localhost:7233")
-    workflow_id = "reqrespactivity_workflow"
-    requester = Requester(client, workflow_id)
-    await requester.start_worker()
-    try:
-        i = 0
-        while True:
-            text = f"foo{i}"  # Create request similar to the Go sample: foo0, foo1, etc.
-            result = await requester.request_uppercase(text)
-            print(f"Requested uppercase for '{text}', got: '{result}'")
-            await asyncio.sleep(1)
-            i += 1
-    finally:
-        await requester.close()
-
-if __name__ == "__main__":
-    asyncio.run(main())
diff --git a/reqrespactivity/starter.py b/reqrespactivity/starter.py
deleted file mode 100644
index 24e0e4b9e..000000000
--- a/reqrespactivity/starter.py
+++ /dev/null
@@ -1,17 +0,0 @@
-# starter.py
-import asyncio
-from temporalio.client import Client
-from workflow import UppercaseWorkflow
-
-async def main():
-    client = await Client.connect("localhost:7233")
-    workflow_id = "reqrespactivity_workflow"
-    handle = await client.start_workflow(
-        UppercaseWorkflow.run,
-        id=workflow_id,
-        task_queue="reqrespactivity",
-    )
-    print(f"Started workflow with ID: {handle.id}")
-
-if __name__ == "__main__":
-    asyncio.run(main())
diff --git a/reqrespactivity/worker.py b/reqrespactivity/worker.py
deleted file mode 100644
index 6c864b239..000000000
--- a/reqrespactivity/worker.py
+++ /dev/null
@@ -1,19 +0,0 @@
-# worker.py
-import asyncio
-from temporalio.client import Client
-from temporalio.worker import Worker
-from workflow import UppercaseWorkflow, uppercase_activity
-
-async def main():
-    client = await Client.connect("localhost:7233")
-    worker = Worker(
-        client,
-        task_queue="reqrespactivity",
-        workflows=[UppercaseWorkflow],
-        activities=[uppercase_activity],
-    )
-    print("Worker started on task queue 'reqrespactivity'")
-    await worker.run()
-
-if __name__ == "__main__":
-    asyncio.run(main())
diff --git a/reqrespactivity/workflow.py b/reqrespactivity/workflow.py
deleted file mode 100644
index 83e8a58f6..000000000
--- a/reqrespactivity/workflow.py
+++ /dev/null
@@ -1,60 +0,0 @@
-# workflow.py
-import asyncio
-from datetime import timedelta
-from dataclasses import dataclass
-from temporalio import workflow, activity
-
-# Define data models similar to the Go structs.
-@dataclass
-class Request:
-    id: str
-    input: str
-    response_activity: str
-    response_task_queue: str
-
-@dataclass
-class Response:
-    id: str
-    output: str
-    error: str = ""
-
-# Activity to convert text to uppercase.
-@activity.defn
-async def uppercase_activity(text: str) -> str:
-    return text.upper()
-
-# Workflow that listens for "request" signals.
-@workflow.defn
-class UppercaseWorkflow:
-    def __init__(self):
-        self.requests = []  # Buffer for incoming requests
-
-    @workflow.signal
-    def request(self, req: Request):
-        self.requests.append(req)
-
-    @workflow.run
-    async def run(self):
-        # Continuously process incoming requests.
-        while True:
-            if self.requests:
-                req = self.requests.pop(0)
-                try:
-                    # Execute the uppercase activity.
-                    result = await workflow.execute_activity(
-                        uppercase_activity,
-                        req.input,
-                        start_to_close_timeout=timedelta(seconds=5),
-                    )
-                    resp = Response(id=req.id, output=result)
-                except Exception as e:
-                    resp = Response(id=req.id, output="", error=str(e))
-                # Call back the requester via the designated response activity.
-                await workflow.execute_activity(
-                    req.response_activity,
-                    resp,
-                    task_queue=req.response_task_queue,
-                    start_to_close_timeout=timedelta(seconds=10),
-                )
-            else:
-                await workflow.sleep(1)
diff --git a/reqrespupdate/README.md b/reqrespupdate/README.md
new file mode 100644
index 000000000..ecdbcf53a
--- /dev/null
+++ b/reqrespupdate/README.md
@@ -0,0 +1,56 @@
+# Request/Response Sample with Update-Based Responses
+
+This sample shows how to send a request to a running workflow and get a response
+back, using [Workflow Update](https://docs.temporal.io/encyclopedia/workflow-message-passing#updates).
+The update handler runs an activity and returns its result directly to the caller,
+so there is no response task queue, no callback activity and no request IDs to
+correlate.
+
+It is a port of the Go [reqrespupdate](https://github.com/temporalio/samples-go/tree/main/reqrespupdate)
+sample.
+
+To run, open three terminals.
+
+Run the worker in the first:
+
+    uv run reqrespupdate/worker.py
+
+Start the long-running workflow in the second:
+
+    uv run reqrespupdate/starter.py
+
+Then request an uppercasing every second in the third:
+
+    uv run reqrespupdate/requester.py
+
+Several requesters can be run at once, in separate terminals, to confirm they are
+independent of one another.
+
+### Comparison with the other request/response approaches
+
+The Go samples also show request/response via
+[activities](https://github.com/temporalio/samples-go/tree/main/reqrespactivity) and via
+[queries](https://github.com/temporalio/samples-go/tree/main/reqrespquery). Both predate
+Workflow Update. Update is the recommended approach and is the only one ported here.
+
+### Continue-as-new and backpressure
+
+Workflow history cannot grow without limit, so a workflow that fields requests
+indefinitely has to continue-as-new periodically. To avoid losing work, it must do so
+only when no handler is still running, which means there has to be a moment where the
+workflow is idle.
+
+`workflow.all_handlers_finished()` reports exactly that: whether any update or signal
+handler is still executing, including one waiting on an activity retry. The workflow
+waits on it before continuing as new, so no in-flight request is interrupted.
+
+If requests arrive faster than they are handled, that idle moment may never come and
+history keeps growing. The workflow therefore rejects requests from the update
+validator once it is draining toward a continue-as-new. A validator rejection is not
+written to history, which is what makes it the right tool when the problem is history
+size. The requester sees the rejection and retries, and because the retry targets the
+same workflow ID it lands on the fresh run.
+
+The retry policy and timeout on the activity affect how long a handler can stay in
+flight, and therefore how long the workflow can be kept from continuing as new. Set
+them balancing resilience against the need for a period of idleness.
diff --git a/reqrespupdate/__init__.py b/reqrespupdate/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/reqrespupdate/activities.py b/reqrespupdate/activities.py
new file mode 100644
index 000000000..4759d46e3
--- /dev/null
+++ b/reqrespupdate/activities.py
@@ -0,0 +1,6 @@
+from temporalio import activity
+
+
+@activity.defn
+async def uppercase(input: str) -> str:
+    return input.upper()
diff --git a/reqrespupdate/requester.py b/reqrespupdate/requester.py
new file mode 100644
index 000000000..85270b8f4
--- /dev/null
+++ b/reqrespupdate/requester.py
@@ -0,0 +1,42 @@
+import asyncio
+
+from temporalio.client import Client, WorkflowUpdateFailedError
+from temporalio.envconfig import ClientConfig
+from temporalio.exceptions import ApplicationError
+
+from reqrespupdate.starter import WORKFLOW_ID
+from reqrespupdate.workflow import BACKOFF_ERROR_TYPE, Request, UppercaseWorkflow
+
+
+async def main():
+    config = ClientConfig.load_client_connect_config()
+    config.setdefault("target_host", "localhost:7233")
+    client = await Client.connect(**config)
+    handle = client.get_workflow_handle(WORKFLOW_ID)
+
+    # Request an uppercasing every second. Several of these can be run at once,
+    # in separate terminals, to confirm the requesters are independent of each
+    # other.
+    i = 0
+    while True:
+        request = Request(input=f"foo{i}")
+        try:
+            response = await handle.execute_update(UppercaseWorkflow.uppercase, request)
+            print(f"Requested uppercase of {request.input}, got {response.output}")
+            i += 1
+        except WorkflowUpdateFailedError as err:
+            # The run we sent to is draining toward a continue-as-new and asked
+            # us to back off. Retrying sends to the same workflow ID, which by
+            # then is the new run. Any other failure is a real one.
+            if (
+                isinstance(err.cause, ApplicationError)
+                and err.cause.type == BACKOFF_ERROR_TYPE
+            ):
+                print("Rejected while the workflow continues as new, retrying")
+            else:
+                raise
+        await asyncio.sleep(1)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/reqrespupdate/starter.py b/reqrespupdate/starter.py
new file mode 100644
index 000000000..b38d9379d
--- /dev/null
+++ b/reqrespupdate/starter.py
@@ -0,0 +1,27 @@
+import asyncio
+
+from temporalio.client import Client
+from temporalio.envconfig import ClientConfig
+
+from reqrespupdate.workflow import UppercaseWorkflow, UppercaseWorkflowInput
+
+WORKFLOW_ID = "reqrespupdate-workflow-id"
+TASK_QUEUE = "reqrespupdate-task-queue"
+
+
+async def main():
+    config = ClientConfig.load_client_connect_config()
+    config.setdefault("target_host", "localhost:7233")
+    client = await Client.connect(**config)
+
+    handle = await client.start_workflow(
+        UppercaseWorkflow.run,
+        UppercaseWorkflowInput(),
+        id=WORKFLOW_ID,
+        task_queue=TASK_QUEUE,
+    )
+    print(f"Started workflow with ID {handle.id}, now run the requester")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/reqrespupdate/worker.py b/reqrespupdate/worker.py
new file mode 100644
index 000000000..c45612a83
--- /dev/null
+++ b/reqrespupdate/worker.py
@@ -0,0 +1,37 @@
+import asyncio
+import logging
+
+from temporalio.client import Client
+from temporalio.envconfig import ClientConfig
+from temporalio.worker import Worker
+
+from reqrespupdate.activities import uppercase
+from reqrespupdate.workflow import UppercaseWorkflow
+
+interrupt_event = asyncio.Event()
+
+
+async def main():
+    config = ClientConfig.load_client_connect_config()
+    config.setdefault("target_host", "localhost:7233")
+    client = await Client.connect(**config)
+
+    async with Worker(
+        client,
+        task_queue="reqrespupdate-task-queue",
+        workflows=[UppercaseWorkflow],
+        activities=[uppercase],
+    ):
+        logging.info("UppercaseWorkflow worker started, ctrl+c to exit")
+        await interrupt_event.wait()
+        logging.info("Shutting down")
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO)
+    loop = asyncio.new_event_loop()
+    try:
+        loop.run_until_complete(main())
+    except KeyboardInterrupt:
+        interrupt_event.set()
+        loop.run_until_complete(loop.shutdown_asyncgens())
diff --git a/reqrespupdate/workflow.py b/reqrespupdate/workflow.py
new file mode 100644
index 000000000..25d2267f7
--- /dev/null
+++ b/reqrespupdate/workflow.py
@@ -0,0 +1,100 @@
+from dataclasses import dataclass
+from datetime import timedelta
+
+from temporalio import workflow
+from temporalio.common import RetryPolicy
+from temporalio.exceptions import ApplicationError
+
+with workflow.unsafe.imports_passed_through():
+    from reqrespupdate.activities import uppercase
+
+
+# Error type used when the workflow rejects a request because it is continuing
+# as new. The requester matches on this to tell "back off and retry" apart from
+# a genuine failure.
+BACKOFF_ERROR_TYPE = "ContinuingAsNew"
+
+
+# Be in the habit of storing message inputs and outputs in serializable
+# structures. This makes it easier to add more over time in a
+# backward-compatible way.
+@dataclass
+class Request:
+    input: str
+
+
+@dataclass
+class Response:
+    output: str
+
+
+@dataclass
+class UppercaseWorkflowInput:
+    # Workflows cannot have infinitely-sized history, so a workflow that fields
+    # requests forever has to continue-as-new periodically. We bound the number
+    # of requests each run accepts.
+    requests_before_continue_as_new: int = 500
+    # If requests arrive faster than they are handled, the workflow may never
+    # get an idle moment in which to continue-as-new, and history keeps growing.
+    # Rejecting requests from the update validator while a continue-as-new is
+    # pending gives the workflow that idle moment and tells the requester to
+    # back off. A validator rejection is not written to history, which is what
+    # makes it a useful tool when the problem is history size.
+    reject_update_on_pending_continue_as_new: bool = True
+
+
+@workflow.defn
+class UppercaseWorkflow:
+    """A long-running workflow that uppercases strings on request.
+
+    The response is returned directly from the update handler, so there is no
+    response task queue, no callback activity and no request IDs to correlate.
+    """
+
+    @workflow.init
+    def __init__(self, input: UppercaseWorkflowInput) -> None:
+        self.input = input
+        self.request_count = 0
+
+    @workflow.run
+    async def run(self, input: UppercaseWorkflowInput) -> None:
+        # Wait until this run has taken its share of requests and no handler is
+        # still running. all_handlers_finished() accounts for in-flight update
+        # handlers, including ones waiting on an activity retry, so there is no
+        # need to track pending handlers by hand.
+        await workflow.wait_condition(
+            lambda: self.continue_as_new_pending() and workflow.all_handlers_finished()
+        )
+        workflow.continue_as_new(input)
+
+    @workflow.update
+    async def uppercase(self, request: Request) -> Response:
+        self.request_count += 1
+        # WARNING: the timeout and retry policy affect how long this handler can
+        # stay in flight, and therefore how long the workflow can be prevented
+        # from continuing as new. Set them balancing resilience against the need
+        # for a period of idleness.
+        output = await workflow.execute_activity(
+            uppercase,
+            request.input,
+            schedule_to_close_timeout=timedelta(seconds=5),
+            retry_policy=RetryPolicy(maximum_attempts=2),
+        )
+        return Response(output=output)
+
+    @uppercase.validator
+    def validate_uppercase(self, request: Request) -> None:
+        # Raising from a validator rejects the update without writing it to
+        # history, and without the workflow having to handle it. Note that a
+        # validator must be synchronous.
+        if (
+            self.input.reject_update_on_pending_continue_as_new
+            and self.continue_as_new_pending()
+        ):
+            raise ApplicationError(
+                "Workflow is continuing as new, please retry",
+                type=BACKOFF_ERROR_TYPE,
+            )
+
+    def continue_as_new_pending(self) -> bool:
+        return self.request_count >= self.input.requests_before_continue_as_new
diff --git a/tests/reqrespupdate/__init__.py b/tests/reqrespupdate/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/reqrespupdate/workflow_test.py b/tests/reqrespupdate/workflow_test.py
new file mode 100644
index 000000000..2b0bacb90
--- /dev/null
+++ b/tests/reqrespupdate/workflow_test.py
@@ -0,0 +1,92 @@
+import uuid
+
+import pytest
+from temporalio.client import Client, WorkflowUpdateFailedError
+from temporalio.exceptions import ApplicationError
+from temporalio.testing import WorkflowEnvironment
+from temporalio.worker import Worker
+
+from reqrespupdate.activities import uppercase
+from reqrespupdate.workflow import (
+    BACKOFF_ERROR_TYPE,
+    Request,
+    UppercaseWorkflow,
+    UppercaseWorkflowInput,
+)
+
+
+async def request_uppercase(handle, text: str) -> str:
+    """Request an uppercasing, retrying if the workflow is continuing as new.
+
+    This is the same backoff the requester in this sample performs.
+    """
+    while True:
+        try:
+            response = await handle.execute_update(
+                UppercaseWorkflow.uppercase, Request(input=text)
+            )
+            return response.output
+        except WorkflowUpdateFailedError as err:
+            if (
+                isinstance(err.cause, ApplicationError)
+                and err.cause.type == BACKOFF_ERROR_TYPE
+            ):
+                continue
+            raise
+
+
+async def test_uppercase(client: Client, env: WorkflowEnvironment):
+    if env.supports_time_skipping:
+        pytest.skip("Java test server does not support workflow update")
+    task_queue = f"tq-{uuid.uuid4()}"
+    async with Worker(
+        client,
+        task_queue=task_queue,
+        workflows=[UppercaseWorkflow],
+        activities=[uppercase],
+    ):
+        handle = await client.start_workflow(
+            UppercaseWorkflow.run,
+            UppercaseWorkflowInput(),
+            id=f"reqrespupdate-{uuid.uuid4()}",
+            task_queue=task_queue,
+        )
+        try:
+            for text in ["foo", "bar", "baz"]:
+                assert await request_uppercase(handle, text) == text.upper()
+        finally:
+            await handle.terminate()
+
+
+async def test_continues_as_new_without_losing_requests(
+    client: Client, env: WorkflowEnvironment
+):
+    if env.supports_time_skipping:
+        pytest.skip("Java test server does not support workflow update")
+    task_queue = f"tq-{uuid.uuid4()}"
+    async with Worker(
+        client,
+        task_queue=task_queue,
+        workflows=[UppercaseWorkflow],
+        activities=[uppercase],
+    ):
+        handle = await client.start_workflow(
+            UppercaseWorkflow.run,
+            # Low enough that the run continues as new several times over the
+            # requests below.
+            UppercaseWorkflowInput(requests_before_continue_as_new=2),
+            id=f"reqrespupdate-{uuid.uuid4()}",
+            task_queue=task_queue,
+        )
+        try:
+            # Whether any individual request is rejected depends on how the
+            # continue-as-new interleaves with it, so we assert the contract the
+            # requester actually relies on: every request eventually succeeds,
+            # and the workflow does continue as new underneath.
+            for i in range(6):
+                assert await request_uppercase(handle, f"foo{i}") == f"FOO{i}"
+
+            description = await client.get_workflow_handle(handle.id).describe()
+            assert description.run_id != handle.first_execution_run_id
+        finally:
+            await client.get_workflow_handle(handle.id).terminate()

From 4245b7c0ed97061899e5102d4ed1428950055071 Mon Sep 17 00:00:00 2001
From: Bar Moshe <1barmoshe1@gmail.com>
Date: Thu, 30 Jul 2026 13:23:35 +0300
Subject: [PATCH 3/4] Run reqrespupdate tests on the time-skipping server too

The tests were skipping when env.supports_time_skipping, on the assumption
that the Java test server does not support workflow update. It does: both
tests pass against it, repeatedly. CI runs the suite twice, once per
environment, so the guard was halving this sample's coverage for no reason.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 tests/reqrespupdate/workflow_test.py | 12 ++----------
 1 file changed, 2 insertions(+), 10 deletions(-)

diff --git a/tests/reqrespupdate/workflow_test.py b/tests/reqrespupdate/workflow_test.py
index 2b0bacb90..1cf9c3ca7 100644
--- a/tests/reqrespupdate/workflow_test.py
+++ b/tests/reqrespupdate/workflow_test.py
@@ -1,9 +1,7 @@
 import uuid
 
-import pytest
 from temporalio.client import Client, WorkflowUpdateFailedError
 from temporalio.exceptions import ApplicationError
-from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import Worker
 
 from reqrespupdate.activities import uppercase
@@ -35,9 +33,7 @@ async def request_uppercase(handle, text: str) -> str:
             raise
 
 
-async def test_uppercase(client: Client, env: WorkflowEnvironment):
-    if env.supports_time_skipping:
-        pytest.skip("Java test server does not support workflow update")
+async def test_uppercase(client: Client):
     task_queue = f"tq-{uuid.uuid4()}"
     async with Worker(
         client,
@@ -58,11 +54,7 @@ async def test_uppercase(client: Client, env: WorkflowEnvironment):
             await handle.terminate()
 
 
-async def test_continues_as_new_without_losing_requests(
-    client: Client, env: WorkflowEnvironment
-):
-    if env.supports_time_skipping:
-        pytest.skip("Java test server does not support workflow update")
+async def test_continues_as_new_without_losing_requests(client: Client):
     task_queue = f"tq-{uuid.uuid4()}"
     async with Worker(
         client,

From 6e93145567c2ec7344a554bd9f8dff1d4ab415cf Mon Sep 17 00:00:00 2001
From: Bar Moshe <1barmoshe1@gmail.com>
Date: Thu, 30 Jul 2026 20:08:18 +0300
Subject: [PATCH 4/4] Address review feedback on reqrespupdate

- Bound the test retry helper with a max attempt count and a short sleep, so
  a workflow that stops continuing as new fails the test instead of retrying
  forever and hanging CI.
- Skip both tests on the time-skipping server, matching the other update
  tests in the repo (temporalio/sdk-java#1903).
- Reject requests_before_continue_as_new < 1 in the workflow constructor.
  Zero made the run continue as new immediately and forever while rejecting
  every request.
- Move TASK_QUEUE and WORKFLOW_ID into __init__.py, as sleep_for_days and
  message_passing/waiting_for_handlers do, instead of the worker hardcoding
  the task queue name.
- Use the existing handle in the continue-as-new test rather than re-fetching
  it; start_workflow does not pin the handle to a run.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 reqrespupdate/__init__.py            |  2 ++
 reqrespupdate/requester.py           |  2 +-
 reqrespupdate/starter.py             |  4 +---
 reqrespupdate/worker.py              |  3 ++-
 reqrespupdate/workflow.py            |  7 ++++++
 tests/reqrespupdate/workflow_test.py | 33 ++++++++++++++++++++++------
 6 files changed, 39 insertions(+), 12 deletions(-)

diff --git a/reqrespupdate/__init__.py b/reqrespupdate/__init__.py
index e69de29bb..c2ede4555 100644
--- a/reqrespupdate/__init__.py
+++ b/reqrespupdate/__init__.py
@@ -0,0 +1,2 @@
+TASK_QUEUE = "reqrespupdate-task-queue"
+WORKFLOW_ID = "reqrespupdate-workflow-id"
diff --git a/reqrespupdate/requester.py b/reqrespupdate/requester.py
index 85270b8f4..8db55baa9 100644
--- a/reqrespupdate/requester.py
+++ b/reqrespupdate/requester.py
@@ -4,7 +4,7 @@
 from temporalio.envconfig import ClientConfig
 from temporalio.exceptions import ApplicationError
 
-from reqrespupdate.starter import WORKFLOW_ID
+from reqrespupdate import WORKFLOW_ID
 from reqrespupdate.workflow import BACKOFF_ERROR_TYPE, Request, UppercaseWorkflow
 
 
diff --git a/reqrespupdate/starter.py b/reqrespupdate/starter.py
index b38d9379d..d0c086091 100644
--- a/reqrespupdate/starter.py
+++ b/reqrespupdate/starter.py
@@ -3,11 +3,9 @@
 from temporalio.client import Client
 from temporalio.envconfig import ClientConfig
 
+from reqrespupdate import TASK_QUEUE, WORKFLOW_ID
 from reqrespupdate.workflow import UppercaseWorkflow, UppercaseWorkflowInput
 
-WORKFLOW_ID = "reqrespupdate-workflow-id"
-TASK_QUEUE = "reqrespupdate-task-queue"
-
 
 async def main():
     config = ClientConfig.load_client_connect_config()
diff --git a/reqrespupdate/worker.py b/reqrespupdate/worker.py
index c45612a83..43cea3135 100644
--- a/reqrespupdate/worker.py
+++ b/reqrespupdate/worker.py
@@ -5,6 +5,7 @@
 from temporalio.envconfig import ClientConfig
 from temporalio.worker import Worker
 
+from reqrespupdate import TASK_QUEUE
 from reqrespupdate.activities import uppercase
 from reqrespupdate.workflow import UppercaseWorkflow
 
@@ -18,7 +19,7 @@ async def main():
 
     async with Worker(
         client,
-        task_queue="reqrespupdate-task-queue",
+        task_queue=TASK_QUEUE,
         workflows=[UppercaseWorkflow],
         activities=[uppercase],
     ):
diff --git a/reqrespupdate/workflow.py b/reqrespupdate/workflow.py
index 25d2267f7..405bec620 100644
--- a/reqrespupdate/workflow.py
+++ b/reqrespupdate/workflow.py
@@ -53,6 +53,13 @@ class UppercaseWorkflow:
 
     @workflow.init
     def __init__(self, input: UppercaseWorkflowInput) -> None:
+        # A run has to accept at least one request, otherwise it continues as
+        # new the moment it starts and the chain never does any work.
+        if input.requests_before_continue_as_new < 1:
+            raise ApplicationError(
+                "requests_before_continue_as_new must be at least 1",
+                non_retryable=True,
+            )
         self.input = input
         self.request_count = 0
 
diff --git a/tests/reqrespupdate/workflow_test.py b/tests/reqrespupdate/workflow_test.py
index 1cf9c3ca7..5084adb92 100644
--- a/tests/reqrespupdate/workflow_test.py
+++ b/tests/reqrespupdate/workflow_test.py
@@ -1,7 +1,10 @@
+import asyncio
 import uuid
 
+import pytest
 from temporalio.client import Client, WorkflowUpdateFailedError
 from temporalio.exceptions import ApplicationError
+from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import Worker
 
 from reqrespupdate.activities import uppercase
@@ -13,12 +16,14 @@
 )
 
 
-async def request_uppercase(handle, text: str) -> str:
+async def request_uppercase(handle, text: str, max_attempts: int = 10) -> str:
     """Request an uppercasing, retrying if the workflow is continuing as new.
 
-    This is the same backoff the requester in this sample performs.
+    This is the same backoff the requester in this sample performs, bounded so
+    that a workflow which never continues as new fails the test instead of
+    retrying forever.
     """
-    while True:
+    for _ in range(max_attempts):
         try:
             response = await handle.execute_update(
                 UppercaseWorkflow.uppercase, Request(input=text)
@@ -29,11 +34,19 @@ async def request_uppercase(handle, text: str) -> str:
                 isinstance(err.cause, ApplicationError)
                 and err.cause.type == BACKOFF_ERROR_TYPE
             ):
+                await asyncio.sleep(0.1)
                 continue
             raise
+    raise AssertionError(
+        f"Request for {text} still rejected after {max_attempts} attempts"
+    )
 
 
-async def test_uppercase(client: Client):
+async def test_uppercase(client: Client, env: WorkflowEnvironment):
+    if env.supports_time_skipping:
+        pytest.skip(
+            "Java test server: https://github.com/temporalio/sdk-java/issues/1903"
+        )
     task_queue = f"tq-{uuid.uuid4()}"
     async with Worker(
         client,
@@ -54,7 +67,13 @@ async def test_uppercase(client: Client):
             await handle.terminate()
 
 
-async def test_continues_as_new_without_losing_requests(client: Client):
+async def test_continues_as_new_without_losing_requests(
+    client: Client, env: WorkflowEnvironment
+):
+    if env.supports_time_skipping:
+        pytest.skip(
+            "Java test server: https://github.com/temporalio/sdk-java/issues/1903"
+        )
     task_queue = f"tq-{uuid.uuid4()}"
     async with Worker(
         client,
@@ -78,7 +97,7 @@ async def test_continues_as_new_without_losing_requests(client: Client):
             for i in range(6):
                 assert await request_uppercase(handle, f"foo{i}") == f"FOO{i}"
 
-            description = await client.get_workflow_handle(handle.id).describe()
+            description = await handle.describe()
             assert description.run_id != handle.first_execution_run_id
         finally:
-            await client.get_workflow_handle(handle.id).terminate()
+            await handle.terminate()