Refactor ReplaySafeLogger to use LoggerAdapter, add tests and docs

- Rewrite ReplaySafeLogger as a logging.LoggerAdapter subclass with a
  single isEnabledFor() override instead of manual method delegation
- Add log() and isEnabledFor() methods with tests
- Add replay-safe logging section to docs/features.md
- Add usage example in examples/activity_sequence.py
- Add cross-package compatibility section to copilot-instructions.md

Author: andystaples

.github/copilot-instructions.md

@@ -107,3 +107,18 @@ python -m pytest
 - `examples/` — example orchestrations (see `examples/README.md`)
 - `tests/` — test suite
 - `dev-requirements.txt` — development dependencies
+
+## Cross-Package Compatibility
+
+The `durabletask-azuremanaged` package extends the core `durabletask`
+package (e.g. `DurableTaskSchedulerWorker` subclasses
+`TaskHubGrpcWorker`). When adding or changing features in
+`durabletask/`, always verify that `durabletask-azuremanaged` still
+works correctly:
+
+- Check whether the azuremanaged worker, client, or tests override or
+  depend on the code you changed.
+- Run the azuremanaged unit tests if they exist for the affected area.
+- If a new public API is added to the core SDK (e.g. a method on
+  `OrchestrationContext`), confirm it is accessible through the
+  azuremanaged package and add a test or example if appropriate.

docs/features.md

@@ -150,6 +150,47 @@ Orchestrations can be suspended using the `suspend_orchestration` client API and
 
 Orchestrations can specify retry policies for activities and sub-orchestrations. These policies control how many times and how frequently an activity or sub-orchestration will be retried in the event of a transient error.
 
+### Replay-safe logging
+
+Orchestrator functions replay their history each time they are resumed,
+which can cause duplicate log messages. The `create_replay_safe_logger`
+method on `OrchestrationContext` returns a `ReplaySafeLogger` that wraps
+a standard `logging.Logger` and automatically suppresses output while
+the orchestrator is replaying. `ReplaySafeLogger` extends Python's
+`logging.LoggerAdapter`, which is the idiomatic way to add context or
+modify behavior on an existing logger.
+
+```python
+import logging
+
+logger = logging.getLogger("my_orchestrator")
+
+def my_orchestrator(ctx: task.OrchestrationContext, input):
+    replay_logger = ctx.create_replay_safe_logger(logger)
+    replay_logger.info("Starting orchestration %s", ctx.instance_id)
+    result = yield ctx.call_activity(my_activity, input=input)
+    replay_logger.info("Activity returned: %s", result)
+    return result
+```
+
+> [!NOTE]
+> Unlike the .NET SDK, where `CreateReplaySafeLogger` accepts a
+> category name string and internally creates the logger via
+> `ILoggerFactory`, the Python SDK requires you to pass an existing
+> `logging.Logger` instance. This is because Python's
+> `logging.getLogger(name)` already serves as the global factory and
+> is the standard way to obtain loggers.
+
+The replay-safe logger supports all standard log levels: `debug`,
+`info`, `warning`, `error`, `critical`, and `exception`, as well as
+the generic `log(level, msg)` method. It also exposes `isEnabledFor`
+which returns `False` during replay so callers can skip expensive
+message formatting.
+
+> [!TIP]
+> Create the replay-safe logger once at the start of your orchestrator
+> and reuse it throughout the function.
+
 ### Large payload externalization
 
 Orchestration inputs, outputs, and event data are transmitted through

durabletask/task.py

@@ -300,10 +300,10 @@ def create_replay_safe_logger(self, logger: logging.Logger) -> ReplaySafeLogger:
         return ReplaySafeLogger(logger, lambda: self.is_replaying)
 
 
-class ReplaySafeLogger:
-    """A logger wrapper that suppresses log messages during orchestration replay.
+class ReplaySafeLogger(logging.LoggerAdapter):
+    """A logger adapter that suppresses log messages during orchestration replay.
 
-    This class wraps a standard :class:`logging.Logger` and only emits log
+    This class extends :class:`logging.LoggerAdapter` and only emits log
     messages when the orchestrator is *not* replaying. Use this to avoid
     duplicate log entries that would otherwise appear every time the
     orchestrator replays its history.
