From 35517502eea28f969f628ab279bfd4af31b26440 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Tue, 30 Jun 2026 13:41:56 +0000 Subject: [PATCH 1/2] feat(slack): show native processing status --- docs/integrations/slack-socket-mode.md | 5 +- .../SlackChannelHealthContractTests.cs | 1 + .../Contracts/SlackGatewayContractTests.cs | 1 + .../SlackSessionBindingContractTests.cs | 220 +++++++++++++++- .../Channels/SlackActorHierarchyTests.cs | 1 + .../Channels/SlackAttachmentIngressTests.cs | 7 + .../Channels/SlackFileFlowIntegrationTests.cs | 24 ++ .../Channels/SlackProactiveThreadTests.cs | 1 + .../Channels/SlackReplyClientTests.cs | 52 ++++ .../SlackThreadBackfillIntegrationTests.cs | 15 ++ .../TestHelpers/FakeSlackApiClient.cs | 7 +- .../Channels/TestHelpers/NoopReplyClient.cs | 7 + .../TestHelpers/RecordingSlackReplyClient.cs | 22 ++ .../TestHelpers/TestChannelRegistries.cs | 39 +++ .../Channels/TestSlackGatewayDeps.cs | 5 + .../ISlackReplyClient.cs | 6 + src/Netclaw.Channels.Slack/SlackChannel.cs | 4 + .../SlackGatewayActor.cs | 1 + .../SlackProcessingOutputRenderer.cs | 38 +++ .../SlackReplyClient.cs | 24 ++ .../SlackThreadBindingActor.cs | 238 +++++++++++++++++- .../ChannelRegistryRegistrationTests.cs | 100 +++++++- ...hannelIntegrationRegistrationExtensions.cs | 6 +- 23 files changed, 814 insertions(+), 10 deletions(-) create mode 100644 src/Netclaw.Channels.Slack/SlackProcessingOutputRenderer.cs diff --git a/docs/integrations/slack-socket-mode.md b/docs/integrations/slack-socket-mode.md index 3b982f6e9..67bded56c 100644 --- a/docs/integrations/slack-socket-mode.md +++ b/docs/integrations/slack-socket-mode.md @@ -9,6 +9,7 @@ The Slack channel runs inside `netclawd` using Slack Socket Mode. - Inbound events: `app_mention`, `message` - Session identity: `{channelId}/{threadTs}` - Reply behavior: assistant replies are posted back to the same thread +- Processing indicator: native Slack thread status via `assistant.threads.setStatus` - Thread behavior: mention starts thread, thread replies continue without mention No public webhook endpoint is required. @@ -23,7 +24,7 @@ Create a Slack app with: - Bot scopes at minimum: - `app_mentions:read` - `channels:history` - - `chat:write` + - `chat:write` for thread replies and native loading status - `groups:history` (if private channels are used) - `channels:read` and `groups:read` (if resolving by channel name) @@ -72,6 +73,8 @@ Supported Slack settings: - When Slack is disabled, daemon starts normally and Slack channel stays inactive. - If Slack is enabled but required tokens are missing, daemon startup fails fast. +- While a session is processing, Netclaw sets the Slack thread status to + `is thinking...` and clears it when the session returns idle. - Socket Mode disconnects are handled by SlackNet reconnecting client behavior. - Slack lifecycle is hosted-service owned rather than actor-owned. Ingress only forwards after the Slack gateway actor exists, while clean reconnect decisions diff --git a/src/Netclaw.Actors.Tests/Channels/Contracts/SlackChannelHealthContractTests.cs b/src/Netclaw.Actors.Tests/Channels/Contracts/SlackChannelHealthContractTests.cs index f5019a992..daef564fa 100644 --- a/src/Netclaw.Actors.Tests/Channels/Contracts/SlackChannelHealthContractTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/Contracts/SlackChannelHealthContractTests.cs @@ -37,6 +37,7 @@ protected override IChannel CreateChannel(bool enabled) new FakeSlackApiClient(auth: new StubAuthApi()), new FakeSlackSocketModeClient(), new RecordingSlackReplyClient(), + TestSlackGatewayDeps.DefaultChannelRegistry, new SessionIngressGate(), new NullContentScanner(), SafePromptInjectionDetector.Instance, diff --git a/src/Netclaw.Actors.Tests/Channels/Contracts/SlackGatewayContractTests.cs b/src/Netclaw.Actors.Tests/Channels/Contracts/SlackGatewayContractTests.cs index 2b3d06776..36607c5b1 100644 --- a/src/Netclaw.Actors.Tests/Channels/Contracts/SlackGatewayContractTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/Contracts/SlackGatewayContractTests.cs @@ -47,6 +47,7 @@ protected override IActorRef CreateGateway(ChannelOptionsBuilder options) Options: slackOptions, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: defaultChannelId, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: new RecordingSlackReplyClient(), ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, diff --git a/src/Netclaw.Actors.Tests/Channels/Contracts/SlackSessionBindingContractTests.cs b/src/Netclaw.Actors.Tests/Channels/Contracts/SlackSessionBindingContractTests.cs index 3a2048274..1728a3611 100644 --- a/src/Netclaw.Actors.Tests/Channels/Contracts/SlackSessionBindingContractTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/Contracts/SlackSessionBindingContractTests.cs @@ -12,6 +12,7 @@ using Netclaw.Actors.Channels; using Netclaw.Actors.Protocol; using Netclaw.Actors.Tests.Channels.TestHelpers; +using Netclaw.Channels; using Netclaw.Channels.Slack; using Netclaw.Configuration; using Netclaw.Security; @@ -143,7 +144,8 @@ private IActorRef CreateActorCore( ISessionPipeline pipeline, ConfigurablePromptInjectionDetector detector, string nameSuffix = "", - IThreadHistoryFetcher? historyFetcher = null) + IThreadHistoryFetcher? historyFetcher = null, + IChannelRegistry? channelRegistry = null) { var paths = TestSlackGatewayDeps.NewTestPaths(); var deps = new SlackGatewayDependencies( @@ -159,6 +161,7 @@ private IActorRef CreateActorCore( }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: channelRegistry ?? TestChannelRegistries.SlackWithProcessingRenderer(_replyClient), ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: historyFetcher ?? EmptyThreadHistoryFetcher.Instance, @@ -176,6 +179,173 @@ private IActorRef CreateActorCore( deps), name); } + [Fact] + public async Task Subscribes_to_processing_state_outputs() + { + var ct = TestContext.Current.CancellationToken; + var detector = new ConfigurablePromptInjectionDetector(PromptInjectionResult.Safe()); + var sid = new SessionId("session-slack-processing-filter"); + var pipeline = new RecordingSessionPipeline(_ => []); + + CreateActorCore(sid, pipeline, detector); + + var options = await pipeline.Created.WaitAsync(ct); + Assert.Equal(OutputFilter.ProcessingState, options.Filter & OutputFilter.ProcessingState); + } + + [Fact] + public async Task Processing_state_output_sets_and_clears_thread_status() + { + var ct = TestContext.Current.CancellationToken; + var detector = new ConfigurablePromptInjectionDetector(PromptInjectionResult.Safe()); + var sid = new SessionId("session-slack-processing-status"); + 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); + + await AwaitAssertAsync(() => + { + Assert.Collection( + _replyClient.Statuses, + status => + { + Assert.Equal("C-test", status.ChannelId.Value); + Assert.Equal("1000.1", status.ThreadTs.Value); + Assert.Equal("is thinking...", status.Status); + }, + status => + { + Assert.Equal("C-test", status.ChannelId.Value); + Assert.Equal("1000.1", status.ThreadTs.Value); + Assert.Equal(string.Empty, status.Status); + }); + }, cancellationToken: ct); + } + + [Fact] + public async Task Processing_state_output_does_not_block_text_when_renderer_stalls() + { + var ct = TestContext.Current.CancellationToken; + var detector = new ConfigurablePromptInjectionDetector(PromptInjectionResult.Safe()); + var sid = new SessionId("session-slack-processing-timeout"); + var renderer = new BlockingProcessingRenderer(); + var registry = TestChannelRegistries.SlackWithProcessingRenderer(renderer); + var pipeline = new RecordingSessionPipeline(_ => + [ + new ProcessingStateOutput(true) { SessionId = sid }, + new TextOutput("visible after status failure") { SessionId = sid }, + new TurnCompleted { SessionId = sid, TurnNumber = new TurnNumber(1) } + ]); + + CreateActorCore(sid, pipeline, detector, channelRegistry: registry); + + await AwaitAssertAsync(() => + { + Assert.Contains(_replyClient.Posts, p => p.Text == "visible after status failure"); + }, cancellationToken: ct); + } + + [Fact] + public async Task Late_processing_start_is_cleared_when_idle_is_latest_state() + { + var ct = TestContext.Current.CancellationToken; + var detector = new ConfigurablePromptInjectionDetector(PromptInjectionResult.Safe()); + var sid = new SessionId("session-slack-processing-late-start"); + var renderer = new ReleasableStartProcessingRenderer(); + 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 AwaitAssertAsync(() => + { + Assert.Equal(new[] { false }, renderer.Statuses); + }, cancellationToken: ct); + + renderer.ReleaseStart(); + + await AwaitAssertAsync(() => + { + Assert.Equal(new[] { false, true, false }, renderer.Statuses); + }, cancellationToken: ct); + } + + [Fact] + public async Task Active_processing_status_is_cleared_when_actor_stops() + { + var ct = TestContext.Current.CancellationToken; + var detector = new ConfigurablePromptInjectionDetector(PromptInjectionResult.Safe()); + var sid = new SessionId("session-slack-processing-stop-clear"); + var pipeline = new RecordingSessionPipeline(_ => + [ + new ProcessingStateOutput(true) { SessionId = sid } + ]); + var actor = CreateActorCore(sid, pipeline, detector); + + await AwaitAssertAsync(() => + { + var status = Assert.Single(_replyClient.Statuses); + Assert.Equal("is thinking...", status.Status); + }, cancellationToken: ct); + + var stopProbe = CreateTestProbe("slack-processing-stop-clear"); + stopProbe.Watch(actor); + Sys.Stop(actor); + await stopProbe.ExpectTerminatedAsync(actor, cancellationToken: ct); + + await AwaitAssertAsync(() => + { + Assert.Collection( + _replyClient.Statuses, + status => Assert.Equal("is thinking...", status.Status), + status => Assert.Equal(string.Empty, status.Status)); + }, cancellationToken: ct); + } + + [Fact] + public async Task Timed_out_processing_start_is_cleared_after_actor_stops() + { + var ct = TestContext.Current.CancellationToken; + var detector = new ConfigurablePromptInjectionDetector(PromptInjectionResult.Safe()); + var sid = new SessionId("session-slack-processing-stop-late-start"); + var renderer = new ReleasableStartProcessingRenderer(); + var registry = TestChannelRegistries.SlackWithProcessingRenderer(renderer); + var pipeline = new RecordingSessionPipeline(_ => + [ + new ProcessingStateOutput(true) { SessionId = sid } + ]); + var actor = CreateActorCore(sid, pipeline, detector, channelRegistry: registry); + + await renderer.StartBlocked.WaitAsync(ct); + + var stopProbe = CreateTestProbe("slack-processing-stop-late-start"); + stopProbe.Watch(actor); + Sys.Stop(actor); + await stopProbe.ExpectTerminatedAsync(actor, cancellationToken: ct); + + await AwaitAssertAsync(() => + { + Assert.Equal(new[] { false }, renderer.Statuses); + }, cancellationToken: ct); + + renderer.ReleaseStart(); + + await AwaitAssertAsync(() => + { + Assert.Equal(new[] { false, true, false }, renderer.Statuses); + }, cancellationToken: ct); + } + // Regression for #939: when the binding has no in-memory pending approval // (passivation between prompt and click) the Slack button payload still // carries the prompt's message TS. The binding must use it to redraw the @@ -405,4 +575,52 @@ await AwaitAssertAsync( Assert.Empty(_replyClient.Updates); } + + private sealed class BlockingProcessingRenderer : IChannelOutputRenderer + { + private readonly TaskCompletionSource _blocked = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public ChannelDescriptorKey Key => ChannelDescriptorKey.FromChannelType(ChannelType.Slack); + + public ValueTask RenderAsync( + ChannelOutputRenderRequest request, + CancellationToken cancellationToken = default) + { + return new ValueTask(_blocked.Task); + } + } + + private sealed class ReleasableStartProcessingRenderer : IChannelOutputRenderer + { + private readonly object _lock = new(); + private readonly List _statuses = []; + private readonly TaskCompletionSource _startBlocked = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _releaseStart = new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _startCount; + + public ChannelDescriptorKey Key => ChannelDescriptorKey.FromChannelType(ChannelType.Slack); + + public IReadOnlyList Statuses + { + get { lock (_lock) return _statuses.ToList(); } + } + + public Task StartBlocked => _startBlocked.Task; + + public void ReleaseStart() => _releaseStart.TrySetResult(); + + public async ValueTask RenderAsync( + ChannelOutputRenderRequest request, + CancellationToken cancellationToken = default) + { + var processing = Assert.IsType(request.Output); + if (processing.IsProcessing && Interlocked.Increment(ref _startCount) == 1) + { + _startBlocked.TrySetResult(); + await _releaseStart.Task; + } + + lock (_lock) _statuses.Add(processing.IsProcessing); + } + } } diff --git a/src/Netclaw.Actors.Tests/Channels/SlackActorHierarchyTests.cs b/src/Netclaw.Actors.Tests/Channels/SlackActorHierarchyTests.cs index 560172556..7dc424ab3 100644 --- a/src/Netclaw.Actors.Tests/Channels/SlackActorHierarchyTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/SlackActorHierarchyTests.cs @@ -279,6 +279,7 @@ private static SlackGatewayDependencies CreateDependencies( }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: new NoopReplyClient(), ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, diff --git a/src/Netclaw.Actors.Tests/Channels/SlackAttachmentIngressTests.cs b/src/Netclaw.Actors.Tests/Channels/SlackAttachmentIngressTests.cs index b604477ed..4ea0eb6c0 100644 --- a/src/Netclaw.Actors.Tests/Channels/SlackAttachmentIngressTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/SlackAttachmentIngressTests.cs @@ -139,6 +139,7 @@ private IActorRef BuildGateway( }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: scanner ?? new MagicByteContentScanner(new ContentPolicy()), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, @@ -748,6 +749,12 @@ public Task UpdateThreadMessageAsync( IReadOnlyList? blocks = null, CancellationToken cancellationToken = default) => Task.CompletedTask; + public Task SetThreadStatusAsync( + SlackChannelId channelId, + SlackThreadTs threadTs, + string status, + CancellationToken cancellationToken = default) => Task.CompletedTask; + public Task UploadFileToThreadAsync( SlackChannelId channelId, SlackThreadTs threadTs, diff --git a/src/Netclaw.Actors.Tests/Channels/SlackFileFlowIntegrationTests.cs b/src/Netclaw.Actors.Tests/Channels/SlackFileFlowIntegrationTests.cs index 7f862e9f3..625814e30 100644 --- a/src/Netclaw.Actors.Tests/Channels/SlackFileFlowIntegrationTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/SlackFileFlowIntegrationTests.cs @@ -131,6 +131,7 @@ public async Task Inbound_image_file_is_downloaded_and_persisted_to_session_medi }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, @@ -215,6 +216,7 @@ public async Task App_mention_with_file_only_is_downloaded_and_persisted() }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, @@ -288,6 +290,7 @@ public async Task File_share_subtype_with_text_flows_through_full_pipeline() }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, @@ -357,6 +360,7 @@ public async Task High_risk_prompt_injection_message_is_blocked_before_session_e }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, @@ -410,6 +414,7 @@ public async Task Failed_turn_posts_single_error_without_generic_fallback() }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, @@ -464,6 +469,7 @@ public async Task Timed_out_slack_post_does_not_block_later_turns() }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, @@ -540,6 +546,7 @@ public async Task Retryable_slack_content_rejection_is_fed_back_to_session_for_c }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, @@ -615,6 +622,7 @@ public async Task Retryable_slack_file_upload_rejection_is_fed_back_to_session() }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, @@ -674,6 +682,7 @@ public async Task Timeout_during_post_sends_transport_failure_feedback_to_sessio }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, @@ -738,6 +747,7 @@ public async Task Text_approval_reply_routes_tool_interaction_response() }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, @@ -829,6 +839,7 @@ public async Task Approval_request_posts_block_buttons_with_text_fallback() }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, @@ -901,6 +912,7 @@ public async Task Button_approval_reply_routes_tool_interaction_response() }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, @@ -973,6 +985,7 @@ public async Task Button_approval_response_forwards_to_session_when_binding_cold }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, @@ -1040,6 +1053,7 @@ public async Task Generic_exception_during_post_sends_unknown_failure_feedback_t }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, @@ -1102,6 +1116,7 @@ public async Task Content_rejection_msg_too_long_sends_message_too_large_feedbac }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, @@ -1165,6 +1180,7 @@ public async Task Content_rejection_invalid_blocks_sends_content_rejected_feedba }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, @@ -1215,6 +1231,7 @@ public async Task Inbound_image_with_real_scanner_flows_to_llm() }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new MagicByteContentScanner(new ContentPolicy()), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, @@ -1277,6 +1294,7 @@ public async Task Scanner_failure_rejects_attachment_and_does_not_inline() }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new FailingContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, @@ -1445,6 +1463,12 @@ public Task UpdateThreadMessageAsync( return Task.CompletedTask; } + public Task SetThreadStatusAsync( + SlackChannelId channelId, + SlackThreadTs threadTs, + string status, + CancellationToken cancellationToken = default) => Task.CompletedTask; + public Task UploadFileToThreadAsync( SlackChannelId channelId, SlackThreadTs threadTs, string filePath, string? filename = null, CancellationToken cancellationToken = default) diff --git a/src/Netclaw.Actors.Tests/Channels/SlackProactiveThreadTests.cs b/src/Netclaw.Actors.Tests/Channels/SlackProactiveThreadTests.cs index b9edc64d1..57aa800df 100644 --- a/src/Netclaw.Actors.Tests/Channels/SlackProactiveThreadTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/SlackProactiveThreadTests.cs @@ -518,6 +518,7 @@ private static SlackGatewayDependencies CreateDependencies( }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: new NoopReplyClient(), ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: EmptyThreadHistoryFetcher.Instance, diff --git a/src/Netclaw.Actors.Tests/Channels/SlackReplyClientTests.cs b/src/Netclaw.Actors.Tests/Channels/SlackReplyClientTests.cs index 5897dc0f3..13c7f4785 100644 --- a/src/Netclaw.Actors.Tests/Channels/SlackReplyClientTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/SlackReplyClientTests.cs @@ -124,6 +124,25 @@ await client.PostThreadReplyAsync(new SlackPostMessage( Assert.Equal("custom approval", section.Text.ToString()); } + [Fact] + public async Task SetThreadStatusAsync_calls_assistant_thread_status_api() + { + var fakeAssistantThreads = new FakeAssistantThreadsApi(); + var fakeClient = new FakeSlackApiClient(assistantThreads: fakeAssistantThreads); + var client = new SlackReplyClient(fakeClient); + + await client.SetThreadStatusAsync( + new SlackChannelId("C123"), + new SlackThreadTs("1234.5678"), + "is thinking...", + TestContext.Current.CancellationToken); + + var status = Assert.Single(fakeAssistantThreads.Statuses); + Assert.Equal("C123", status.ChannelId); + Assert.Equal("1234.5678", status.ThreadTs); + Assert.Equal("is thinking...", status.Status); + } + [Fact] public void Approval_block_builder_uses_unique_action_ids_per_button() { @@ -203,4 +222,37 @@ public Task PostMessage(Message message, CancellationToken public Task StopStream(string channel, string ts, string? markdownText = null, IEnumerable? blocks = null, object? metadataObject = null, MessageMetadata? metadataJson = null, CancellationToken cancellationToken = default) => throw new NotImplementedException(); public Task UpdateStream(string channel, string ts, IEnumerable? newBlocks = null, string? markdownText = null, object? metadataObject = null, MessageMetadata? metadataJson = null, CancellationToken cancellationToken = default) => throw new NotImplementedException(); } + + private sealed class FakeAssistantThreadsApi : IAssistantThreadsApi + { + public List Statuses { get; } = []; + + public Task SetStatus( + string channelId, + string threadTs, + string status, + IEnumerable? loadingMessages = null, + CancellationToken cancellationToken = default) + { + Statuses.Add(new StatusRecord(channelId, threadTs, status)); + return Task.CompletedTask; + } + + public Task SetSuggestedPrompts( + string channelId, + string threadTs, + IEnumerable prompts, + string? title = null, + CancellationToken cancellationToken = default) + => throw new NotImplementedException(); + + public Task SetTitle( + string channelId, + string threadTs, + string title, + CancellationToken cancellationToken = default) + => throw new NotImplementedException(); + + public sealed record StatusRecord(string ChannelId, string ThreadTs, string Status); + } } diff --git a/src/Netclaw.Actors.Tests/Channels/SlackThreadBackfillIntegrationTests.cs b/src/Netclaw.Actors.Tests/Channels/SlackThreadBackfillIntegrationTests.cs index 4c042c554..55f5fec02 100644 --- a/src/Netclaw.Actors.Tests/Channels/SlackThreadBackfillIntegrationTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/SlackThreadBackfillIntegrationTests.cs @@ -131,6 +131,7 @@ public async Task Backfill_messages_are_merged_into_single_user_turn_and_exclude }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), HttpClient: httpClient, @@ -222,6 +223,7 @@ public async Task Backfill_runs_once_per_runtime_and_runs_again_after_restart() }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), HttpClient: httpClient, @@ -358,6 +360,7 @@ public async Task High_risk_backfill_messages_are_dropped_before_turn_assembly() }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), ThreadHistoryFetcher: fetcher, @@ -482,6 +485,7 @@ public async Task Bot_replies_below_thread_root_are_excluded_from_adopted_contex }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), HttpClient: httpClient, @@ -592,6 +596,7 @@ public async Task Bot_authored_thread_root_is_hydrated_for_proactive_post_bootst }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), HttpClient: httpClient, @@ -710,6 +715,7 @@ public async Task Proactively_created_thread_adopts_bot_root_on_first_authorized }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), HttpClient: httpClient, @@ -797,6 +803,7 @@ public async Task Older_out_of_order_live_event_is_dropped_after_cursor_advances }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), HttpClient: httpClient, @@ -928,6 +935,7 @@ public async Task Backfill_document_in_public_channel_is_not_forwarded_as_data_c }, BotUserId: new SlackUserId("UBOT"), DefaultChannelId: null, + ChannelRegistry: TestSlackGatewayDeps.DefaultChannelRegistry, ReplyClient: _replyClient, ContentScanner: new NullContentScanner(), HttpClient: httpClient, @@ -1100,6 +1108,13 @@ public Task UpdateThreadMessageAsync( CancellationToken cancellationToken = default) => Task.CompletedTask; + public Task SetThreadStatusAsync( + SlackChannelId channelId, + SlackThreadTs threadTs, + string status, + CancellationToken cancellationToken = default) + => Task.CompletedTask; + public Task UploadFileToThreadAsync( SlackChannelId channelId, SlackThreadTs threadTs, string filePath, string? filename = null, CancellationToken cancellationToken = default) diff --git a/src/Netclaw.Actors.Tests/Channels/TestHelpers/FakeSlackApiClient.cs b/src/Netclaw.Actors.Tests/Channels/TestHelpers/FakeSlackApiClient.cs index 665d35324..9e6d094d4 100644 --- a/src/Netclaw.Actors.Tests/Channels/TestHelpers/FakeSlackApiClient.cs +++ b/src/Netclaw.Actors.Tests/Channels/TestHelpers/FakeSlackApiClient.cs @@ -13,7 +13,10 @@ namespace Netclaw.Actors.Tests.Channels.TestHelpers; /// test supplies (Chat and/or Auth); every other member throws so unexpected /// API usage fails loud instead of silently succeeding. /// -internal sealed class FakeSlackApiClient(IChatApi? chat = null, IAuthApi? auth = null) : ISlackApiClient +internal sealed class FakeSlackApiClient( + IChatApi? chat = null, + IAuthApi? auth = null, + IAssistantThreadsApi? assistantThreads = null) : ISlackApiClient { public IChatApi Chat => chat ?? throw new NotImplementedException(); public IAuthApi Auth => auth ?? throw new NotImplementedException(); @@ -22,7 +25,7 @@ internal sealed class FakeSlackApiClient(IChatApi? chat = null, IAuthApi? auth = public IAppsConnectionsApi AppsConnectionsApi => throw new NotImplementedException(); public IAppsEventAuthorizationsApi AppsEventAuthorizations => throw new NotImplementedException(); public IAssistantSearchApi AssistantSearch => throw new NotImplementedException(); - public IAssistantThreadsApi AssistantThreads => throw new NotImplementedException(); + public IAssistantThreadsApi AssistantThreads => assistantThreads ?? throw new NotImplementedException(); public IBookmarksApi Bookmarks => throw new NotImplementedException(); public IBotsApi Bots => throw new NotImplementedException(); public ICallParticipantsApi CallParticipants => throw new NotImplementedException(); diff --git a/src/Netclaw.Actors.Tests/Channels/TestHelpers/NoopReplyClient.cs b/src/Netclaw.Actors.Tests/Channels/TestHelpers/NoopReplyClient.cs index 80b663d61..84ee68727 100644 --- a/src/Netclaw.Actors.Tests/Channels/TestHelpers/NoopReplyClient.cs +++ b/src/Netclaw.Actors.Tests/Channels/TestHelpers/NoopReplyClient.cs @@ -23,6 +23,13 @@ public Task UpdateThreadMessageAsync( CancellationToken cancellationToken = default) => Task.CompletedTask; + public Task SetThreadStatusAsync( + SlackChannelId channelId, + SlackThreadTs threadTs, + string status, + CancellationToken cancellationToken = default) + => Task.CompletedTask; + public Task UploadFileToThreadAsync(SlackChannelId channelId, SlackThreadTs threadTs, string filePath, string? filename = null, CancellationToken cancellationToken = default) => Task.CompletedTask; } diff --git a/src/Netclaw.Actors.Tests/Channels/TestHelpers/RecordingSlackReplyClient.cs b/src/Netclaw.Actors.Tests/Channels/TestHelpers/RecordingSlackReplyClient.cs index 531382e7a..aa952a460 100644 --- a/src/Netclaw.Actors.Tests/Channels/TestHelpers/RecordingSlackReplyClient.cs +++ b/src/Netclaw.Actors.Tests/Channels/TestHelpers/RecordingSlackReplyClient.cs @@ -13,6 +13,7 @@ public sealed class RecordingSlackReplyClient : ISlackReplyClient private readonly object _lock = new(); private readonly List _posts = []; private readonly List _updates = []; + private readonly List _statuses = []; public IReadOnlyList Posts { @@ -24,6 +25,11 @@ public IReadOnlyList Updates get { lock (_lock) return _updates.ToList(); } } + public IReadOnlyList Statuses + { + get { lock (_lock) return _statuses.ToList(); } + } + public Exception? ThrowOnPost { get; set; } // Throws on the next post only, then auto-clears. Lets a test fail a content @@ -36,6 +42,7 @@ public void Clear() { _posts.Clear(); _updates.Clear(); + _statuses.Clear(); } } @@ -77,6 +84,16 @@ public Task UpdateThreadMessageAsync( return Task.CompletedTask; } + public Task SetThreadStatusAsync( + SlackChannelId channelId, + SlackThreadTs threadTs, + string status, + CancellationToken cancellationToken = default) + { + lock (_lock) _statuses.Add(new StatusRecord(channelId, threadTs, status)); + return Task.CompletedTask; + } + public Task UploadFileToThreadAsync( SlackChannelId channelId, SlackThreadTs threadTs, @@ -89,4 +106,9 @@ public sealed record UpdateRecord( SlackEventTs MessageTs, string Text, IReadOnlyList? Blocks); + + public sealed record StatusRecord( + SlackChannelId ChannelId, + SlackThreadTs ThreadTs, + string Status); } diff --git a/src/Netclaw.Actors.Tests/Channels/TestHelpers/TestChannelRegistries.cs b/src/Netclaw.Actors.Tests/Channels/TestHelpers/TestChannelRegistries.cs index 7e6023f7c..877195472 100644 --- a/src/Netclaw.Actors.Tests/Channels/TestHelpers/TestChannelRegistries.cs +++ b/src/Netclaw.Actors.Tests/Channels/TestHelpers/TestChannelRegistries.cs @@ -6,6 +6,7 @@ using Netclaw.Actors.Channels; using Netclaw.Channels; using Netclaw.Channels.Discord; +using Netclaw.Channels.Slack; namespace Netclaw.Actors.Tests.Channels.TestHelpers; @@ -45,4 +46,42 @@ [new StaticChannelDescriptorProvider(descriptor)], [], outputRenderers: [new DiscordProcessingOutputRenderer(replyClient)]); } + + public static IChannelRegistry SlackWithProcessingRenderer(ISlackReplyClient replyClient) + => SlackWithProcessingRenderer(new SlackProcessingOutputRenderer(replyClient)); + + public static IChannelRegistry SlackWithProcessingRenderer(IChannelOutputRenderer renderer) + { + var key = ChannelDescriptorKey.FromChannelType(ChannelType.Slack); + var descriptor = new ChannelDescriptor( + key, + ChannelType.Slack, + ChannelKind.RemoteChat, + "Slack", + IsEnabled: true, + ChannelCapabilities.ReceiveMessages + | ChannelCapabilities.SendMessages + | ChannelCapabilities.ThreadedConversations + | ChannelCapabilities.InteractiveApproval, + ToolIntents: new HashSet + { + ChannelToolIntentKind.SendMessage + }, + AddressKinds: new HashSet + { + ChannelAddressKind.Destination, + ChannelAddressKind.Thread + }, + SupportedOutputEffects: new HashSet + { + ChannelOutputEffectKind.TextMessage, + ChannelOutputEffectKind.InteractiveApproval, + ChannelOutputEffectKind.ProcessingIndicator + }); + + return new ChannelRegistry( + [new StaticChannelDescriptorProvider(descriptor)], + [], + outputRenderers: [renderer]); + } } diff --git a/src/Netclaw.Actors.Tests/Channels/TestSlackGatewayDeps.cs b/src/Netclaw.Actors.Tests/Channels/TestSlackGatewayDeps.cs index 9b42ba887..d7112c4f7 100644 --- a/src/Netclaw.Actors.Tests/Channels/TestSlackGatewayDeps.cs +++ b/src/Netclaw.Actors.Tests/Channels/TestSlackGatewayDeps.cs @@ -4,6 +4,8 @@ // // ----------------------------------------------------------------------- using Netclaw.Configuration; +using Netclaw.Actors.Tests.Channels.TestHelpers; +using Netclaw.Channels; namespace Netclaw.Actors.Tests.Channels; @@ -37,6 +39,9 @@ public static ModelCapabilities DefaultTextOnlyModel OutputModalities = ModelModality.Text }; + public static IChannelRegistry DefaultChannelRegistry + => TestChannelRegistries.SlackWithProcessingRenderer(new NoopReplyClient()); + public static NetclawPaths NewTestPaths() { var path = new NetclawPaths(Path.Combine(Path.GetTempPath(), $"netclaw-slack-test-{Guid.NewGuid():N}")); diff --git a/src/Netclaw.Channels.Slack/ISlackReplyClient.cs b/src/Netclaw.Channels.Slack/ISlackReplyClient.cs index 13e7bd622..2d298e442 100644 --- a/src/Netclaw.Channels.Slack/ISlackReplyClient.cs +++ b/src/Netclaw.Channels.Slack/ISlackReplyClient.cs @@ -20,6 +20,12 @@ Task UpdateThreadMessageAsync( IReadOnlyList? blocks = null, CancellationToken cancellationToken = default); + Task SetThreadStatusAsync( + SlackChannelId channelId, + SlackThreadTs threadTs, + string status, + CancellationToken cancellationToken = default); + Task UploadFileToThreadAsync( SlackChannelId channelId, SlackThreadTs threadTs, diff --git a/src/Netclaw.Channels.Slack/SlackChannel.cs b/src/Netclaw.Channels.Slack/SlackChannel.cs index 7e0c878a4..42a0116b2 100644 --- a/src/Netclaw.Channels.Slack/SlackChannel.cs +++ b/src/Netclaw.Channels.Slack/SlackChannel.cs @@ -30,6 +30,7 @@ public sealed class SlackChannel : IChannel, IEventHandler, IEvent private readonly ISlackApiClient _slack; private readonly ISlackSocketModeClient _socketModeClient; private readonly ISlackReplyClient _replyClient; + private readonly IChannelRegistry _channelRegistry; private readonly SessionIngressGate _ingressGate; private readonly IContentScanner _contentScanner; private readonly IPromptInjectionDetector _promptInjectionDetector; @@ -59,6 +60,7 @@ public SlackChannel( ISlackApiClient slack, ISlackSocketModeClient socketModeClient, ISlackReplyClient replyClient, + IChannelRegistry channelRegistry, SessionIngressGate ingressGate, IContentScanner contentScanner, IPromptInjectionDetector? promptInjectionDetector, @@ -77,6 +79,7 @@ public SlackChannel( _slack = slack; _socketModeClient = socketModeClient; _replyClient = replyClient; + _channelRegistry = channelRegistry; _ingressGate = ingressGate; _contentScanner = contentScanner; // Fail loud rather than substituting a no-op detector — a no-op reports @@ -237,6 +240,7 @@ private void CompleteConnectionSetup() Options: _options, BotUserId: _botUserId, DefaultChannelId: _defaultChannelId, + ChannelRegistry: _channelRegistry, ReplyClient: _replyClient, ContentScanner: _contentScanner, ThreadHistoryFetcher: _threadHistoryFetcher, diff --git a/src/Netclaw.Channels.Slack/SlackGatewayActor.cs b/src/Netclaw.Channels.Slack/SlackGatewayActor.cs index b452419f3..69b6251b8 100644 --- a/src/Netclaw.Channels.Slack/SlackGatewayActor.cs +++ b/src/Netclaw.Channels.Slack/SlackGatewayActor.cs @@ -102,6 +102,7 @@ public sealed record SlackGatewayDependencies( SlackChannelOptions Options, SlackUserId? BotUserId, SlackChannelId? DefaultChannelId, + IChannelRegistry ChannelRegistry, ISlackReplyClient ReplyClient, IContentScanner ContentScanner, IThreadHistoryFetcher ThreadHistoryFetcher, diff --git a/src/Netclaw.Channels.Slack/SlackProcessingOutputRenderer.cs b/src/Netclaw.Channels.Slack/SlackProcessingOutputRenderer.cs new file mode 100644 index 000000000..93a6f7b58 --- /dev/null +++ b/src/Netclaw.Channels.Slack/SlackProcessingOutputRenderer.cs @@ -0,0 +1,38 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using Netclaw.Actors.Channels; +using Netclaw.Actors.Protocol; +using Netclaw.Channels; +using static Netclaw.Actors.Sessions.SessionProtocol; + +namespace Netclaw.Channels.Slack; + +public sealed class SlackProcessingOutputRenderer(ISlackReplyClient replyClient) : IChannelOutputRenderer +{ + private const string ThinkingStatus = "is thinking..."; + + public ChannelDescriptorKey Key => ChannelDescriptorKey.FromChannelType(ChannelType.Slack); + + public async ValueTask RenderAsync( + ChannelOutputRenderRequest request, + CancellationToken cancellationToken = default) + { + if (request.EffectKind != ChannelOutputEffectKind.ProcessingIndicator) + throw new InvalidOperationException("Slack processing renderer only supports processing indicator effects."); + + if (request.Output is not ProcessingStateOutput processing) + throw new InvalidOperationException("Slack processing renderer requires a processing state output."); + + if (string.IsNullOrWhiteSpace(request.Target.ThreadOrRootId)) + throw new InvalidOperationException("Slack processing indicators require a thread timestamp."); + + await replyClient.SetThreadStatusAsync( + new SlackChannelId(request.Target.Destination.StableId), + new SlackThreadTs(request.Target.ThreadOrRootId), + processing.IsProcessing ? ThinkingStatus : string.Empty, + cancellationToken); + } +} diff --git a/src/Netclaw.Channels.Slack/SlackReplyClient.cs b/src/Netclaw.Channels.Slack/SlackReplyClient.cs index 0ea3afb1d..b9dc23330 100644 --- a/src/Netclaw.Channels.Slack/SlackReplyClient.cs +++ b/src/Netclaw.Channels.Slack/SlackReplyClient.cs @@ -78,6 +78,30 @@ await slackApiClient.Chat.Update(new MessageUpdate } } + public async Task SetThreadStatusAsync( + SlackChannelId channelId, + SlackThreadTs threadTs, + string status, + CancellationToken cancellationToken = default) + { + try + { + await slackApiClient.AssistantThreads.SetStatus( + channelId.Value, + threadTs.Value, + status, + cancellationToken: cancellationToken); + } + catch (SlackException ex) + { + throw new SlackMessageDeliveryException( + ex.ErrorCode, + MapFailureKind(ex.ErrorCode), + ex.Message, + ex); + } + } + internal static DeliveryFailureKind MapFailureKind(string? errorCode) => errorCode switch { diff --git a/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs b/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs index c69384f07..b219fd1d6 100644 --- a/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs +++ b/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs @@ -61,6 +61,9 @@ internal sealed class SlackThreadBindingActor : ReceivePersistentActor, IWithTim private readonly SessionPipelineHandle _handle; private SlackEventTs? _cursorTs; private SlackEventTs? _pendingCursorTs; + private long _processingIndicatorVersion; + private readonly ProcessingIndicatorState _processingIndicatorState = new(); + private Task _processingIndicatorRenderQueue = 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 @@ -73,6 +76,7 @@ internal sealed class SlackThreadBindingActor : ReceivePersistentActor, IWithTim private static readonly object ReinitializeTimerKey = new(); private static readonly TimeSpan InboundProcessingTimeout = TimeSpan.FromSeconds(30); private static readonly TimeSpan OperationTimeout = TimeSpan.FromSeconds(10); + private static readonly TimeSpan ProcessingIndicatorTimeout = TimeSpan.FromSeconds(1); private const string BackfillDetectorWarning = ":warning: I couldn't safely analyze some earlier thread messages, so they were excluded from context."; private const string LiveDetectorUnavailableWarning = ":warning: I couldn't safely analyze your message — please try again in a moment."; private const string LiveInjectionBlockedWarning = ":warning: Message blocked by prompt-injection policy."; @@ -134,6 +138,7 @@ protected override void PreStart() protected override void PostStop() { + QueueTerminalProcessingIndicatorClear(); _handle.Dispose(); base.PostStop(); } @@ -193,6 +198,7 @@ private void Active() CommandAsync(HandleProactiveThreadAsync); CommandAsync(HandleTrustedReminderAsync); CommandAsync(HandleOutputAsync); + Command(HandleLateProcessingRenderCompleted); Command(msg => { if (msg.Generation != _handle.Generation) @@ -574,7 +580,7 @@ private Task TryIngestSingleAttachmentAsync( private SessionPipelineOptions BuildOptions() => new() { ChannelType = Actors.Channels.ChannelType.Slack, - Filter = OutputFilter.Text | OutputFilter.Files + Filter = OutputFilter.Text | OutputFilter.Files | OutputFilter.ProcessingState }; private async Task EnsureInitializedAsync() @@ -1016,6 +1022,7 @@ private void ApplyPendingApprovalPromptCleared(PendingApprovalPromptCleared clea private async Task ReinitializePipelineAsync(string reason) { + QueueProcessingIndicatorClearIfActive(); _pendingCursorTs = null; // Reset per-turn delivery flags: a reinit aborts the in-flight turn, // and a stale _postedThisTurn=true would otherwise leak into the next @@ -1060,6 +1067,10 @@ private async Task HandleOutputAsync(ThreadOutput threadOutput) _lastFailedPost = uploadResult; break; + case ProcessingStateOutput processing: + await RenderProcessingStateAsync(processing); + break; + // BufferFlush and TextDeltaOutput are not received — Slack subscribes // to OutputFilter.Text (final assembled text), not TextStreaming. @@ -1163,6 +1174,230 @@ private async Task HandleOutputAsync(ThreadOutput threadOutput) } } + private Task RenderProcessingStateAsync(ProcessingStateOutput output) + { + var version = RecordProcessingIndicatorDesiredState(output.IsProcessing); + return QueueProcessingStateRender(output, output.IsRequired, version); + } + + private Task QueueProcessingStateRender( + ProcessingStateOutput output, + bool isRequired, + long version) + { + var requirement = isRequired + ? ChannelOutputRequirement.Required + : ChannelOutputRequirement.Optional; + var request = new ChannelOutputRenderRequest( + BuildOutputRenderTarget(), + output, + ChannelOutputEffectKind.ProcessingIndicator, + requirement); + + var work = new ProcessingRenderWork( + _dependencies.ChannelRegistry, + _log, + request, + isRequired, + output.IsProcessing, + version, + _processingIndicatorState, + Self); + if (isRequired) + return RenderProcessingStateRequestAsync(work); + + _processingIndicatorRenderQueue = _processingIndicatorRenderQueue.ContinueWith( + static async (previous, state) => + { + _ = previous.Exception; + await RenderProcessingStateRequestAsync((ProcessingRenderWork)state!); + }, + work, + CancellationToken.None, + TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default).Unwrap(); + + return Task.CompletedTask; + } + + private void HandleLateProcessingRenderCompleted(LateProcessingRenderCompleted completed) + { + var latest = Volatile.Read(ref _processingIndicatorState.Current); + if (latest.Version <= completed.Version || latest.IsProcessing == completed.IsProcessing) + return; + + _ = QueueProcessingStateRender( + new ProcessingStateOutput(latest.IsProcessing) + { + SessionId = _sessionId + }, + isRequired: false, + latest.Version); + } + + private long RecordProcessingIndicatorDesiredState(bool isProcessing) + { + var version = Interlocked.Increment(ref _processingIndicatorVersion); + Volatile.Write( + ref _processingIndicatorState.Current, + new ProcessingIndicatorSnapshot(version, isProcessing)); + return version; + } + + private void QueueProcessingIndicatorClearIfActive() + { + var current = Volatile.Read(ref _processingIndicatorState.Current); + if (!current.IsProcessing) + return; + + _ = RenderProcessingStateAsync(new ProcessingStateOutput(false) + { + SessionId = _sessionId + }); + } + + private void QueueTerminalProcessingIndicatorClear() + { + var current = Volatile.Read(ref _processingIndicatorState.Current); + if (current.Version == 0) + return; + + Volatile.Write(ref _processingIndicatorState.TerminalCleanupRequested, 1); + var version = RecordProcessingIndicatorDesiredState(isProcessing: false); + _ = QueueProcessingStateRender( + new ProcessingStateOutput(false) + { + SessionId = _sessionId + }, + isRequired: false, + version); + } + + private static async Task RenderProcessingStateRequestAsync(ProcessingRenderWork work) + { + if (ShouldSkipStaleOptionalProcessingRender(work)) + return; + + try + { + await RenderOutputWithTimeoutAsync(work); + } + catch (Exception ex) when (!work.IsRequired) + { + work.Log.Warning(ex, "Failed rendering optional Slack processing indicator"); + } + } + + private static bool ShouldSkipStaleOptionalProcessingRender(ProcessingRenderWork work) + { + if (work.IsRequired) + return false; + + var latest = Volatile.Read(ref work.State.Current); + return latest.Version > work.Version && latest.IsProcessing != work.IsProcessing; + } + + private static async Task RenderOutputWithTimeoutAsync(ProcessingRenderWork work) + { + var renderTask = work.Registry.RenderOutputAsync(work.Request, CancellationToken.None).AsTask(); + try + { + await renderTask.WaitAsync(ProcessingIndicatorTimeout); + } + finally + { + if (!renderTask.IsCompleted) + ObserveLateProcessingRender(renderTask, work); + } + } + + private static void ObserveLateProcessingRender( + Task renderTask, + ProcessingRenderWork work) + { + _ = renderTask.ContinueWith( + static (task, state) => + { + var renderWork = (ProcessingRenderWork)state!; + _ = task.Exception; + + if (Volatile.Read(ref renderWork.State.TerminalCleanupRequested) == 1) + { + if (renderWork.IsProcessing) + _ = ClearTerminalProcessingStateAfterLateStartAsync(renderWork); + return; + } + + renderWork.Owner.Tell(new LateProcessingRenderCompleted(renderWork.Version, renderWork.IsProcessing)); + }, + work, + CancellationToken.None, + TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); + } + + private static async Task ClearTerminalProcessingStateAfterLateStartAsync(ProcessingRenderWork work) + { + var latest = Volatile.Read(ref work.State.Current); + if (latest.IsProcessing) + return; + + var clearWork = work with + { + Request = work.Request with + { + Output = new ProcessingStateOutput(false) + { + SessionId = work.Request.Output.SessionId + }, + Requirement = ChannelOutputRequirement.Optional + }, + IsRequired = false, + IsProcessing = false, + Version = latest.Version + }; + + try + { + await RenderOutputWithTimeoutAsync(clearWork); + } + catch (Exception ex) + { + work.Log.Warning(ex, "Failed clearing terminal Slack processing indicator after late status render"); + } + } + + private sealed class ProcessingIndicatorState + { + public ProcessingIndicatorSnapshot Current = new(0, IsProcessing: false); + public int TerminalCleanupRequested; + } + + private sealed record ProcessingIndicatorSnapshot(long Version, bool IsProcessing); + + private sealed record ProcessingRenderWork( + IChannelRegistry Registry, + ILoggingAdapter Log, + ChannelOutputRenderRequest Request, + bool IsRequired, + bool IsProcessing, + long Version, + ProcessingIndicatorState State, + IActorRef Owner); + + private ChannelDeliveryTarget BuildOutputRenderTarget() + { + var channelKey = ChannelDescriptorKey.FromChannelType(ChannelType.Slack); + return new ChannelDeliveryTarget( + channelKey, + new ResolvedChannelAddress( + channelKey, + ChannelAddressKind.Destination, + _channelId.Value, + _channelId.Value), + _threadTs.Value); + } + private async Task TryHandleTextApprovalResponseAsync(SlackThreadInbound message) { if (_pendingApprovalRequests.Count == 0) @@ -1666,6 +1901,7 @@ await _dependencies.ReplyClient.UploadFileToThreadAsync( private sealed record ThreadOutput(SessionOutput Output) : INoSerializationVerificationNeeded; private sealed record OutputStreamTerminated(int Generation, Exception? Cause) : INoSerializationVerificationNeeded; + private sealed record LateProcessingRenderCompleted(long Version, bool IsProcessing) : INoSerializationVerificationNeeded; private sealed record ReinitializePipeline(string Reason) : INoSerializationVerificationNeeded; private sealed class PendingApprovalRequest { diff --git a/src/Netclaw.Daemon.Tests/Configuration/ChannelRegistryRegistrationTests.cs b/src/Netclaw.Daemon.Tests/Configuration/ChannelRegistryRegistrationTests.cs index 2efeb2a45..4d16d6a6e 100644 --- a/src/Netclaw.Daemon.Tests/Configuration/ChannelRegistryRegistrationTests.cs +++ b/src/Netclaw.Daemon.Tests/Configuration/ChannelRegistryRegistrationTests.cs @@ -81,7 +81,7 @@ public void Registry_enumerates_output_capable_channels_only() Assert.Contains(ChannelOutputEffectKind.FileAttachment, descriptors["mattermost"].SupportedOutputEffects); Assert.Contains(ChannelOutputEffectKind.ProcessingIndicator, descriptors["discord"].SupportedOutputEffects); - Assert.DoesNotContain(ChannelOutputEffectKind.ProcessingIndicator, descriptors["slack"].SupportedOutputEffects); + Assert.Contains(ChannelOutputEffectKind.ProcessingIndicator, descriptors["slack"].SupportedOutputEffects); Assert.DoesNotContain(ChannelOutputEffectKind.ProcessingIndicator, descriptors["mattermost"].SupportedOutputEffects); } @@ -153,6 +153,7 @@ public void Enabled_remote_channels_register_expected_channel_tools() Assert.True(IsRegistered(services)); Assert.True(IsRegistered(services)); Assert.False(typeof(IChannelTool).IsAssignableFrom(typeof(LookupMattermostUserTool))); + Assert.True(IsRegistered(services)); Assert.True(IsRegistered(services)); } @@ -471,6 +472,57 @@ await renderer.RenderAsync( Assert.Equal("channel-1", channelId.Value); } + [Fact] + public async Task Slack_processing_renderer_sets_and_clears_thread_status() + { + var replyClient = new RecordingSlackReplyClient(); + var renderer = new SlackProcessingOutputRenderer(replyClient); + var key = ChannelDescriptorKey.FromChannelType(ChannelType.Slack); + + await renderer.RenderAsync( + BuildProcessingRenderRequest(key), + TestContext.Current.CancellationToken); + await renderer.RenderAsync( + BuildProcessingRenderRequest(key, isProcessing: false), + TestContext.Current.CancellationToken); + + Assert.Collection( + replyClient.Statuses, + status => + { + Assert.Equal("channel-1", status.ChannelId.Value); + Assert.Equal("thread-1", status.ThreadTs.Value); + Assert.Equal("is thinking...", status.Status); + }, + status => + { + Assert.Equal("channel-1", status.ChannelId.Value); + Assert.Equal("thread-1", status.ThreadTs.Value); + Assert.Equal(string.Empty, status.Status); + }); + } + + [Fact] + public async Task Slack_processing_renderer_rejects_non_processing_outputs() + { + var replyClient = new RecordingSlackReplyClient(); + var renderer = new SlackProcessingOutputRenderer(replyClient); + var key = ChannelDescriptorKey.FromChannelType(ChannelType.Slack); + var request = new ChannelOutputRenderRequest( + new ChannelDeliveryTarget( + key, + new ResolvedChannelAddress(key, ChannelAddressKind.Destination, "channel-1", "channel-1"), + "thread-1"), + new TextOutput("hello") + { + SessionId = new SessionId("session-1") + }, + ChannelOutputEffectKind.TextMessage); + + await Assert.ThrowsAsync(async () => + await renderer.RenderAsync(request, TestContext.Current.CancellationToken)); + } + private static IReadOnlyDictionary BuildDescriptors( IReadOnlyDictionary settings) { @@ -509,15 +561,18 @@ private static bool IsRegistered(IServiceCollection services) return services.Any(descriptor => descriptor.ServiceType == typeof(T)); } - private static ChannelOutputRenderRequest BuildProcessingRenderRequest(ChannelDescriptorKey key) + private static ChannelOutputRenderRequest BuildProcessingRenderRequest( + ChannelDescriptorKey key, + bool isProcessing = true) { var target = new ChannelDeliveryTarget( key, - new ResolvedChannelAddress(key, ChannelAddressKind.Destination, "channel-1", "channel-1")); + new ResolvedChannelAddress(key, ChannelAddressKind.Destination, "channel-1", "channel-1"), + "thread-1"); return new ChannelOutputRenderRequest( target, - new ProcessingStateOutput(true) + new ProcessingStateOutput(isProcessing) { SessionId = new SessionId("session-1") }, @@ -646,4 +701,41 @@ public Task TriggerTypingAsync(DiscordReplyChannelId channelId, CancellationToke public Task UploadFileAsync(DiscordFileUpload upload, CancellationToken cancellationToken = default) => Task.FromResult(new DiscordMessageId("file-1")); } + + private sealed class RecordingSlackReplyClient : ISlackReplyClient + { + public List<(SlackChannelId ChannelId, SlackThreadTs ThreadTs, string Status)> Statuses { get; } = []; + + public Task PostThreadReplyAsync(SlackPostMessage message, CancellationToken cancellationToken = default) + => throw new NotSupportedException(); + + public Task PostThreadReplyWithTsAsync(SlackPostMessage message, CancellationToken cancellationToken = default) + => throw new NotSupportedException(); + + public Task UpdateThreadMessageAsync( + SlackChannelId channelId, + SlackEventTs messageTs, + string text, + IReadOnlyList? blocks = null, + CancellationToken cancellationToken = default) + => throw new NotSupportedException(); + + public Task SetThreadStatusAsync( + SlackChannelId channelId, + SlackThreadTs threadTs, + string status, + CancellationToken cancellationToken = default) + { + Statuses.Add((channelId, threadTs, status)); + return Task.CompletedTask; + } + + public Task UploadFileToThreadAsync( + SlackChannelId channelId, + SlackThreadTs threadTs, + string filePath, + string? filename = null, + CancellationToken cancellationToken = default) + => throw new NotSupportedException(); + } } diff --git a/src/Netclaw.Daemon/Configuration/ChannelIntegrationRegistrationExtensions.cs b/src/Netclaw.Daemon/Configuration/ChannelIntegrationRegistrationExtensions.cs index ce89b94dc..1b8f8a530 100644 --- a/src/Netclaw.Daemon/Configuration/ChannelIntegrationRegistrationExtensions.cs +++ b/src/Netclaw.Daemon/Configuration/ChannelIntegrationRegistrationExtensions.cs @@ -40,7 +40,10 @@ public static void AddChannelIntegrations(this IServiceCollection services, ICon internal static void AddSlackChannel(IServiceCollection services, IConfiguration configuration) { - services.AddRemoteChatChannel(ChannelType.Slack, configuration) + services.AddRemoteChatChannel( + ChannelType.Slack, + configuration, + new HashSet { ChannelOutputEffectKind.ProcessingIndicator }) // Token validity is NOT checked here: an exception thrown from this // registration path aborts host construction and crashes the daemon. // A missing/invalid token is handled as a contained channel failure in @@ -68,6 +71,7 @@ internal static void AddSlackChannel(IServiceCollection services, IConfiguration }) .WithOutboundClient() .WithLookupClient() + .WithRenderer() .WithReminderResolver() // The default-channel accessor reads the RUNTIME-resolved ID from // SlackChannel (StartAsync resolves DefaultChannelName → ID), not From 1b378208b52ed36ecc05c6c81631457a166035cf Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Tue, 30 Jun 2026 15:35:46 +0000 Subject: [PATCH 2/2] refactor(slack): simplify processing status rendering --- .../SlackSessionBindingContractTests.cs | 97 --------- .../SlackThreadBindingActor.cs | 186 +++--------------- 2 files changed, 22 insertions(+), 261 deletions(-) diff --git a/src/Netclaw.Actors.Tests/Channels/Contracts/SlackSessionBindingContractTests.cs b/src/Netclaw.Actors.Tests/Channels/Contracts/SlackSessionBindingContractTests.cs index 1728a3611..623c15e17 100644 --- a/src/Netclaw.Actors.Tests/Channels/Contracts/SlackSessionBindingContractTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/Contracts/SlackSessionBindingContractTests.cs @@ -250,36 +250,6 @@ await AwaitAssertAsync(() => }, cancellationToken: ct); } - [Fact] - public async Task Late_processing_start_is_cleared_when_idle_is_latest_state() - { - var ct = TestContext.Current.CancellationToken; - var detector = new ConfigurablePromptInjectionDetector(PromptInjectionResult.Safe()); - var sid = new SessionId("session-slack-processing-late-start"); - var renderer = new ReleasableStartProcessingRenderer(); - 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 AwaitAssertAsync(() => - { - Assert.Equal(new[] { false }, renderer.Statuses); - }, cancellationToken: ct); - - renderer.ReleaseStart(); - - await AwaitAssertAsync(() => - { - Assert.Equal(new[] { false, true, false }, renderer.Statuses); - }, cancellationToken: ct); - } - [Fact] public async Task Active_processing_status_is_cleared_when_actor_stops() { @@ -312,40 +282,6 @@ await AwaitAssertAsync(() => }, cancellationToken: ct); } - [Fact] - public async Task Timed_out_processing_start_is_cleared_after_actor_stops() - { - var ct = TestContext.Current.CancellationToken; - var detector = new ConfigurablePromptInjectionDetector(PromptInjectionResult.Safe()); - var sid = new SessionId("session-slack-processing-stop-late-start"); - var renderer = new ReleasableStartProcessingRenderer(); - var registry = TestChannelRegistries.SlackWithProcessingRenderer(renderer); - var pipeline = new RecordingSessionPipeline(_ => - [ - new ProcessingStateOutput(true) { SessionId = sid } - ]); - var actor = CreateActorCore(sid, pipeline, detector, channelRegistry: registry); - - await renderer.StartBlocked.WaitAsync(ct); - - var stopProbe = CreateTestProbe("slack-processing-stop-late-start"); - stopProbe.Watch(actor); - Sys.Stop(actor); - await stopProbe.ExpectTerminatedAsync(actor, cancellationToken: ct); - - await AwaitAssertAsync(() => - { - Assert.Equal(new[] { false }, renderer.Statuses); - }, cancellationToken: ct); - - renderer.ReleaseStart(); - - await AwaitAssertAsync(() => - { - Assert.Equal(new[] { false, true, false }, renderer.Statuses); - }, cancellationToken: ct); - } - // Regression for #939: when the binding has no in-memory pending approval // (passivation between prompt and click) the Slack button payload still // carries the prompt's message TS. The binding must use it to redraw the @@ -590,37 +526,4 @@ public ValueTask RenderAsync( } } - private sealed class ReleasableStartProcessingRenderer : IChannelOutputRenderer - { - private readonly object _lock = new(); - private readonly List _statuses = []; - private readonly TaskCompletionSource _startBlocked = new(TaskCreationOptions.RunContinuationsAsynchronously); - private readonly TaskCompletionSource _releaseStart = new(TaskCreationOptions.RunContinuationsAsynchronously); - private int _startCount; - - public ChannelDescriptorKey Key => ChannelDescriptorKey.FromChannelType(ChannelType.Slack); - - public IReadOnlyList Statuses - { - get { lock (_lock) return _statuses.ToList(); } - } - - public Task StartBlocked => _startBlocked.Task; - - public void ReleaseStart() => _releaseStart.TrySetResult(); - - public async ValueTask RenderAsync( - ChannelOutputRenderRequest request, - CancellationToken cancellationToken = default) - { - var processing = Assert.IsType(request.Output); - if (processing.IsProcessing && Interlocked.Increment(ref _startCount) == 1) - { - _startBlocked.TrySetResult(); - await _releaseStart.Task; - } - - lock (_lock) _statuses.Add(processing.IsProcessing); - } - } } diff --git a/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs b/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs index b219fd1d6..32f71bf9c 100644 --- a/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs +++ b/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs @@ -61,9 +61,7 @@ internal sealed class SlackThreadBindingActor : ReceivePersistentActor, IWithTim private readonly SessionPipelineHandle _handle; private SlackEventTs? _cursorTs; private SlackEventTs? _pendingCursorTs; - private long _processingIndicatorVersion; - private readonly ProcessingIndicatorState _processingIndicatorState = new(); - private Task _processingIndicatorRenderQueue = Task.CompletedTask; + private volatile bool _processingIndicatorActive; // Set when PerformOneShotHydrationAsync fetched a non-empty thread gap but // found no authorized trigger to anchor a turn. This is the proactive-thread @@ -138,7 +136,7 @@ protected override void PreStart() protected override void PostStop() { - QueueTerminalProcessingIndicatorClear(); + QueueProcessingIndicatorClearIfActive(); _handle.Dispose(); base.PostStop(); } @@ -198,7 +196,6 @@ private void Active() CommandAsync(HandleProactiveThreadAsync); CommandAsync(HandleTrustedReminderAsync); CommandAsync(HandleOutputAsync); - Command(HandleLateProcessingRenderCompleted); Command(msg => { if (msg.Generation != _handle.Generation) @@ -1176,16 +1173,8 @@ private async Task HandleOutputAsync(ThreadOutput threadOutput) private Task RenderProcessingStateAsync(ProcessingStateOutput output) { - var version = RecordProcessingIndicatorDesiredState(output.IsProcessing); - return QueueProcessingStateRender(output, output.IsRequired, version); - } - - private Task QueueProcessingStateRender( - ProcessingStateOutput output, - bool isRequired, - long version) - { - var requirement = isRequired + _processingIndicatorActive = output.IsProcessing; + var requirement = output.IsRequired ? ChannelOutputRequirement.Required : ChannelOutputRequirement.Optional; var request = new ChannelOutputRenderRequest( @@ -1194,60 +1183,16 @@ private Task QueueProcessingStateRender( ChannelOutputEffectKind.ProcessingIndicator, requirement); - var work = new ProcessingRenderWork( - _dependencies.ChannelRegistry, - _log, - request, - isRequired, - output.IsProcessing, - version, - _processingIndicatorState, - Self); - if (isRequired) - return RenderProcessingStateRequestAsync(work); - - _processingIndicatorRenderQueue = _processingIndicatorRenderQueue.ContinueWith( - static async (previous, state) => - { - _ = previous.Exception; - await RenderProcessingStateRequestAsync((ProcessingRenderWork)state!); - }, - work, - CancellationToken.None, - TaskContinuationOptions.ExecuteSynchronously, - TaskScheduler.Default).Unwrap(); + if (output.IsRequired) + return RenderProcessingStateRequestAsync(request, isRequired: true); + _ = RenderProcessingStateRequestAsync(request, isRequired: false); return Task.CompletedTask; } - private void HandleLateProcessingRenderCompleted(LateProcessingRenderCompleted completed) - { - var latest = Volatile.Read(ref _processingIndicatorState.Current); - if (latest.Version <= completed.Version || latest.IsProcessing == completed.IsProcessing) - return; - - _ = QueueProcessingStateRender( - new ProcessingStateOutput(latest.IsProcessing) - { - SessionId = _sessionId - }, - isRequired: false, - latest.Version); - } - - private long RecordProcessingIndicatorDesiredState(bool isProcessing) - { - var version = Interlocked.Increment(ref _processingIndicatorVersion); - Volatile.Write( - ref _processingIndicatorState.Current, - new ProcessingIndicatorSnapshot(version, isProcessing)); - return version; - } - private void QueueProcessingIndicatorClearIfActive() { - var current = Volatile.Read(ref _processingIndicatorState.Current); - if (!current.IsProcessing) + if (!_processingIndicatorActive) return; _ = RenderProcessingStateAsync(new ProcessingStateOutput(false) @@ -1256,50 +1201,26 @@ private void QueueProcessingIndicatorClearIfActive() }); } - private void QueueTerminalProcessingIndicatorClear() + private async Task RenderProcessingStateRequestAsync( + ChannelOutputRenderRequest request, + bool isRequired) { - var current = Volatile.Read(ref _processingIndicatorState.Current); - if (current.Version == 0) - return; - - Volatile.Write(ref _processingIndicatorState.TerminalCleanupRequested, 1); - var version = RecordProcessingIndicatorDesiredState(isProcessing: false); - _ = QueueProcessingStateRender( - new ProcessingStateOutput(false) - { - SessionId = _sessionId - }, - isRequired: false, - version); - } - - private static async Task RenderProcessingStateRequestAsync(ProcessingRenderWork work) - { - if (ShouldSkipStaleOptionalProcessingRender(work)) - return; - try { - await RenderOutputWithTimeoutAsync(work); + await RenderOutputWithTimeoutAsync(_dependencies.ChannelRegistry, request); } - catch (Exception ex) when (!work.IsRequired) + catch (Exception ex) when (!isRequired) { - work.Log.Warning(ex, "Failed rendering optional Slack processing indicator"); + _log.Warning(ex, "Failed rendering optional Slack processing indicator"); } } - private static bool ShouldSkipStaleOptionalProcessingRender(ProcessingRenderWork work) + private static async Task RenderOutputWithTimeoutAsync( + IChannelRegistry registry, + ChannelOutputRenderRequest request) { - if (work.IsRequired) - return false; - - var latest = Volatile.Read(ref work.State.Current); - return latest.Version > work.Version && latest.IsProcessing != work.IsProcessing; - } - - private static async Task RenderOutputWithTimeoutAsync(ProcessingRenderWork work) - { - var renderTask = work.Registry.RenderOutputAsync(work.Request, CancellationToken.None).AsTask(); + using var renderCts = new CancellationTokenSource(ProcessingIndicatorTimeout); + var renderTask = registry.RenderOutputAsync(request, renderCts.Token).AsTask(); try { await renderTask.WaitAsync(ProcessingIndicatorTimeout); @@ -1307,84 +1228,22 @@ private static async Task RenderOutputWithTimeoutAsync(ProcessingRenderWork work finally { if (!renderTask.IsCompleted) - ObserveLateProcessingRender(renderTask, work); + ObserveLateProcessingRender(renderTask); } } - private static void ObserveLateProcessingRender( - Task renderTask, - ProcessingRenderWork work) + private static void ObserveLateProcessingRender(Task renderTask) { _ = renderTask.ContinueWith( - static (task, state) => + static task => { - var renderWork = (ProcessingRenderWork)state!; _ = task.Exception; - - if (Volatile.Read(ref renderWork.State.TerminalCleanupRequested) == 1) - { - if (renderWork.IsProcessing) - _ = ClearTerminalProcessingStateAfterLateStartAsync(renderWork); - return; - } - - renderWork.Owner.Tell(new LateProcessingRenderCompleted(renderWork.Version, renderWork.IsProcessing)); }, - work, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); } - private static async Task ClearTerminalProcessingStateAfterLateStartAsync(ProcessingRenderWork work) - { - var latest = Volatile.Read(ref work.State.Current); - if (latest.IsProcessing) - return; - - var clearWork = work with - { - Request = work.Request with - { - Output = new ProcessingStateOutput(false) - { - SessionId = work.Request.Output.SessionId - }, - Requirement = ChannelOutputRequirement.Optional - }, - IsRequired = false, - IsProcessing = false, - Version = latest.Version - }; - - try - { - await RenderOutputWithTimeoutAsync(clearWork); - } - catch (Exception ex) - { - work.Log.Warning(ex, "Failed clearing terminal Slack processing indicator after late status render"); - } - } - - private sealed class ProcessingIndicatorState - { - public ProcessingIndicatorSnapshot Current = new(0, IsProcessing: false); - public int TerminalCleanupRequested; - } - - private sealed record ProcessingIndicatorSnapshot(long Version, bool IsProcessing); - - private sealed record ProcessingRenderWork( - IChannelRegistry Registry, - ILoggingAdapter Log, - ChannelOutputRenderRequest Request, - bool IsRequired, - bool IsProcessing, - long Version, - ProcessingIndicatorState State, - IActorRef Owner); - private ChannelDeliveryTarget BuildOutputRenderTarget() { var channelKey = ChannelDescriptorKey.FromChannelType(ChannelType.Slack); @@ -1901,7 +1760,6 @@ await _dependencies.ReplyClient.UploadFileToThreadAsync( private sealed record ThreadOutput(SessionOutput Output) : INoSerializationVerificationNeeded; private sealed record OutputStreamTerminated(int Generation, Exception? Cause) : INoSerializationVerificationNeeded; - private sealed record LateProcessingRenderCompleted(long Version, bool IsProcessing) : INoSerializationVerificationNeeded; private sealed record ReinitializePipeline(string Reason) : INoSerializationVerificationNeeded; private sealed class PendingApprovalRequest {