Add replay-aware workflow logging

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 ## Unreleased
 
+- Workflow code can use `ctx.logger()` to send new-execution observations to
+  the application's `tracing` subscriber with namespace/workflow/run identity.
+  Committed replay, query reconstruction and discarded invocations stay quiet.
+  `ctx.replay_state()` exposes the live consumed-history boundary. Logs from
+  uncommitted task attempts may repeat on retry; handlers must be nonblocking.
+
 ## 4.16.0
 
 - Completion payload planning reads the optional advertised

README.md

@@ -495,6 +495,41 @@ Keep workflow command order deterministic:
 See the runnable [`workflow_testing`](examples/workflow_testing.rs) example and
 [`workflow_testing` tests](tests/workflow_testing.rs).
 
+## Workflow logging
+
+Use `ctx.logger()` for observations written from workflow code. It sends events
+to the application's existing `tracing` subscriber with the target
+`durable_workflow::workflow` and SDK-bound `namespace`, `workflow_instance_id`
+and `workflow_run_id` fields. No global subscriber is installed.
+
+```rust,no_run
+# use durable_workflow::{Client, Worker, json};
+# fn configure(client: Client) {
+let mut worker = Worker::new(client, "orders");
+worker.register_workflow("order", |ctx, _input| async move {
+    ctx.logger().info("processing order");
+    let result = ctx.activity("charge", json!([])).await?;
+    ctx.logger().info("charge completed");
+    Ok(result)
+});
+# }
+```
+
+Ordinary `tracing` calls execute again when a worker reconstructs a workflow.
+The workflow logger suppresses the committed prefix, query/offline replay and
+completed histories, including after a cold worker replacement. It emits when
+execution reaches new code beyond that prefix. `ctx.replay_state()` exposes
+that live boundary as `WorkflowReplayState`; a retained context outside its
+invocation is `Unavailable`. Use the logger for observations and keep durable
+command order independent of the observation state.
+
+This is process logging. A log from an uncommitted task attempt can repeat if
+that attempt crashes or retries. It is not an exactly-once outbox. Keep host
+handlers bounded and nonblocking, for example with an application-owned logging
+queue, so they do not stall the worker. Unwinding subscriber panics are isolated
+from the workflow result; aborting processes and blocking handlers remain the
+application's responsibility.
+
 ## Inspect a run's activities
 
 Use `Client::get_run_activities(workflow_id, run_id)` next to the selected run's

regression-corpus-policy.json

@@ -72,6 +72,12 @@
         {
           "glob": "src/replay_clock.rs"
         },
+        {
+          "glob": "src/workflow_logging.rs"
+        },
+        {
+          "glob": "src/nexus.rs"
+        },
         {
           "glob": "src/replay_clock/condition.rs"
         },

src/lib.rs

@@ -36,6 +36,8 @@ mod update_outcome;
 mod worker_capacity;
 mod worker_session;
 mod workflow_hydration;
+mod workflow_logging;
+pub use workflow_logging::{WorkflowLogger, WorkflowReplayState};
 
 pub use activity_error::ActivityError;
 pub use activity_retry::{ActivityAttemptLimit, ActivityRetryInfo};
@@ -10242,6 +10244,7 @@ impl Worker {
             self.allow_cancellation_scope_authoring,
             self.allow_cancellation_scope_delivery,
         )?;
