diff --git a/src/Netclaw.Actors.Tests/Channels/Contracts/SlackSessionBindingContractTests.cs b/src/Netclaw.Actors.Tests/Channels/Contracts/SlackSessionBindingContractTests.cs index defc6a63e..7bbcc26ec 100644 --- a/src/Netclaw.Actors.Tests/Channels/Contracts/SlackSessionBindingContractTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/Contracts/SlackSessionBindingContractTests.cs @@ -227,6 +227,61 @@ await AwaitAssertAsync(() => }, cancellationToken: ct); } + [Fact] + public async Task Processing_state_renders_are_serialized_in_output_order() + { + var ct = TestContext.Current.CancellationToken; + var detector = new ConfigurablePromptInjectionDetector(PromptInjectionResult.Safe()); + var sid = new SessionId("session-slack-processing-ordered"); + var renderer = new OrderedProcessingRenderer(); + var registry = TestChannelRegistries.SlackWithProcessingRenderer(renderer); + var pipeline = new RecordingSessionPipeline(_ => + [ + new ProcessingStateOutput(true) { SessionId = sid }, + new ProcessingStateOutput(false) { SessionId = sid }, + new TurnCompleted { SessionId = sid, TurnNumber = new TurnNumber(1) } + ]); + + CreateActorCore(sid, pipeline, detector, channelRegistry: registry); + + await renderer.FirstStarted.WaitAsync(ct); + await AwaitAssertAsync( + () => Assert.NotEmpty(_replyClient.Posts), + cancellationToken: ct); + Assert.False(renderer.SecondStarted.IsCompleted); + + renderer.ReleaseFirst(); + await renderer.SecondStarted.WaitAsync(ct); + + Assert.Collection( + renderer.States, + state => Assert.True(state), + state => Assert.False(state)); + } + + [Fact] + public async Task Turn_completion_does_not_clear_status_while_session_remains_processing() + { + var ct = TestContext.Current.CancellationToken; + var detector = new ConfigurablePromptInjectionDetector(PromptInjectionResult.Safe()); + var sid = new SessionId("session-slack-processing-buffered-turn"); + var pipeline = new RecordingSessionPipeline(_ => + [ + new ProcessingStateOutput(true) { SessionId = sid }, + new TextOutput("First turn completed; continuing with buffered input.") { SessionId = sid }, + new TurnCompleted { SessionId = sid, TurnNumber = new TurnNumber(1) } + ]); + + CreateActorCore(sid, pipeline, detector); + + await AwaitAssertAsync(() => + { + Assert.Contains(_replyClient.Posts, p => p.Text == "First turn completed; continuing with buffered input."); + Assert.NotEmpty(_replyClient.Statuses); + Assert.All(_replyClient.Statuses, status => Assert.Equal("is thinking...", status.Status)); + }, cancellationToken: ct); + } + [Fact] public async Task Inbound_message_refreshes_active_processing_status() { @@ -580,4 +635,45 @@ public ValueTask RenderAsync( } } + private sealed class OrderedProcessingRenderer : IChannelOutputRenderer + { + private readonly object _lock = new(); + private readonly TaskCompletionSource _firstStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _releaseFirst = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _secondStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly List _states = []; + + public ChannelDescriptorKey Key => ChannelDescriptorKey.FromChannelType(ChannelType.Slack); + public Task FirstStarted => _firstStarted.Task; + public Task SecondStarted => _secondStarted.Task; + public IReadOnlyList States + { + get { lock (_lock) return _states.ToList(); } + } + + public void ReleaseFirst() => _releaseFirst.TrySetResult(); + + public ValueTask RenderAsync( + ChannelOutputRenderRequest request, + CancellationToken cancellationToken = default) + { + var state = Assert.IsType(request.Output).IsProcessing; + int invocation; + lock (_lock) + { + _states.Add(state); + invocation = _states.Count; + } + + if (invocation == 1) + { + _firstStarted.TrySetResult(); + return new ValueTask(_releaseFirst.Task); + } + + _secondStarted.TrySetResult(); + return ValueTask.CompletedTask; + } + } + } diff --git a/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs b/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs index 6a3435edb..b58891f38 100644 --- a/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs +++ b/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs @@ -62,6 +62,8 @@ internal sealed class SlackThreadBindingActor : ReceivePersistentActor, IWithTim private SlackEventTs? _cursorTs; private SlackEventTs? _pendingCursorTs; private volatile bool _processingIndicatorActive; + private readonly object _processingIndicatorRenderLock = new(); + private Task _processingIndicatorRenderTail = Task.CompletedTask; // Set when PerformOneShotHydrationAsync fetched a non-empty thread gap but // found no authorized trigger to anchor a turn. This is the proactive-thread @@ -1168,7 +1170,6 @@ private async Task HandleOutputAsync(ThreadOutput threadOutput) cleared => ApplyPendingApprovalPromptCleared(cleared)); } _pendingApprovalRequests.Clear(); - break; } } @@ -1185,11 +1186,39 @@ private Task RenderProcessingStateAsync(ProcessingStateOutput output) ChannelOutputEffectKind.ProcessingIndicator, requirement); - if (output.IsRequired) - return RenderProcessingStateRequestAsync(request, isRequired: true); + var renderTask = QueueProcessingStateRender(request, output.IsRequired); + return output.IsRequired ? renderTask : Task.CompletedTask; + } + + private Task QueueProcessingStateRender(ChannelOutputRenderRequest request, bool isRequired) + { + lock (_processingIndicatorRenderLock) + { + _processingIndicatorRenderTail = RenderAfterPreviousAsync( + _processingIndicatorRenderTail, + request, + isRequired); + return _processingIndicatorRenderTail; + } + } + + private async Task RenderAfterPreviousAsync( + Task previous, + ChannelOutputRenderRequest request, + bool isRequired) + { + try + { + await previous.ConfigureAwait(false); + } + catch (Exception ex) + { + // A failed required render is reported to its caller. It must not + // poison the queue and prevent newer state from reaching Slack. + _log.Warning(ex, "Previous required Slack processing indicator render failed; continuing with newer state"); + } - _ = RenderProcessingStateRequestAsync(request, isRequired: false); - return Task.CompletedTask; + await RenderProcessingStateRequestAsync(request, isRequired).ConfigureAwait(false); } private void QueueProcessingIndicatorClearIfActive()