Fix replay of durable calls completed in the crash window

cli/golem-cli/src/model/agent/oplog.rs

@@ -150,6 +150,14 @@ impl TextOutput for PublicOplogEntry {
                 ));
                 logln(format!("{pad}start index:       {}", params.start_index));
             }
+            PublicOplogEntry::CompletionDelivered(params) => {
+                logln(format_message_highlight("COMPLETION DELIVERED"));
+                logln(format!(
+                    "{pad}at:                {}",
+                    format_id(&params.timestamp)
+                ));
+                logln(format!("{pad}start index:       {}", params.start_index));
+            }
             PublicOplogEntry::AgentInvocationStarted(params) => {
                 log_agent_invocation(
                     AgentInvocationRenderKind::Started,

cli/golem-cli/wit/deps/golem-1.x/golem-oplog.wit

@@ -165,6 +165,15 @@ interface oplog {
         start-index: oplog-index,
     }
 
+    /// Parameters of a `completion-delivered` entry: the successful result of the durable host
+    /// call started at `start-index` was handed to the agent at this point in the recorded
+    /// execution. Replay may prepare the recorded host result earlier, but does not hand it to the
+    /// agent until this marker and prevents later oplog entries from advancing until that handoff.
+    record completion-delivered-parameters {
+        timestamp: datetime,
+        start-index: oplog-index,
+    }
+
     record local-span-data {
         span-id: span-id,
         start: datetime,
@@ -620,6 +629,11 @@ interface oplog {
         start-index: oplog-index,
     }
 
+    record raw-completion-delivered-parameters {
+        timestamp: datetime,
+        start-index: oplog-index,
+    }
+
     /// Parameters for a host-stream-frame oplog entry, with the frame payload in raw
     /// (possibly externally stored) form.
     record raw-host-stream-frame-parameters {
@@ -844,7 +858,10 @@ interface oplog {
         /// The successful completion of the durable host call started by the matching `start`
         /// was persisted, but its response was never delivered to the agent (the agent dropped
         /// the completion future after the `end` was recorded)
-        completion-discarded(raw-completion-discarded-parameters)
+        completion-discarded(raw-completion-discarded-parameters),
+        /// The successful completion of the durable host call started by the matching `start`
+        /// was delivered to the agent at this point in the recorded execution
+        completion-delivered(raw-completion-delivered-parameters)
     }
 
     variant public-oplog-entry {
@@ -953,7 +970,10 @@ interface oplog {
         /// The successful completion of the durable host call started by the matching `start`
         /// was persisted, but its response was never delivered to the agent (the agent dropped
         /// the completion future after the `end` was recorded)
-        completion-discarded(completion-discarded-parameters)
+        completion-discarded(completion-discarded-parameters),
+        /// The successful completion of the durable host call started by the matching `start`
+        /// was delivered to the agent at this point in the recorded execution
+        completion-delivered(completion-delivered-parameters)
     }
 
     /// Enriches raw oplog entries into public oplog entries by resolving oplog payloads

golem-api-grpc/proto/golem/worker/public_oplog.proto

@@ -73,6 +73,7 @@ message OplogEntry {
     CardExpiredParameters CardExpired = 51;
     HostStreamFrameParameters HostStreamFrame = 52;
     CompletionDiscardedParameters CompletionDiscarded = 53;
+    CompletionDeliveredParameters CompletionDelivered = 54;
   }
 }
 
@@ -154,6 +155,11 @@ message CompletionDiscardedParameters {
   uint64 start_index = 2;
 }
 
+message CompletionDeliveredParameters {
+  google.protobuf.Timestamp timestamp = 1;
+  uint64 start_index = 2;
+}
+
 message AgentInvocationStartedParameters {
   google.protobuf.Timestamp timestamp = 1;
   PublicAgentInvocation invocation = 2;

golem-api-grpc/proto/golem/worker/raw_oplog.proto

@@ -65,6 +65,7 @@ message RawOplogEntry {
     RawCardExpiredParameters card_expired = 50;
     RawHostStreamFrameParameters host_stream_frame = 51;
     RawCompletionDiscardedParameters completion_discarded = 52;
+    RawCompletionDeliveredParameters completion_delivered = 53;
   }
 }
 
@@ -174,6 +175,10 @@ message RawCompletionDiscardedParameters {
   uint64 start_index = 1;
 }
 
+message RawCompletionDeliveredParameters {
+  uint64 start_index = 1;
+}
+
 message RawAgentInvocationStartedParameters {
   IdempotencyKey idempotency_key = 1;
   RawOplogPayload payload = 2;

golem-common/src/base_model/oplog/mod.rs

@@ -786,4 +786,23 @@ oplog_entry! {
             start_index: OplogIndex,
         }
     },
+    /// Marks the point where the successful completion (`End`) of the durable host call started
+    /// by the `Start` at `start_index` was delivered to the guest. Unlike `End`, this is a guest
+    /// execution boundary: replay may prepare the recorded host result earlier, but must not hand
+    /// it to the guest until this marker, so callbacks run in their recorded order.
+    ///
+    /// The marker is a hint entry and always lies physically after its `End`. Replay indexes it
+    /// before resolving the `End`; only the matching completion may consume it at its physical
+    /// position, and later oplog entries remain blocked until the result crosses to the guest.
+    CompletionDelivered {
+        hint: true
+        wit_raw_type: "raw-completion-delivered-parameters"
+        wit_public_type: "completion-delivered-parameters"
+        raw {
+            start_index: OplogIndex,
+        }
+        public {
+            start_index: OplogIndex,
+        }
+    }
 }

golem-common/src/model/oplog/matcher.rs

@@ -151,6 +151,10 @@ impl PublicOplogEntry {
                 Self::string_match("CompletionDiscarded", &[], query_path, query)
                     || Self::string_match("completion-discarded", &[], query_path, query)
             }
+            PublicOplogEntry::CompletionDelivered(_) => {
+                Self::string_match("CompletionDelivered", &[], query_path, query)
+                    || Self::string_match("completion-delivered", &[], query_path, query)
+            }
             PublicOplogEntry::AgentInvocationStarted(params) => {
                 Self::string_match("agentinvocationstarted", &[], query_path, query)
                     || Self::string_match("invoke", &[], query_path, query)

golem-common/src/model/oplog/mod.rs

@@ -136,6 +136,7 @@ impl OplogEntry {
             // free of concurrent side effects until its semantics have been reviewed.
             OplogEntry::Create { .. }
             | OplogEntry::CompletionDiscarded { .. }
+            | OplogEntry::CompletionDelivered { .. }
             | OplogEntry::AgentInvocationStarted { .. }
             | OplogEntry::Suspend { .. }
             | OplogEntry::Error { .. }
@@ -213,6 +214,7 @@ impl OplogEntry {
             | OplogEntry::End { .. }
             | OplogEntry::Cancelled { .. }
             | OplogEntry::CompletionDiscarded { .. }
+            | OplogEntry::CompletionDelivered { .. }
             | OplogEntry::AgentInvocationStarted { .. }
             | OplogEntry::AgentInvocationFinished { .. }
             | OplogEntry::Suspend { .. }
@@ -313,7 +315,8 @@ impl OplogScopeProjection {
             }
             OplogEntry::End { start_index, .. }
             | OplogEntry::Cancelled { start_index, .. }
-            | OplogEntry::CompletionDiscarded { start_index, .. } => {
+            | OplogEntry::CompletionDiscarded { start_index, .. }
+            | OplogEntry::CompletionDelivered { start_index, .. } => {
                 self.starts.contains(start_index)
             }
             OplogEntry::HostStreamFrame {

golem-common/src/model/oplog/protobuf.rs

@@ -44,15 +44,16 @@ use crate::model::oplog::public_oplog_entry::{
     ActivatePluginParams, AgentInvocationFinishedParams, AgentInvocationStartedParams,
     BeginAtomicRegionParams, BeginRemoteTransactionParams, CancelPendingInvocationParams,
     CancelledParams, CardEventQueuedParams, CardInstallFailedParams, CardInstalledParams,
-    CardRevokedParams, CommittedRemoteTransactionParams, CompletionDiscardedParams, CreateParams,
-    CreateResourceParams, DeactivatePluginParams, DropResourceParams, EndAtomicRegionParams,
-    EndParams, ErrorParams, ExitedParams, FailedUpdateParams, FilesystemStorageUsageUpdateParams,
-    FinishSpanParams, GrowMemoryParams, HostStreamFrameParams, InterruptedParams, JumpParams,
-    LogParams, NoOpParams, OplogProcessorCheckpointParams, PendingAgentInvocationParams,
-    PendingUpdateParams, PreCommitRemoteTransactionParams, PreRollbackRemoteTransactionParams,
-    RemoveRetryPolicyParams, RestartParams, RevertParams, RolledBackRemoteTransactionParams,
-    SetRetryPolicyParams, SetSpanAttributeParams, SnapshotParams, StartParams, StartSpanParams,
-    SuccessfulUpdateParams, SuspendParams,
+    CardRevokedParams, CommittedRemoteTransactionParams, CompletionDeliveredParams,
+    CompletionDiscardedParams, CreateParams, CreateResourceParams, DeactivatePluginParams,
+    DropResourceParams, EndAtomicRegionParams, EndParams, ErrorParams, ExitedParams,
+    FailedUpdateParams, FilesystemStorageUsageUpdateParams, FinishSpanParams, GrowMemoryParams,
+    HostStreamFrameParams, InterruptedParams, JumpParams, LogParams, NoOpParams,
+    OplogProcessorCheckpointParams, PendingAgentInvocationParams, PendingUpdateParams,
+    PreCommitRemoteTransactionParams, PreRollbackRemoteTransactionParams, RemoveRetryPolicyParams,
+    RestartParams, RevertParams, RolledBackRemoteTransactionParams, SetRetryPolicyParams,
+    SetSpanAttributeParams, SnapshotParams, StartParams, StartSpanParams, SuccessfulUpdateParams,
+    SuspendParams,
 };
 use crate::model::oplog::{
     AgentTerminatedByQuotaError, DurableFunctionType, EphemeralCannotSuspendError,
@@ -582,6 +583,17 @@ impl TryFrom<golem_api_grpc::proto::golem::worker::OplogEntry> for PublicOplogEn
                     ),
                 }),
             ),
+            oplog_entry::Entry::CompletionDelivered(completion_delivered) => Ok(
+                PublicOplogEntry::CompletionDelivered(CompletionDeliveredParams {
+                    timestamp: completion_delivered
+                        .timestamp
+                        .ok_or("Missing timestamp field")?
+                        .into(),
+                    start_index: crate::base_model::OplogIndex::from_u64(
+                        completion_delivered.start_index,
+                    ),
+                }),
+            ),
             oplog_entry::Entry::AgentInvocationStarted(agent_invocation_started) => Ok(
                 PublicOplogEntry::AgentInvocationStarted(AgentInvocationStartedParams {
                     timestamp: agent_invocation_started
@@ -1093,6 +1105,16 @@ impl TryFrom<PublicOplogEntry> for golem_api_grpc::proto::golem::worker::OplogEn
                     )),
                 }
             }
+            PublicOplogEntry::CompletionDelivered(completion_delivered) => {
+                golem_api_grpc::proto::golem::worker::OplogEntry {
+                    entry: Some(oplog_entry::Entry::CompletionDelivered(
+                        golem_api_grpc::proto::golem::worker::CompletionDeliveredParameters {
+                            timestamp: Some(completion_delivered.timestamp.into()),
+                            start_index: completion_delivered.start_index.as_u64(),
+                        },
+                    )),
+                }
+            }
             PublicOplogEntry::AgentInvocationStarted(agent_invocation_started) => {
                 golem_api_grpc::proto::golem::worker::OplogEntry {
                     entry: Some(oplog_entry::Entry::AgentInvocationStarted(
@@ -2519,6 +2541,12 @@ impl TryFrom<PublicOplogEntry> for OplogEntry {
                     start_index: completion_discarded.start_index,
                 })
             }
+            PublicOplogEntry::CompletionDelivered(completion_delivered) => {
+                Ok(OplogEntry::CompletionDelivered {
+                    timestamp: completion_delivered.timestamp,
+                    start_index: completion_delivered.start_index,
+                })
+            }
             PublicOplogEntry::AgentInvocationStarted(_) => {
                 Err("Converting AgentInvocationStarted from public to raw oplog entry is not yet supported".to_string())
             }
@@ -3162,10 +3190,10 @@ impl TryFrom<OplogEntry> for golem_api_grpc::proto::golem::worker::RawOplogEntry
             RawAgentInvocationStartedParameters, RawBeginRemoteTransactionParameters,
             RawCancelPendingInvocationParameters, RawCancelledParameters,
             RawCardEventQueuedParameters, RawCardInstallFailedParameters,
-            RawCardInstalledParameters, RawCardRevokedParameters, RawCompletionDiscardedParameters,
-            RawCreateParameters, RawCreateResourceParameters, RawDeactivatePluginParameters,
-            RawDropResourceParameters, RawEndAtomicRegionParameters, RawEndParameters, RawEnvVar,
-            RawErrorParameters, RawFailedUpdateParameters,
+            RawCardInstalledParameters, RawCardRevokedParameters, RawCompletionDeliveredParameters,
+            RawCompletionDiscardedParameters, RawCreateParameters, RawCreateResourceParameters,
+            RawDeactivatePluginParameters, RawDropResourceParameters, RawEndAtomicRegionParameters,
+            RawEndParameters, RawEnvVar, RawErrorParameters, RawFailedUpdateParameters,
             RawFilesystemStorageUsageUpdateParameters, RawFinishSpanParameters,
             RawGrowMemoryParameters, RawHostStreamFrameParameters, RawJumpParameters,
             RawLogParameters, RawOplogProcessorCheckpointParameters, RawOplogRegion,
@@ -3259,6 +3287,11 @@ impl TryFrom<OplogEntry> for golem_api_grpc::proto::golem::worker::RawOplogEntry
                     start_index: start_index.as_u64(),
                 })
             }
+            OplogEntry::CompletionDelivered { start_index, .. } => {
+                Entry::CompletionDelivered(RawCompletionDeliveredParameters {
+                    start_index: start_index.as_u64(),
+                })
+            }
             OplogEntry::AgentInvocationStarted {
                 idempotency_key,
                 payload,
@@ -3704,6 +3737,10 @@ impl TryFrom<golem_api_grpc::proto::golem::worker::RawOplogEntry> for OplogEntry
                 timestamp,
                 start_index: crate::base_model::OplogIndex::from_u64(p.start_index),
             }),
+            Entry::CompletionDelivered(p) => Ok(OplogEntry::CompletionDelivered {
+                timestamp,
+                start_index: crate::base_model::OplogIndex::from_u64(p.start_index),
+            }),
             Entry::AgentInvocationStarted(p) => {
                 let idempotency_key = p.idempotency_key.ok_or("Missing idempotency_key")?.into();
                 let payload = oplog_payload_from_proto(p.payload.ok_or("Missing payload")?)?;

golem-common/wit/deps/golem-1.x/golem-oplog.wit

@@ -165,6 +165,15 @@ interface oplog {
         start-index: oplog-index,
     }
 
+    /// Parameters of a `completion-delivered` entry: the successful result of the durable host
+    /// call started at `start-index` was handed to the agent at this point in the recorded
+    /// execution. Replay may prepare the recorded host result earlier, but does not hand it to the
+    /// agent until this marker and prevents later oplog entries from advancing until that handoff.
+    record completion-delivered-parameters {
+        timestamp: datetime,
+        start-index: oplog-index,
+    }
+
     record local-span-data {
         span-id: span-id,
         start: datetime,
@@ -620,6 +629,11 @@ interface oplog {
         start-index: oplog-index,
     }
 
+    record raw-completion-delivered-parameters {
+        timestamp: datetime,
+        start-index: oplog-index,
+    }
+
     /// Parameters for a host-stream-frame oplog entry, with the frame payload in raw
     /// (possibly externally stored) form.
     record raw-host-stream-frame-parameters {
@@ -844,7 +858,10 @@ interface oplog {
         /// The successful completion of the durable host call started by the matching `start`
         /// was persisted, but its response was never delivered to the agent (the agent dropped
         /// the completion future after the `end` was recorded)
-        completion-discarded(raw-completion-discarded-parameters)
+        completion-discarded(raw-completion-discarded-parameters),
+        /// The successful completion of the durable host call started by the matching `start`
+        /// was delivered to the agent at this point in the recorded execution
+        completion-delivered(raw-completion-delivered-parameters)
     }
 
     variant public-oplog-entry {
@@ -953,7 +970,10 @@ interface oplog {
         /// The successful completion of the durable host call started by the matching `start`
         /// was persisted, but its response was never delivered to the agent (the agent dropped
         /// the completion future after the `end` was recorded)
-        completion-discarded(completion-discarded-parameters)
+        completion-discarded(completion-discarded-parameters),
+        /// The successful completion of the durable host call started by the matching `start`
+        /// was delivered to the agent at this point in the recorded execution
+        completion-delivered(completion-delivered-parameters)
     }
 
     /// Enriches raw oplog entries into public oplog entries by resolving oplog payloads

golem-debugging-service/src/debug_session.rs

@@ -194,6 +194,9 @@ enum PairingSignature {
     CompletionDiscarded {
         start_index: OplogIndex,
     },
+    CompletionDelivered {
+        start_index: OplogIndex,
+    },
     HostStreamFrame {
         parent_start_index: OplogIndex,
         kind: HostStreamKind,
@@ -258,6 +261,11 @@ impl PairingSignature {
                     start_index: *start_index,
                 }
             }
+            OplogEntry::CompletionDelivered { start_index, .. } => {
+                PairingSignature::CompletionDelivered {
+                    start_index: *start_index,
+                }
+            }
             OplogEntry::HostStreamFrame {
                 parent_start_index,
                 kind,

golem-debugging-service/src/oplog/debug_oplog.rs

@@ -16,7 +16,7 @@ use crate::debug_session::{DebugSessionId, DebugSessions};
 use async_trait::async_trait;
 use golem_common::model::oplog::{OplogEntry, OplogIndex, PayloadId, RawOplogPayload};
 use golem_worker_executor::services::oplog::{
-    CommitLevel, Oplog, OrderedOplogStart, PendingUpload,
+    CommitLevel, Oplog, OplogAddReceipt, OrderedOplogStart, PendingUpload,
 };
 use std::collections::{BTreeMap, HashMap};
 use std::fmt::Debug;
@@ -82,6 +82,10 @@ impl Oplog for DebugOplog {
         OplogIndex::NONE
     }
 
+    fn enqueue_add(&self, _entry: OplogEntry) -> OplogAddReceipt {
+        Box::pin(async { OplogIndex::NONE })
+    }
+
     // Mirrors `add`: a debugging session never writes to the oplog, so both entries are built (to
     // satisfy the closure contract) and discarded.
     async fn add_pair(