+        workflow_state.observation_namespace = Some(self.client.namespace.clone());
         workflow_state.history_budget = history_budget;
         workflow_state.workflow_command_identity = workflow_command_identity;
         if let Some(canonical) = cancellation_history {
@@ -12366,6 +12369,11 @@ fn is_authored_command_open_event(event: &HistoryEvent) -> bool {
 
 #[derive(Debug)]
 struct WorkflowState {
+    observation_namespace: Option<String>,
+    observation_terminal: bool,
+    // A pending recorded call consumes its descriptor before yielding. Cursor
+    // exhaustion alone therefore cannot make the surrounding poll live.
+    observation_pending: bool,
     workflow_id: Option<String>,
     run_id: Option<String>,
     task_queue: String,
@@ -12663,6 +12671,19 @@ impl WorkflowState {
                 })
             });
         Ok(Self {
+            observation_namespace: None,
+            observation_terminal: history.iter().any(|event| {
+                matches!(
+                    event.event_type.as_str(),
+                    "WorkflowCompleted"
+                        | "WorkflowFailed"
+                        | "WorkflowCancelled"
+                        | "WorkflowTerminated"
+                        | "WorkflowTimedOut"
+                        | "WorkflowContinuedAsNew"
+                )
+            }),
+            observation_pending: false,
             workflow_command_identity: String::new(),
             workflow_stream_command_counter: 0,
             workflow_id,
@@ -14734,6 +14755,7 @@ impl Future for SelectCall {
                 // on, so an otherwise commandless replay remains legitimately
                 // pending instead of failing as an untracked yield.
                 state.matched_recorded_pending = true;
+                state.observation_pending = true;
             }
             marker
         };
@@ -15812,6 +15834,7 @@ impl ActivityCall {
                         return Poll::Ready(outcome.map_err(Error::ActivityFailed));
                     }
                     state.matched_recorded_pending = true;
+                    state.observation_pending = true;
                     self.scheduled = true;
                     return Poll::Pending;
                 }
@@ -16010,6 +16033,7 @@ impl Future for TimerCall {
                         return Poll::Ready(Ok(()));
                     }
                     state.matched_recorded_pending = true;
+                    state.observation_pending = true;
                     self.scheduled = true;
                     self.matched_pending = true;
                     return Poll::Pending;
@@ -16592,6 +16616,7 @@ impl ChildWorkflowCall {
                         return Poll::Ready(outcome.map_err(Error::ChildWorkflowFailed));
                     }
                     state.matched_recorded_pending = true;
+                    state.observation_pending = true;
                     self.scheduled = true;
                     self.matched_pending = true;
                     return Poll::Pending;
@@ -16840,6 +16865,7 @@ impl SignalCall {
                     }
 
                     state.matched_recorded_pending = true;
+                    state.observation_pending = true;
                     self.opened_wait = true;
                     self.matched_pending = true;
                     return Poll::Pending;

src/nexus.rs

@@ -518,6 +518,7 @@ impl Future for NexusServiceCall {
                 );
             }
             state.matched_recorded_pending = true;
+            state.observation_pending = true;
             self.suspended = true;
             return Poll::Pending;
         }

src/replay_clock.rs

@@ -54,6 +54,11 @@ impl Drop for Guard {
     }
 }
 
+pub(super) fn observation_mode(context: &WorkflowContext) -> Option<bool> {
+    let active = ACTIVE.with(|active| active.borrow().as_ref().and_then(Weak::upgrade))?;
+    Arc::ptr_eq(&active, &context.state).then(|| QUERY_REPLAY.with(Cell::get))
+}
+
 fn unavailable() -> Error {
     Error::ReplayClockUnavailable
 }
