cut a lot of duplicate stuff from python gotchas, address comments

references/python/error-handling.md

@@ -87,6 +87,9 @@ class MyWorkflow:
         return result
 ```
 
+Only set options such as maximum_interval, maximum_attempts etc. if you have a domain-specific reason to.
+If not, prefer to leave them at their defaults.
+
 ## Timeout Configuration
 
 ```python
@@ -101,7 +104,7 @@ class MyWorkflow:
             my_activity,
             start_to_close_timeout=timedelta(minutes=5),      # Single attempt
             schedule_to_close_timeout=timedelta(minutes=30),  # Including retries
-            heartbeat_timeout=timedelta(seconds=30),          # Between heartbeats
+            heartbeat_timeout=timedelta(minutes=2),          # Between heartbeats
         )
 ```
 

references/python/gotchas.md

@@ -1,130 +1,6 @@
 # Python Gotchas
 
-Python-specific mistakes and anti-patterns. See also [Common Gotchas](../core/common-gotchas.md) for language-agnostic concepts.
-
-## Idempotency
-
-```python
-@dataclass
-class ChargePaymentInput:
-    order_id: str
-    amount: float
-
-# BAD - May charge customer multiple times on retry
-@activity.defn
-async def charge_payment(input: ChargePaymentInput) -> str:
-    return await payment_api.charge(input.customer_id, input.amount)
-
-# GOOD - Safe for retries
-@activity.defn
-async def charge_payment(input: ChargePaymentInput) -> str:
-    return await payment_api.charge(
-        input.customer_id,
-        input.amount,
-        idempotency_key=f"order-{input.order_id}"
-    )
-```
-
-## Replay Safety
-
-### Side Effects in Workflows
-
-```python
-# BAD - Prints on every replay, notification runs in workflow
-@workflow.defn
-class NotificationWorkflow:
-    @workflow.run
-    async def run(self):
-        print("Starting workflow")  # Runs on replay too
-        send_slack_notification("Started")  # Side effect in workflow!
-        await workflow.execute_activity(...)
-
-# GOOD - Replay-safe
-@workflow.defn
-class NotificationWorkflow:
-    @workflow.run
-    async def run(self):
-        workflow.logger.info("Starting workflow")  # Only logs on first execution
-        await workflow.execute_activity(send_notification, "Started")
-```
-
-### Time-Based Logic
-
-```python
-# BAD - Different time on replay
-if datetime.now() > deadline:
-    await cancel_order()
-
-# GOOD - Consistent across replays
-if workflow.now() > deadline:
-    await cancel_order()
-```
-
-### Other Non-Deterministic Operations
-
-```python
-# BAD - Different values on replay
-random_id = str(uuid.uuid4())
-random_value = random.random()
-
-# GOOD - Deterministic alternatives
-random_id = workflow.uuid4()
-random_value = workflow.random().random()
-```
-
-## Query Handlers
-
-### Modifying State
-
-```python
-# BAD - Query modifies state
-@workflow.defn
-class QueueWorkflow:
-    def __init__(self):
-        self._queue = []
-
-    @workflow.query
-    def get_next_item(self) -> str | None:
-        if self._queue:
-            return self._queue.pop(0)  # Mutates state!
-        return None
-
-# GOOD - Query reads, Update modifies
-@workflow.defn
-class QueueWorkflow:
-    def __init__(self):
-        self._queue = []
-
-    @workflow.query
-    def peek(self) -> str | None:
-        return self._queue[0] if self._queue else None
-
-    @workflow.update
-    def dequeue(self) -> str | None:
-        if self._queue:
-            return self._queue.pop(0)
-        return None
-```
-
-### Blocking in Queries
-
-```python
-# BAD - Queries cannot await
-@workflow.query
-async def get_data_with_refresh(self) -> dict:
-    if self._data is None:
-        self._data = await workflow.execute_activity(fetch_data, ...)
-    return self._data
-
-# GOOD - Query returns state, signal triggers refresh
-@workflow.signal
-async def refresh_data(self):
-    self._data = await workflow.execute_activity(fetch_data, ...)
-
-@workflow.query
-def get_data(self) -> dict | None:
-    return self._data
-```
+Python-specific mistakes and anti-patterns. See also [Common Gotchas](references/core/common-gotchas.md) for language-agnostic concepts.
 
 ## File Organization
 
@@ -153,6 +29,8 @@ class MyWorkflow:
     pass
 ```
 
