diff --git a/src/Netclaw.Actors.Tests/Reminders/ReminderExecutionActorTests.cs b/src/Netclaw.Actors.Tests/Reminders/ReminderExecutionActorTests.cs index 7d7540b2c..8851f80da 100644 --- a/src/Netclaw.Actors.Tests/Reminders/ReminderExecutionActorTests.cs +++ b/src/Netclaw.Actors.Tests/Reminders/ReminderExecutionActorTests.cs @@ -240,6 +240,44 @@ public async Task Execution_pipeline_requests_streaming_and_tool_call_output() Assert.True(pipeline.CapturedOptions!.Filter.HasFlag(OutputFilter.ToolCalls)); } + [Fact] + public async Task Mode_A_wedged_session_is_failed_by_stall_backstop_releasing_the_guard() + { + // #1492 regression: a Channel-delivery (Mode A) reminder whose session + // wedges — stops producing output without ever emitting TurnCompleted or + // Error — used to hold the duplicate-execution guard forever (the actor + // had no execution ceiling). The stall backstop must conclude it as a + // failure so the parent clears _activeExecutions and the next fire runs. + var originalTimeout = ReminderExecutionActor.ExecutionStallTimeout; + ReminderExecutionActor.ExecutionStallTimeout = TimeSpan.FromMilliseconds(250); + try + { + // Emits one non-terminal output, then goes silent forever — no + // TurnCompleted/Error ever arrives. + var pipeline = new ScriptedSessionPipeline(sessionId => + [ + new TextOutput("Working on it...") { SessionId = sessionId } + ]); + + var definition = CreateDefinition("mode-a-stall"); + var probe = CreateTestProbe(); + Sys.ActorOf( + Props.Create(() => new ParentProxy(probe.Ref, definition, pipeline, _historyStore)), + "exec-mode-a-stall"); + + var completed = await probe.ExpectMsgAsync( + TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); + + Assert.False(completed.Success); + Assert.Equal("mode-a-stall", completed.Id.Value); + Assert.Contains("stalled", completed.ErrorMessage!, StringComparison.OrdinalIgnoreCase); + } + finally + { + ReminderExecutionActor.ExecutionStallTimeout = originalTimeout; + } + } + private static ReminderDefinition CreateDefinition(string id) { var now = TimeProvider.System.GetUtcNow(); diff --git a/src/Netclaw.Actors/Reminders/ReminderExecutionActor.cs b/src/Netclaw.Actors/Reminders/ReminderExecutionActor.cs index ded4e1170..1fe2ddacf 100644 --- a/src/Netclaw.Actors/Reminders/ReminderExecutionActor.cs +++ b/src/Netclaw.Actors/Reminders/ReminderExecutionActor.cs @@ -37,6 +37,20 @@ internal sealed class ReminderExecutionActor : ReceiveActor /// internal static TimeSpan DeliveryObservedTimeout = TimeSpan.FromHours(1); + /// + /// Backstop inactivity ceiling for the Mode A (Channel / own-pipeline) path. + /// The actor runs its own session pipeline and concludes when it emits + /// TurnCompleted/Error; if the session wedges and stops + /// producing output without ever reaching a terminal signal, nothing else + /// stops this actor, so it would hold the duplicate-execution guard forever + /// (see #1492). A reset by every session output + /// fires after this much silence and concludes the run as failed, releasing + /// the guard so the next fire can run. Reset by real output, so it never + /// preempts a live turn. Mode B (CurrentSession) does NOT arm this — its wait + /// is bounded separately by . + /// + internal static TimeSpan ExecutionStallTimeout = TimeSpan.FromMinutes(20); + private readonly Guid _executionId; private readonly ReminderDefinition _definition; private readonly ReminderHistoryStore _historyStore; @@ -100,6 +114,7 @@ public ReminderExecutionActor( Receive(_ => { }); Receive(HandleDeliveryResult); Receive(HandleDeliveryBackstopTimeout); + Receive(_ => HandleExecutionStall()); } protected override void PreStart() @@ -166,6 +181,13 @@ await inputQueue.OfferAsync(new ChannelInput }); inputQueue.Complete(); + + // Arm the Mode A stall backstop: the pipeline now streams output to + // this actor, each ExecutionOutput resets the ReceiveTimeout, and a + // terminal TurnCompleted/Error stops us first. If the session wedges + // and goes silent without a terminal signal, this fires and releases + // the duplicate-execution guard instead of hanging forever (#1492). + Context.SetReceiveTimeout(ExecutionStallTimeout); } catch (Exception ex) { @@ -541,6 +563,18 @@ private void HandleOutput(ExecutionOutput wrapper) } } + private void HandleExecutionStall() + { + if (_completed) + return; + + var elapsed = _timeProvider.GetUtcNow() - _dispatchedAt; + _log.Warning( + "ReminderExecution Stalled: execution_id={0} reminder_id={1} title={2} no session output for {3} (elapsed={4}); concluding as failed to release the execution guard.", + _executionId, _definition.Id, _definition.Title, ExecutionStallTimeout, elapsed); + ReportAndStop(false, $"Reminder execution stalled: no session output for {ExecutionStallTimeout}."); + } + private void ReportAndStop(bool success, string? errorMessage = null) { if (_completed) @@ -548,6 +582,9 @@ private void ReportAndStop(bool success, string? errorMessage = null) _completed = true; + // Disarm the Mode A stall backstop so it cannot fire during the drain. + Context.SetReceiveTimeout(null); + var durationMs = (long)(_timeProvider.GetUtcNow() - _dispatchedAt).TotalMilliseconds; _pendingHistory = new HistoryRecord( FiredAt: _dispatchedAt,