@@ -445,11 +450,15 @@ impl WorkflowState {
     }
 
     pub(super) fn consume_unproven_boundary(&mut self, path: &[ParallelGroupMetadata]) {
+        if path.is_empty() {
+            self.observation_pending = false;
+        }
         self.replay_clock.unavailable();
         self.unavailable_cancellation_boundary(path);
     }
 
     pub(super) fn consume_replay_group(&mut self, sequences: &[u64], failed: Option<u64>) {
+        self.observation_pending = false;
         // Core's query group-send branch supplies no aggregate time. Do not
         // transfer executor authority into reconstructed query execution.
         self.replay_clock.time = if QUERY_REPLAY.with(Cell::get) {
@@ -477,16 +486,19 @@ impl WorkflowState {
     }
 
     pub(super) fn consume_replay_selection(&mut self, group: &str) {
+        self.observation_pending = false;
         self.replay_clock.unavailable();
         self.advance_cancellation_selection(group);
     }
 
     pub(super) fn consume_replay_receipt(&mut self, group: &str, sequence: u64) {
+        self.observation_pending = false;
         self.replay_clock.unavailable();
         self.advance_cancellation_receipt(group, sequence);
     }
 
     pub(super) fn consume_replay_handle(&mut self, handle: &DurableOperationHandle, failed: bool) {
+        self.observation_pending = false;
         self.replay_clock.unavailable();
         self.advance_cancellation_handle(handle, failed);
     }
@@ -501,6 +513,7 @@ impl WorkflowState {
     }
 
     pub(super) fn observe_replay_sequence(&mut self, sequence: u64) {
+        self.observation_pending = false;
         self.replay_clock.time = self
             .replay_clock
             .boundaries

src/testing/environment.rs

@@ -119,6 +119,7 @@ impl WorkflowEnvironment {
     /// Newly emitted durable commands are divergences, including commands added
     /// after the recorded history ended. No activity callbacks are invoked.
     pub fn replay(&self, history: &WorkflowHistory) -> Result<ReplayReport> {
+        let _observations = crate::workflow_logging::ReconstructionGuard::enter();
         self.worker.validate_registration()?;
         let mut worker = self.worker.clone();
         if let Some(queue) = &history.task_queue {

src/workflow_logging.rs

@@ -0,0 +1,162 @@
+use crate::{replay_clock, WorkflowContext, WorkflowState};
+use std::{cell::Cell, panic::AssertUnwindSafe};
+
+thread_local! {
+    static RECONSTRUCTING: Cell<bool> = const { Cell::new(false) };
+}
+
+/// The live execution region for workflow observations.
+///
+/// This read-only view does not authorize different durable workflow commands.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum WorkflowReplayState {
+    /// Reconstructing committed history or a query. Process logs are suppressed.
+    Replaying,
+    /// New execution beyond the committed prefix. A task retry can repeat logs.
+    Executing,
+    /// Outside this workflow invocation, or its state cannot be read safely.
+    Unavailable,
+}
+
+pub(crate) struct ReconstructionGuard(bool);
+
+impl ReconstructionGuard {
+    pub(crate) fn enter() -> Self {
+        Self(RECONSTRUCTING.with(|active| active.replace(true)))
+    }
+}
+
+impl Drop for ReconstructionGuard {
+    fn drop(&mut self) {
+        RECONSTRUCTING.with(|active| active.set(self.0));
+    }
+}
+
+fn replaying(state: &WorkflowState) -> bool {
+    state.observation_terminal
+        || state.command_cursor < state.recorded_commands.len()
+        || state.selection_marker_cursor < state.selection_markers.len()
+        || (state.observation_pending && state.commands.is_empty())
+        || state.cancellation_delivery_intent.is_some()
+        || state.cancellation_scope_opening.is_some()
+        || (state.cancellation_history.delivery.is_some() && !state.cancellation_consumed)
+        || (state.recorded_continue_as_new_sequence.is_some() && !state.continue_as_new_consumed)
+        || state.scope_delivery.as_ref().is_some_and(|replay| {
+            replay.intent.is_some()
+                || replay
+                    .canonical
+                    .deliveries
+                    .keys()
+                    .any(|sequence| !replay.consumed.contains(sequence))
+        })
+}
+
+impl WorkflowContext {
+    /// Observe the current replay boundary without changing commands or history.
+    ///
+    /// Query/offline reconstruction remains `Replaying` after the last recorded
+    /// command. A retained context outside its invocation is `Unavailable`.
+    pub fn replay_state(&self) -> WorkflowReplayState {
+        let Some(query) = replay_clock::observation_mode(self) else {
+            return WorkflowReplayState::Unavailable;
+        };
+        let Ok(state) = self.state.lock() else {
+            return WorkflowReplayState::Unavailable;
+        };
+        if query || RECONSTRUCTING.with(Cell::get) || replaying(&state) {
+            WorkflowReplayState::Replaying
+        } else {
+            WorkflowReplayState::Executing
+        }
+    }
+
+    /// Log new workflow execution through the application's `tracing` subscriber.
+    ///
+    /// Committed replay, queries and discarded invocations stay quiet. Handlers
+    /// must be bounded/nonblocking. This process logger is not an exactly-once
+    /// outbox: logs from an uncommitted task can repeat when the task retries.
+    pub fn logger(&self) -> WorkflowLogger {
+        WorkflowLogger {
+            context: self.clone(),
+        }
+    }
+}
+
+/// Replay-aware host logging with SDK-bound namespace/workflow/run fields.
+#[derive(Clone)]
+pub struct WorkflowLogger {
+    context: WorkflowContext,
+}
+
+impl std::fmt::Debug for WorkflowLogger {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("WorkflowLogger").finish_non_exhaustive()
+    }
+}
+
+impl WorkflowLogger {
+    /// Emit only in a new execution region with a complete bound run identity.
+    /// Logger failures do not manufacture success, commands or workflow history.
+    pub fn log(&self, level: tracing::Level, message: &str) {
+        if self.context.replay_state() != WorkflowReplayState::Executing {
+            return;
+        }
+        let identity = self.context.state.lock().ok().and_then(|state| {
+            let namespace = state
+                .observation_namespace
+                .as_ref()
+                .filter(|v| !v.is_empty())?;
+            let workflow = state.workflow_id.as_ref().filter(|v| !v.is_empty())?;
+            let run = state.run_id.as_ref().filter(|v| !v.is_empty())?;
+            Some((namespace.clone(), workflow.clone(), run.clone()))
+        });
+        let Some((namespace, workflow, run)) = identity else {
+            return;
+        };
+        // Invoke application code without holding workflow state. The temporary
+        // observation scope also suppresses recursive logs from a subscriber.
+        let _scope = replay_clock::Guard::unavailable();
+        let _ = std::panic::catch_unwind(AssertUnwindSafe(|| {
+            macro_rules! emit {
+                ($level:expr) => {
+                    tracing::event!(target: "durable_workflow::workflow", $level,
+                        namespace = namespace.as_str(),
+                        workflow_instance_id = workflow.as_str(),
+                        workflow_run_id = run.as_str(), message)
+                };
+            }
+            match level {
+                tracing::Level::TRACE => emit!(tracing::Level::TRACE),
+                tracing::Level::DEBUG => emit!(tracing::Level::DEBUG),
+                tracing::Level::INFO => emit!(tracing::Level::INFO),
+                tracing::Level::WARN => emit!(tracing::Level::WARN),
+                tracing::Level::ERROR => emit!(tracing::Level::ERROR),
+            }
+        }));
+    }
+
+    /// Emit a trace observation in new workflow execution.
+    pub fn trace(&self, message: &str) {
+        self.log(tracing::Level::TRACE, message);
+    }
+    /// Emit a debug observation in new workflow execution.
+    pub fn debug(&self, message: &str) {
+        self.log(tracing::Level::DEBUG, message);
+    }
+    /// Emit an informational observation in new workflow execution.
+    pub fn info(&self, message: &str) {
+        self.log(tracing::Level::INFO, message);
+    }
+    /// Emit a warning observation in new workflow execution.
+    pub fn warn(&self, message: &str) {
+        self.log(tracing::Level::WARN, message);
+    }
+    /// Emit an error observation in new workflow execution.
+    pub fn error(&self, message: &str) {
+        self.log(tracing::Level::ERROR, message);
+    }
+}
+
+#[cfg(test)]
+mod tests;