test: improve test_workflow_logging_replay structure and cleanup

- Add try/finally to restore full_workflow_info_on_extra after test
- Improve docstring to explain what this test validates vs parameterized test
- Add clear phase comments for first execution vs replay path

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: harsh543

tests/worker/test_workflow.py

@@ -2051,56 +2051,67 @@ async def test_workflow_logging(client: Client, temporal_extra_mode: str):
 
 
 async def test_workflow_logging_replay(client: Client):
-    """Test that replayed logs are suppressed and full_workflow_info_on_extra works."""
+    """Test replay log suppression and full_workflow_info_on_extra.
+
+    This test validates behavior the parameterized dict/flatten test does NOT cover:
+    1. Replay suppression: logs from replayed history should not be emitted again.
+    2. full_workflow_info_on_extra: when enabled, logs include workflow.Info object.
+    """
+    original_full_info = workflow.logger.full_workflow_info_on_extra
     workflow.logger.full_workflow_info_on_extra = True
-    with LogCapturer().logs_captured(
-        workflow.logger.base_logger, activity.logger.base_logger
-    ) as capturer:
-        # Log two signals and kill worker before completing. Need to disable
-        # workflow cache since we restart the worker and don't want to pay the
-        # sticky queue penalty.
-        async with new_worker(
-            client, LoggingWorkflow, max_cached_workflows=0
-        ) as worker:
-            handle = await client.start_workflow(
-                LoggingWorkflow.run,
-                id=f"workflow-{uuid.uuid4()}",
-                task_queue=worker.task_queue,
-            )
-            await handle.signal(LoggingWorkflow.my_signal, "signal 1")
-            await handle.signal(LoggingWorkflow.my_signal, "signal 2")
-            await handle.execute_update(
-                LoggingWorkflow.my_update, "update 1", id="update-1"
-            )
-            assert "signal 2" == await handle.query(LoggingWorkflow.last_signal)
+    try:
+        with LogCapturer().logs_captured(
+            workflow.logger.base_logger, activity.logger.base_logger
+        ) as capturer:
+            # --- First execution: logs should appear ---
+            # Disable workflow cache so worker restart triggers replay
+            async with new_worker(
+                client, LoggingWorkflow, max_cached_workflows=0
+            ) as worker:
+                handle = await client.start_workflow(
+                    LoggingWorkflow.run,
+                    id=f"workflow-{uuid.uuid4()}",
+                    task_queue=worker.task_queue,
+                )
+                await handle.signal(LoggingWorkflow.my_signal, "signal 1")
+                await handle.signal(LoggingWorkflow.my_signal, "signal 2")
+                await handle.execute_update(
+                    LoggingWorkflow.my_update, "update 1", id="update-1"
+                )
+                assert "signal 2" == await handle.query(LoggingWorkflow.last_signal)
 
-        # Confirm logs were produced
-        assert capturer.find_log("Signal: signal 1")
-        assert capturer.find_log("Signal: signal 2")
-        assert capturer.find_log("Update: update 1")
-        assert capturer.find_log("Query called")
+            # Verify logs from first execution
+            assert capturer.find_log("Signal: signal 1")
+            assert capturer.find_log("Signal: signal 2")
+            assert capturer.find_log("Update: update 1")
+            assert capturer.find_log("Query called")
 
-        # Since we enabled full info, make sure it's there
-        record = capturer.find_log("Signal: signal 1")
-        assert record and isinstance(record.__dict__["workflow_info"], workflow.Info)
+            # Verify full_workflow_info_on_extra
+            record = capturer.find_log("Signal: signal 1")
+            assert record and isinstance(record.__dict__["workflow_info"], workflow.Info)
 
-        # Clear queue and start a new one with more signals
-        capturer.log_queue.queue.clear()
-        async with new_worker(
-            client,
-            LoggingWorkflow,
-            task_queue=worker.task_queue,
-            max_cached_workflows=0,
-        ) as worker:
-            await handle.signal(LoggingWorkflow.my_signal, "signal 3")
-            await handle.signal(LoggingWorkflow.my_signal, "finish")
-            await handle.result()
+            # --- Clear logs and continue execution (replay path) ---
+            # When the new worker starts, it replays the workflow history (signals 1 & 2).
+            # Replay suppression should prevent those logs from appearing again.
+            capturer.log_queue.queue.clear()
 
-        # Confirm replayed logs are not present but new ones are
-        assert not capturer.find_log("Signal: signal 1")
-        assert not capturer.find_log("Signal: signal 2")
-        assert capturer.find_log("Signal: signal 3")
-        assert capturer.find_log("Signal: finish")
+            async with new_worker(
+                client,
+                LoggingWorkflow,
+                task_queue=worker.task_queue,
+                max_cached_workflows=0,
+            ) as worker:
+                await handle.signal(LoggingWorkflow.my_signal, "signal 3")
+                await handle.signal(LoggingWorkflow.my_signal, "finish")
+                await handle.result()
+
+            # --- Replay execution: no duplicate logs ---
+            assert not capturer.find_log("Signal: signal 1")
+            assert not capturer.find_log("Signal: signal 2")
+            assert capturer.find_log("Signal: signal 3")
+            assert capturer.find_log("Signal: finish")
+    finally:
+        workflow.logger.full_workflow_info_on_extra = original_full_info
 
 
 @activity.defn