+`references/python/sandbox.md` contains more info about the Python sandbox.
+
 ### Mixing Workflows and Activities
 
 ```python
@@ -185,6 +63,8 @@ async def my_activity():
 
 ## Async vs Sync Activities
 
+The Temporal Python SDK supports both async and sync activities. See `references/python/sync-vs-async.md` to understand which to choose. Below are important anti-patterns for both aysnc and sync activities.
+
 ### Blocking in Async Activities
 
 ```python
@@ -218,7 +98,7 @@ async def process_file(path: str) -> str:
 ### Missing Executor for Sync Activities
 
 ```python
-# BAD - Sync activity without executor blocks worker
+# BAD - Sync activity REQUIRES executor
 @activity.defn
 def slow_computation(data: str) -> str:
     return heavy_cpu_work(data)
@@ -227,7 +107,7 @@ Worker(
     client,
     task_queue="my-queue",
     activities=[slow_computation],
-    # Missing activity_executor!
+    # Missing activity_executor! --> THIS IMMEDIATELY RAISES AN EXCEPTION!
 )
 
 # GOOD - Provide executor
@@ -239,76 +119,10 @@ Worker(
 )
 ```
 
-## Error Handling
-
-### Swallowing Errors
-
-```python
-# BAD - Error is hidden
-@workflow.defn
-class SilentFailureWorkflow:
-    @workflow.run
-    async def run(self):
-        try:
-            await workflow.execute_activity(...)
-        except Exception:
-            pass  # Error is lost!
-
-# GOOD - Handle appropriately
-@workflow.defn
-class ProperErrorHandlingWorkflow:
-    @workflow.run
-    async def run(self):
-        try:
-            await workflow.execute_activity(...)
-        except ActivityError as e:
-            workflow.logger.error(f"Activity failed: {e}")
-            raise  # Or use fallback, compensate, etc.
-```
-
-### Wrong Retry Classification
-
-```python
-# BAD - Network errors should be retried
-@activity.defn
-async def call_api():
-    try:
-        return await http_client.get(url)
-    except ConnectionError:
-        raise ApplicationError("Connection failed", non_retryable=True)
-
-# GOOD - Only permanent failures are non-retryable
-@activity.defn
-async def call_api():
-    try:
-        return await http_client.get(url)
-    except ConnectionError:
-        raise  # Let Temporal retry
-    except InvalidCredentialsError:
-        raise ApplicationError("Invalid API key", non_retryable=True)
-```
-
-## Retry Policies
+## Wrong Retry Classification
 
-### Too Aggressive
-
-```python
-# BAD - Gives up too easily
-result = await workflow.execute_activity(
-    flaky_api_call,
-    start_to_close_timeout=timedelta(seconds=30),
-    retry_policy=RetryPolicy(maximum_attempts=1),
-)
-
-# GOOD - Resilient to transient failures
-result = await workflow.execute_activity(
-    flaky_api_call,
-    start_to_close_timeout=timedelta(minutes=10),
-    retry_policy=RetryPolicy(maximum_attempts=10),
-)
-```
-
-Generally, prefer to use the default RetryPolicy.
+**Example:** Transient networks errors should be retried. Authentication errors should not be.
+See `references/python/error-handling.md` to understand how to classify errors.
 
 ## Heartbeating
 
@@ -318,13 +132,13 @@ Generally, prefer to use the default RetryPolicy.
 # BAD - No heartbeat, can't detect stuck activities
 @activity.defn
 async def process_large_file(path: str):
-    for chunk in read_chunks(path):
+    async for chunk in read_chunks(path):
         process(chunk)  # Takes hours, no heartbeat
 
 # GOOD - Regular heartbeats with progress
 @activity.defn
 async def process_large_file(path: str):
-    for i, chunk in enumerate(read_chunks(path)):
+    async for i, chunk in enumerate(read_chunks(path)):
         activity.heartbeat(f"Processing chunk {i}")
         process(chunk)
 ```
@@ -351,12 +165,14 @@ await workflow.execute_activity(
 
 ### Not Testing Failures
 
+Below shows an example of how to test failure cases:
+
 ```python
 # Test failure scenarios
 @pytest.mark.asyncio
 async def test_activity_failure_handling():
-    async with await WorkflowEnvironment.start_time_skipping() as env:
-        # Create activity that always fails
+    async with await WorkflowEnvironment.start_local() as env:
+        # An example activity that always fails
         @activity.defn
         async def failing_activity() -> str:
             raise ApplicationError("Simulated failure", non_retryable=True)
@@ -377,6 +193,8 @@ async def test_activity_failure_handling():
 
 ### Not Testing Replay
 
+Replay tests let you test that you do not have hidden sources of non-determinism bugs in your workflow code:
+
 ```python
 from temporalio.worker import Replayer