|
| 1 | +/* |
| 2 | + * Copyright 2026 The Dapr Authors |
| 3 | + * Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | + * you may not use this file except in compliance with the License. |
| 5 | + * You may obtain a copy of the License at |
| 6 | + * http://www.apache.org/licenses/LICENSE-2.0 |
| 7 | + * Unless required by applicable law or agreed to in writing, software |
| 8 | + * distributed under the License is distributed on an "AS IS" BASIS, |
| 9 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 10 | + * See the License for the specific language governing permissions and |
| 11 | + * limitations under the License. |
| 12 | + */ |
| 13 | + |
| 14 | +package io.dapr.durabletask; |
| 15 | + |
| 16 | +import com.google.protobuf.StringValue; |
| 17 | +import com.google.protobuf.Timestamp; |
| 18 | +import io.dapr.durabletask.implementation.protobuf.HistoryEvents; |
| 19 | +import io.dapr.durabletask.implementation.protobuf.Orchestration; |
| 20 | +import io.dapr.durabletask.implementation.protobuf.OrchestratorActions; |
| 21 | +import io.dapr.durabletask.orchestration.TaskOrchestrationFactories; |
| 22 | +import io.dapr.durabletask.orchestration.TaskOrchestrationFactory; |
| 23 | +import org.junit.jupiter.api.Test; |
| 24 | + |
| 25 | +import java.time.Duration; |
| 26 | +import java.time.Instant; |
| 27 | +import java.util.List; |
| 28 | +import java.util.logging.Logger; |
| 29 | + |
| 30 | +import static org.junit.jupiter.api.Assertions.assertTrue; |
| 31 | + |
| 32 | +/** |
| 33 | + * Reproduces an agent-loop stall: the final external event arrives in the |
| 34 | + * same work item as the activity completion that resumes the orchestrator, |
| 35 | + * and must both complete the re-armed wait and not leave its obsolete timer |
| 36 | + * blocking implicit completion. History reconstructed from a failing |
| 37 | + * production instance. |
| 38 | + */ |
| 39 | +class AgentLoopEventDeliveryTest { |
| 40 | + |
| 41 | + private static final Logger logger = Logger.getLogger(AgentLoopEventDeliveryTest.class.getName()); |
| 42 | + private static final Duration MAX_TIMER_INTERVAL = Duration.ofDays(3); |
| 43 | + private static final Instant TEST_INSTANT = Instant.parse("2026-06-12T11:53:03Z"); |
| 44 | + private static final String TEST_INSTANCE = "agent-loop-instance"; |
| 45 | + private static final String EVENT_NAME = "agent-event"; |
| 46 | + |
| 47 | + // ================================================================================== |
| 48 | + // History-event builders |
| 49 | + // ================================================================================== |
| 50 | + |
| 51 | + private static Timestamp ts(Instant instant) { |
| 52 | + return Timestamp.newBuilder() |
| 53 | + .setSeconds(instant.getEpochSecond()) |
| 54 | + .setNanos(instant.getNano()) |
| 55 | + .build(); |
| 56 | + } |
| 57 | + |
| 58 | + private static HistoryEvents.HistoryEvent workflowStarted() { |
| 59 | + return HistoryEvents.HistoryEvent.newBuilder() |
| 60 | + .setEventId(-1) |
| 61 | + .setTimestamp(ts(TEST_INSTANT)) |
| 62 | + .setWorkflowStarted(HistoryEvents.WorkflowStartedEvent.newBuilder().build()) |
| 63 | + .build(); |
| 64 | + } |
| 65 | + |
| 66 | + private static HistoryEvents.HistoryEvent executionStarted(String name) { |
| 67 | + return HistoryEvents.HistoryEvent.newBuilder() |
| 68 | + .setEventId(-1) |
| 69 | + .setTimestamp(ts(TEST_INSTANT)) |
| 70 | + .setExecutionStarted(HistoryEvents.ExecutionStartedEvent.newBuilder() |
| 71 | + .setName(name) |
| 72 | + .setWorkflowInstance( |
| 73 | + Orchestration.WorkflowInstance.newBuilder().setInstanceId(TEST_INSTANCE).build()) |
| 74 | + .build()) |
| 75 | + .build(); |
| 76 | + } |
| 77 | + |
| 78 | + private static HistoryEvents.HistoryEvent eventRaised(String jsonPayload) { |
| 79 | + return HistoryEvents.HistoryEvent.newBuilder() |
| 80 | + .setEventId(-1) |
| 81 | + .setTimestamp(ts(TEST_INSTANT)) |
| 82 | + .setEventRaised(HistoryEvents.EventRaisedEvent.newBuilder() |
| 83 | + .setName(EVENT_NAME) |
| 84 | + .setInput(StringValue.of(jsonPayload)) |
| 85 | + .build()) |
| 86 | + .build(); |
| 87 | + } |
| 88 | + |
| 89 | + private static HistoryEvents.HistoryEvent taskScheduled(int eventId, String name) { |
| 90 | + return HistoryEvents.HistoryEvent.newBuilder() |
| 91 | + .setEventId(eventId) |
| 92 | + .setTimestamp(ts(TEST_INSTANT)) |
| 93 | + .setTaskScheduled(HistoryEvents.TaskScheduledEvent.newBuilder() |
| 94 | + .setName(name) |
| 95 | + .build()) |
| 96 | + .build(); |
| 97 | + } |
| 98 | + |
| 99 | + private static HistoryEvents.HistoryEvent taskCompleted(int taskScheduledId, String jsonResult) { |
| 100 | + return HistoryEvents.HistoryEvent.newBuilder() |
| 101 | + .setEventId(-1) |
| 102 | + .setTimestamp(ts(TEST_INSTANT)) |
| 103 | + .setTaskCompleted(HistoryEvents.TaskCompletedEvent.newBuilder() |
| 104 | + .setTaskScheduledId(taskScheduledId) |
| 105 | + .setResult(StringValue.of(jsonResult)) |
| 106 | + .build()) |
| 107 | + .build(); |
| 108 | + } |
| 109 | + |
| 110 | + // Synthetic "optional" timer emitted by an indefinite waitForExternalEvent: |
| 111 | + // origin ExternalEvent, fireAt = the indefinite-wait sentinel. |
| 112 | + private static HistoryEvents.HistoryEvent sentinelTimerCreated(int eventId) { |
| 113 | + return HistoryEvents.HistoryEvent.newBuilder() |
| 114 | + .setEventId(eventId) |
| 115 | + .setTimestamp(ts(TEST_INSTANT)) |
| 116 | + .setTimerCreated(HistoryEvents.TimerCreatedEvent.newBuilder() |
| 117 | + .setFireAt(ts(TaskOrchestrationExecutor.EXTERNAL_EVENT_INDEFINITE_FIRE_AT)) |
| 118 | + .setExternalEvent( |
| 119 | + HistoryEvents.TimerOriginExternalEvent.newBuilder().setName(EVENT_NAME).build()) |
| 120 | + .build()) |
| 121 | + .build(); |
| 122 | + } |
| 123 | + |
| 124 | + // ================================================================================== |
| 125 | + // The agent-loop orchestrator |
| 126 | + // ================================================================================== |
| 127 | + |
| 128 | + private static TaskOrchestration agentLoop(boolean explicitComplete) { |
| 129 | + return ctx -> { |
| 130 | + while (true) { |
| 131 | + String payload = ctx.waitForExternalEvent(EVENT_NAME, null, String.class).await(); |
| 132 | + if ("done".equals(payload)) { |
| 133 | + if (explicitComplete) { |
| 134 | + ctx.complete("agent-done"); |
| 135 | + } |
| 136 | + return; |
| 137 | + } |
| 138 | + if ("llm".equals(payload)) { |
| 139 | + ctx.callActivity("llm-call", null, null, String.class).await(); |
| 140 | + } else { |
| 141 | + ctx.callActivity("tool-call", null, null, String.class).await(); |
| 142 | + } |
| 143 | + } |
| 144 | + }; |
| 145 | + } |
| 146 | + |
| 147 | + private TaskOrchestrationExecutor createExecutor() { |
| 148 | + return createExecutor(true); |
| 149 | + } |
| 150 | + |
| 151 | + private TaskOrchestrationExecutor createExecutor(boolean explicitComplete) { |
| 152 | + TaskOrchestrationFactories factories = new TaskOrchestrationFactories(); |
| 153 | + factories.addOrchestration(new TaskOrchestrationFactory() { |
| 154 | + @Override |
| 155 | + public String getName() { |
| 156 | + return "AgentLoop"; |
| 157 | + } |
| 158 | + |
| 159 | + @Override |
| 160 | + public TaskOrchestration create() { |
| 161 | + return agentLoop(explicitComplete); |
| 162 | + } |
| 163 | + |
| 164 | + @Override |
| 165 | + public String getVersionName() { |
| 166 | + return null; |
| 167 | + } |
| 168 | + |
| 169 | + @Override |
| 170 | + public Boolean isLatestVersion() { |
| 171 | + return false; |
| 172 | + } |
| 173 | + }); |
| 174 | + return new TaskOrchestrationExecutor(factories, new JacksonDataConverter(), MAX_TIMER_INTERVAL, logger, null); |
| 175 | + } |
| 176 | + |
| 177 | + // Past events reconstructed from the stuck instance (24 events, six turns). |
| 178 | + private static List<HistoryEvents.HistoryEvent> agentLoopHistory() { |
| 179 | + return List.of( |
| 180 | + // Turn 1: started, iter:0 wait armed (event task id=0, sentinel id=1) |
| 181 | + workflowStarted(), executionStarted("AgentLoop"), sentinelTimerCreated(1), |
| 182 | + // Turn 2: event "llm" completes iter:0 wait; llm-call scheduled (id=2) |
| 183 | + workflowStarted(), eventRaised("\"llm\""), taskScheduled(2, "llm-call"), |
| 184 | + // Turn 3: llm completes; iter:1 wait (ids 3,4); event "tool"; tool-call (id=5) |
| 185 | + workflowStarted(), taskCompleted(2, "\"ok\""), eventRaised("\"tool\""), |
| 186 | + sentinelTimerCreated(4), taskScheduled(5, "tool-call"), |
| 187 | + // Turn 4: event buffered before completion; iter:2 wait buffer-hits (id=6, |
| 188 | + // no sentinel); tool-call (id=7) |
| 189 | + workflowStarted(), eventRaised("\"tool\""), taskCompleted(5, "\"ok\""), |
| 190 | + taskScheduled(7, "tool-call"), |
| 191 | + // Turn 5: same shape; iter:3 wait buffer-hits (id=8); tool-call (id=9) |
| 192 | + workflowStarted(), eventRaised("\"tool\""), taskCompleted(7, "\"ok\""), |
| 193 | + taskScheduled(9, "tool-call"), |
| 194 | + // Turn 6: tool completes; iter:4 wait (ids 10,11); event "llm"; llm-call (id=12) |
| 195 | + workflowStarted(), taskCompleted(9, "\"ok\""), eventRaised("\"llm\""), |
| 196 | + sentinelTimerCreated(11), taskScheduled(12, "llm-call")); |
| 197 | + } |
| 198 | + |
| 199 | + private static void assertCompleted(TaskOrchestratorResult result) { |
| 200 | + StringBuilder actions = new StringBuilder(); |
| 201 | + boolean completed = false; |
| 202 | + for (OrchestratorActions.WorkflowAction action : result.getActions()) { |
| 203 | + actions.append(action.getWorkflowActionTypeCase()).append(' '); |
| 204 | + if (action.hasCompleteWorkflow() |
| 205 | + && action.getCompleteWorkflow().getWorkflowStatus() |
| 206 | + == Orchestration.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED) { |
| 207 | + completed = true; |
| 208 | + } |
| 209 | + } |
| 210 | + assertTrue(completed, |
| 211 | + "the 'done' event must complete the workflow, but actions were: [" + actions + "]"); |
| 212 | + } |
| 213 | + |
| 214 | + // Completion precedes the event in the batch; the event must complete the |
| 215 | + // waiter registered in the same execution. |
| 216 | + @Test |
| 217 | + void doneEventDeliveredWhenBatchedAfterActivityCompletion() { |
| 218 | + TaskOrchestratorResult result = createExecutor().execute( |
| 219 | + agentLoopHistory(), |
| 220 | + List.of(workflowStarted(), taskCompleted(12, "\"guide\""), eventRaised("\"done\""))); |
| 221 | + assertCompleted(result); |
| 222 | + } |
| 223 | + |
| 224 | + // Mirrored ordering: the event precedes the completion, gets buffered, and |
| 225 | + // the wait registered afterwards must consume it from the buffer. |
| 226 | + @Test |
| 227 | + void doneEventDeliveredWhenBatchedBeforeActivityCompletion() { |
| 228 | + TaskOrchestratorResult result = createExecutor().execute( |
| 229 | + agentLoopHistory(), |
| 230 | + List.of(workflowStarted(), eventRaised("\"done\""), taskCompleted(12, "\"guide\""))); |
| 231 | + assertCompleted(result); |
| 232 | + } |
| 233 | + |
| 234 | + // The production stall: no explicit ctx.complete — the resolved wait's |
| 235 | + // pending timer must not block implicit completion (before the fix this |
| 236 | + // yielded [CreateTimer] and the workflow stayed RUNNING). |
| 237 | + @Test |
| 238 | + void implicitCompletionNotBlockedByResolvedWaitTimer() { |
| 239 | + TaskOrchestratorResult result = createExecutor(false).execute( |
| 240 | + agentLoopHistory(), |
| 241 | + List.of(workflowStarted(), taskCompleted(12, "\"guide\""), eventRaised("\"done\""))); |
| 242 | + assertCompleted(result); |
| 243 | + } |
| 244 | + |
| 245 | + // Without the event in the batch, re-arming the wait and yielding a single |
| 246 | + // sentinel CreateTimer is the correct response. |
| 247 | + @Test |
| 248 | + void missingEventYieldsSentinelTimerOnly() { |
| 249 | + TaskOrchestratorResult result = createExecutor().execute( |
| 250 | + agentLoopHistory(), |
| 251 | + List.of(workflowStarted(), taskCompleted(12, "\"guide\""))); |
| 252 | + |
| 253 | + int timers = 0; |
| 254 | + for (OrchestratorActions.WorkflowAction action : result.getActions()) { |
| 255 | + assertTrue(action.hasCreateTimer(), |
| 256 | + "expected only CreateTimer actions, got " + action.getWorkflowActionTypeCase()); |
| 257 | + timers++; |
| 258 | + } |
| 259 | + assertTrue(timers == 1, "expected exactly one sentinel CreateTimer, got " + timers); |
| 260 | + } |
| 261 | +} |
0 commit comments