@@ -312,41 +312,18 @@ class ReplaySafeLogger:
     """
 
     def __init__(self, logger: logging.Logger, is_replaying: Callable[[], bool]) -> None:
-        self._logger = logger
+        super().__init__(logger, {})
         self._is_replaying = is_replaying
 
-    def _should_log(self) -> bool:
-        return not self._is_replaying()
-
-    def debug(self, msg: str, *args: Any, **kwargs: Any) -> None:
-        """Log a DEBUG-level message if the orchestrator is not replaying."""
-        if self._should_log():
-            self._logger.debug(msg, *args, **kwargs)
-
-    def info(self, msg: str, *args: Any, **kwargs: Any) -> None:
-        """Log an INFO-level message if the orchestrator is not replaying."""
-        if self._should_log():
-            self._logger.info(msg, *args, **kwargs)
-
-    def warning(self, msg: str, *args: Any, **kwargs: Any) -> None:
-        """Log a WARNING-level message if the orchestrator is not replaying."""
-        if self._should_log():
-            self._logger.warning(msg, *args, **kwargs)
-
-    def error(self, msg: str, *args: Any, **kwargs: Any) -> None:
-        """Log an ERROR-level message if the orchestrator is not replaying."""
-        if self._should_log():
-            self._logger.error(msg, *args, **kwargs)
-
-    def critical(self, msg: str, *args: Any, **kwargs: Any) -> None:
-        """Log a CRITICAL-level message if the orchestrator is not replaying."""
-        if self._should_log():
-            self._logger.critical(msg, *args, **kwargs)
-
-    def exception(self, msg: str, *args: Any, **kwargs: Any) -> None:
-        """Log an ERROR-level message with exception info if the orchestrator is not replaying."""
-        if self._should_log():
-            self._logger.exception(msg, *args, **kwargs)
+    def isEnabledFor(self, level: int) -> bool:
+        """Return whether logging is enabled for the given level.
+
+        Returns ``False`` while the orchestrator is replaying so that callers
+        can skip expensive message formatting during replay.
+        """
+        if self._is_replaying():
+            return False
+        return self.logger.isEnabledFor(level)
 
 
 class FailureDetails:

examples/activity_sequence.py

@@ -1,5 +1,6 @@
 """End-to-end sample that demonstrates how to configure an orchestrator
 that calls an activity function in a sequence and prints the outputs."""
+import logging
 import os
 
 from azure.identity import DefaultAzureCredential
@@ -8,6 +9,8 @@
 from durabletask.azuremanaged.client import DurableTaskSchedulerClient
 from durabletask.azuremanaged.worker import DurableTaskSchedulerWorker
 
+logger = logging.getLogger("activity_sequence")
+
 
 def hello(ctx: task.ActivityContext, name: str) -> str:
     """Activity function that returns a greeting"""
@@ -16,10 +19,15 @@ def hello(ctx: task.ActivityContext, name: str) -> str:
 
 def sequence(ctx: task.OrchestrationContext, _):
     """Orchestrator function that calls the 'hello' activity function in a sequence"""
+    # Create a replay-safe logger to avoid duplicate log messages during replay
+    replay_logger = ctx.create_replay_safe_logger(logger)
+
+    replay_logger.info("Starting activity sequence for instance %s", ctx.instance_id)
     # call "hello" activity function in a sequence
     result1 = yield ctx.call_activity(hello, input='Tokyo')
     result2 = yield ctx.call_activity(hello, input='Seattle')
     result3 = yield ctx.call_activity(hello, input='London')
+    replay_logger.info("All activities completed")
 
     # return an array of results
     return [result1, result2, result3]

tests/durabletask/test_orchestration_executor.py

@@ -1269,6 +1269,53 @@ def emit(self, record: logging.LogRecord) -> None:
     assert log_calls == ["should appear"]
 
 
+def test_replay_safe_logger_log_method():
+    """Validates the generic log() method respects the replay flag."""
+    log_calls: list[str] = []
+
+    class _RecordingHandler(logging.Handler):
+        def emit(self, record: logging.LogRecord) -> None:
+            log_calls.append(record.getMessage())
+
+    inner_logger = logging.getLogger("test_replay_safe_logger_log_method")
+    inner_logger.setLevel(logging.DEBUG)
+    inner_logger.addHandler(_RecordingHandler())
+
+    replaying = True
+    replay_logger = task.ReplaySafeLogger(inner_logger, lambda: replaying)
+
+    replay_logger.log(logging.WARNING, "suppressed warning")
+    assert log_calls == []
+
+    replaying = False
+    replay_logger.log(logging.WARNING, "visible warning")
+    assert log_calls == ["visible warning"]
+
+
+def test_replay_safe_logger_is_enabled_for():
+    """Validates isEnabledFor returns False during replay."""
+    inner_logger = logging.getLogger("test_replay_safe_logger_enabled")
+    inner_logger.setLevel(logging.DEBUG)
+
+    replaying = True
+    replay_logger = task.ReplaySafeLogger(inner_logger, lambda: replaying)
+
+    # During replay, isEnabledFor should always return False
+    assert replay_logger.isEnabledFor(logging.DEBUG) is False
+    assert replay_logger.isEnabledFor(logging.INFO) is False
+    assert replay_logger.isEnabledFor(logging.CRITICAL) is False
+
+    # After replay, delegates to the inner logger
+    replaying = False
+    assert replay_logger.isEnabledFor(logging.DEBUG) is True
+    assert replay_logger.isEnabledFor(logging.INFO) is True
+
+    # If a level is below the inner logger's level, should return False
+    inner_logger.setLevel(logging.WARNING)
+    assert replay_logger.isEnabledFor(logging.DEBUG) is False
+    assert replay_logger.isEnabledFor(logging.WARNING) is True
+
+
 def test_when_any_with_retry():
     """Tests that a when_any pattern works correctly with retries"""
     def dummy_activity(_, inp: str):