From 28a457e19850870b9a2623fab6f885b4cc5f982c Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Thu, 25 Jun 2026 11:35:45 -0500 Subject: [PATCH 1/5] refactor(sessions): type reminder/job IDs as value objects + extract InFlightTurnDedup Convert reminder/background-job identities from string to the existing ReminderId/BackgroundJobId value objects along the dedup + journaling path (MessageSource, ChannelInput, SessionState dedup ledgers, TurnRecorded event, TurnCompleted output, channel reminder-delivery observer dictionaries). Value objects map to the existing bare-string proto/DTO fields at the serialization boundary, so persisted journals/snapshots and the SignalR wire form are byte-identical (proven by new SerializationRoundTripTests). Extract the in-flight reminder/job dedup state out of LlmSessionActor into a new Sessions/Handlers/InFlightTurnDedup handler, mirroring ActiveExecutionTracker. --- .../Channels/ChannelPipelineAckTargetTests.cs | 3 +- .../Contracts/SessionBindingContractTests.cs | 10 +-- .../Channels/DiscordConversationActorTests.cs | 3 +- .../Channels/DiscordGatewayActorTests.cs | 5 +- .../MattermostConversationActorTests.cs | 3 +- .../Channels/MessageSourceFactoryTests.cs | 5 +- .../Channels/SlackActorHierarchyTests.cs | 3 +- .../Channels/TurnContextTests.cs | 6 +- .../Jobs/BackgroundJobIntegrationTests.cs | 2 +- .../Jobs/BackgroundJobManagerActorTests.cs | 2 +- .../Protocol/SerializationRoundTripTests.cs | 67 ++++++++++++++++ .../Reminders/ReminderManagerActorTests.cs | 10 +-- .../BackgroundJobSessionStateTests.cs | 8 +- .../Sessions/LlmSessionIntegrationTests.cs | 3 +- .../Sessions/SessionStateTests.cs | 19 ++--- .../Sessions/SubAgentSpawnIntegrationTests.cs | 3 +- src/Netclaw.Actors/Channels/ChannelInput.cs | 3 +- src/Netclaw.Actors/Channels/MessageSource.cs | 6 +- .../Jobs/BackgroundJobManagerActor.cs | 2 +- .../Protocol/SessionOutputDtoMapper.cs | 5 +- .../Reminders/ReminderExecutionActor.cs | 2 +- .../Serialization/NetclawProtoMapper.cs | 14 ++-- .../Sessions/Handlers/InFlightTurnDedup.cs | 54 +++++++++++++ .../Sessions/LlmSessionActor.cs | 78 +++++++------------ .../Sessions/SessionProtocol.Events.cs | 6 +- .../Sessions/SessionProtocol.Outputs.cs | 3 +- src/Netclaw.Actors/Sessions/SessionState.cs | 21 ++--- .../DiscordSessionBindingActor.cs | 16 ++-- .../MattermostSessionBindingActor.cs | 16 ++-- .../SlackThreadBindingActor.cs | 16 ++-- .../Gateway/SignalRSessionActor.cs | 16 ++-- 31 files changed, 267 insertions(+), 143 deletions(-) create mode 100644 src/Netclaw.Actors/Sessions/Handlers/InFlightTurnDedup.cs diff --git a/src/Netclaw.Actors.Tests/Channels/ChannelPipelineAckTargetTests.cs b/src/Netclaw.Actors.Tests/Channels/ChannelPipelineAckTargetTests.cs index 4fb870ba1..281bc01d9 100644 --- a/src/Netclaw.Actors.Tests/Channels/ChannelPipelineAckTargetTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/ChannelPipelineAckTargetTests.cs @@ -11,6 +11,7 @@ using Microsoft.Extensions.AI; using Netclaw.Actors.Channels; using Netclaw.Actors.Protocol; +using Netclaw.Actors.Reminders; using Netclaw.Configuration; using Xunit; using static Netclaw.Actors.Sessions.SessionProtocol; @@ -91,7 +92,7 @@ await Source.Single(cmd) SenderId = new SenderId("user-1"), Contents = [new TextContent("hello")], ReceivedAt = DateTimeOffset.UtcNow, - ReminderId = reminderId, + ReminderId = reminderId is null ? null : new ReminderId(reminderId), AckTarget = ackTarget, Audience = TrustAudience.Public, Boundary = TrustBoundary.Public, diff --git a/src/Netclaw.Actors.Tests/Channels/Contracts/SessionBindingContractTests.cs b/src/Netclaw.Actors.Tests/Channels/Contracts/SessionBindingContractTests.cs index 16f55f900..36cd420a6 100644 --- a/src/Netclaw.Actors.Tests/Channels/Contracts/SessionBindingContractTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/Contracts/SessionBindingContractTests.cs @@ -276,7 +276,7 @@ public async Task Reminder_delivery_reports_success_when_post_succeeds() var pipeline = new RecordingSessionPipeline(_ => [ new TextOutput("reminder output") { SessionId = sid }, - new TurnCompleted { SessionId = sid, TurnNumber = new Netclaw.Actors.Protocol.TurnNumber(1), SourceReminderId = reminderKey } + new TurnCompleted { SessionId = sid, TurnNumber = new Netclaw.Actors.Protocol.TurnNumber(1), SourceReminderId = new ReminderId(reminderKey) } ], reactive: true); var observer = CreateTestProbe(); @@ -305,7 +305,7 @@ public async Task Reminder_delivery_reports_failure_when_post_throws() var pipeline = new RecordingSessionPipeline(_ => [ new TextOutput("reminder output") { SessionId = sid }, - new TurnCompleted { SessionId = sid, TurnNumber = new Netclaw.Actors.Protocol.TurnNumber(1), SourceReminderId = reminderKey } + new TurnCompleted { SessionId = sid, TurnNumber = new Netclaw.Actors.Protocol.TurnNumber(1), SourceReminderId = new ReminderId(reminderKey) } ], reactive: true); SetReplyClientThrows(new InvalidOperationException("channel API down")); @@ -339,9 +339,9 @@ public async Task Concurrent_reminders_to_same_session_each_get_their_own_result var pipeline = new RecordingSessionPipeline(_ => [ new TextOutput("reply A") { SessionId = sid }, - new TurnCompleted { SessionId = sid, TurnNumber = new Netclaw.Actors.Protocol.TurnNumber(1), SourceReminderId = keyA }, + new TurnCompleted { SessionId = sid, TurnNumber = new Netclaw.Actors.Protocol.TurnNumber(1), SourceReminderId = new ReminderId(keyA) }, new TextOutput("reply B") { SessionId = sid }, - new TurnCompleted { SessionId = sid, TurnNumber = new Netclaw.Actors.Protocol.TurnNumber(2), SourceReminderId = keyB } + new TurnCompleted { SessionId = sid, TurnNumber = new Netclaw.Actors.Protocol.TurnNumber(2), SourceReminderId = new ReminderId(keyB) } ], reactive: true); var observerA = CreateTestProbe(); @@ -412,7 +412,7 @@ private MessageSource CreateReminderSource(string reminderKey, IActorRef deliver SourceKind = new SourceKind("reminder") }, ReceivedAt = DateTimeOffset.UnixEpoch, - ReminderId = reminderKey, + ReminderId = new ReminderId(reminderKey), DeliveryObserver = deliveryObserver }; diff --git a/src/Netclaw.Actors.Tests/Channels/DiscordConversationActorTests.cs b/src/Netclaw.Actors.Tests/Channels/DiscordConversationActorTests.cs index 13d340068..e88879161 100644 --- a/src/Netclaw.Actors.Tests/Channels/DiscordConversationActorTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/DiscordConversationActorTests.cs @@ -11,6 +11,7 @@ using Microsoft.Extensions.Hosting; using Netclaw.Actors.Channels; using Netclaw.Actors.Protocol; +using Netclaw.Actors.Reminders; using Netclaw.Actors.Tests.Channels.TestHelpers; using Netclaw.Channels.Discord; using Netclaw.Configuration; @@ -543,6 +544,6 @@ private static DiscordGatewayMessage CreateMessage( { SourceKind = new Netclaw.Actors.Channels.SourceKind("reminder") }, - ReminderId = "rem-1" + ReminderId = new ReminderId("rem-1") }; } diff --git a/src/Netclaw.Actors.Tests/Channels/DiscordGatewayActorTests.cs b/src/Netclaw.Actors.Tests/Channels/DiscordGatewayActorTests.cs index 5ef9ef75f..9ad1d6e4b 100644 --- a/src/Netclaw.Actors.Tests/Channels/DiscordGatewayActorTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/DiscordGatewayActorTests.cs @@ -11,6 +11,7 @@ using Microsoft.Extensions.Hosting; using Netclaw.Actors.Channels; using Netclaw.Actors.Protocol; +using Netclaw.Actors.Reminders; using Netclaw.Actors.Tests.Channels.TestHelpers; using Netclaw.Channels.Discord; using Netclaw.Configuration; @@ -251,7 +252,7 @@ public async Task Gateway_routes_trusted_session_turn_to_conversation_actor() { SourceKind = new Netclaw.Actors.Channels.SourceKind("reminder") }, - ReminderId = "rem-1" + ReminderId = new ReminderId("rem-1") }); gateway.Tell(turn); @@ -285,7 +286,7 @@ public async Task Gateway_nacks_trusted_session_turn_with_invalid_session_id() { SourceKind = new Netclaw.Actors.Channels.SourceKind("reminder") }, - ReminderId = "rem-1" + ReminderId = new ReminderId("rem-1") }); gateway.Tell(turn, TestActor); diff --git a/src/Netclaw.Actors.Tests/Channels/MattermostConversationActorTests.cs b/src/Netclaw.Actors.Tests/Channels/MattermostConversationActorTests.cs index 3c572bf98..6e1564b54 100644 --- a/src/Netclaw.Actors.Tests/Channels/MattermostConversationActorTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/MattermostConversationActorTests.cs @@ -11,6 +11,7 @@ using Microsoft.Extensions.Hosting; using Netclaw.Actors.Channels; using Netclaw.Actors.Protocol; +using Netclaw.Actors.Reminders; using Netclaw.Actors.Tests.Channels.TestHelpers; using Netclaw.Channels.Mattermost; using Netclaw.Configuration; @@ -573,6 +574,6 @@ private static MattermostGatewayMessage CreateMessage( { SourceKind = new SourceKind("reminder") }, - ReminderId = "rem-1" + ReminderId = new ReminderId("rem-1") }; } diff --git a/src/Netclaw.Actors.Tests/Channels/MessageSourceFactoryTests.cs b/src/Netclaw.Actors.Tests/Channels/MessageSourceFactoryTests.cs index 562e7b74b..95d7dbba7 100644 --- a/src/Netclaw.Actors.Tests/Channels/MessageSourceFactoryTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/MessageSourceFactoryTests.cs @@ -8,6 +8,7 @@ using Akka.Hosting.TestKit; using Microsoft.Extensions.AI; using Netclaw.Actors.Channels; +using Netclaw.Actors.Reminders; using Netclaw.Configuration; using Netclaw.Tools; using Xunit; @@ -41,7 +42,7 @@ private static ChannelInput BuildInput( ?? new SourceProvenance(TransportAuthenticity.Verified, PayloadTaint.Public), Contents = [new TextContent("hello")], ReceivedAt = DateTimeOffset.UtcNow, - ReminderId = reminderId, + ReminderId = reminderId is null ? null : new ReminderId(reminderId), AckTarget = ackTarget, DefaultDeliveryTarget = defaultDeliveryTarget, RequestedDeliveryTarget = requestedDeliveryTarget, @@ -96,7 +97,7 @@ public void Create_propagates_ReminderId_and_AckTarget_from_ChannelInput() var result = MessageSourceFactory.Create( input, new SessionPipelineOptions { ChannelType = ChannelType.Slack }, new Netclaw.Actors.Protocol.TurnId("turn-1")); - Assert.Equal("check-pr:1712000000000", result.ReminderId); + Assert.Equal(new ReminderId("check-pr:1712000000000"), result.ReminderId); Assert.Same(probe.Ref, result.AckTarget); } diff --git a/src/Netclaw.Actors.Tests/Channels/SlackActorHierarchyTests.cs b/src/Netclaw.Actors.Tests/Channels/SlackActorHierarchyTests.cs index 3b93313a5..560172556 100644 --- a/src/Netclaw.Actors.Tests/Channels/SlackActorHierarchyTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/SlackActorHierarchyTests.cs @@ -11,6 +11,7 @@ using Microsoft.Extensions.Hosting; using Netclaw.Actors.Channels; using Netclaw.Actors.Protocol; +using Netclaw.Actors.Reminders; using Netclaw.Actors.Tests.Channels.TestHelpers; using Netclaw.Channels.Slack; using Netclaw.Configuration; @@ -442,7 +443,7 @@ public async Task Conversation_rejects_DeliverTrustedSessionTurn_for_other_chann SourceKind = new Netclaw.Actors.Channels.SourceKind("reminder") }, ReceivedAt = DateTimeOffset.UtcNow, - ReminderId = reminderId + ReminderId = new ReminderId(reminderId) }; } diff --git a/src/Netclaw.Actors.Tests/Channels/TurnContextTests.cs b/src/Netclaw.Actors.Tests/Channels/TurnContextTests.cs index f995f0a73..35a918b20 100644 --- a/src/Netclaw.Actors.Tests/Channels/TurnContextTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/TurnContextTests.cs @@ -5,7 +5,9 @@ // ----------------------------------------------------------------------- using Akka.Actor; using Netclaw.Actors.Channels; +using Netclaw.Actors.Jobs; using Netclaw.Actors.Protocol; +using Netclaw.Actors.Reminders; using Netclaw.Actors.Sessions; using Netclaw.Configuration; using Netclaw.Tools; @@ -54,8 +56,8 @@ public void FromMessageSource_captures_durable_authority_fields() AdoptedContextProjection = "quoted context", AdoptedContextLowerBound = "1700000000.000000", AdoptedContextUpperBound = "1700000000.000001", - ReminderId = "reminder:1700000000000", - BackgroundJobId = "bg-job:42", + ReminderId = new ReminderId("reminder:1700000000000"), + BackgroundJobId = new BackgroundJobId("bg-job:42"), AckTarget = ActorRefs.Nobody }; diff --git a/src/Netclaw.Actors.Tests/Jobs/BackgroundJobIntegrationTests.cs b/src/Netclaw.Actors.Tests/Jobs/BackgroundJobIntegrationTests.cs index f8163effa..dd4116393 100644 --- a/src/Netclaw.Actors.Tests/Jobs/BackgroundJobIntegrationTests.cs +++ b/src/Netclaw.Actors.Tests/Jobs/BackgroundJobIntegrationTests.cs @@ -98,7 +98,7 @@ public async Task BackgroundJob_Completes_And_DeliversResult_ViaGateway() Assert.Equal(PrincipalClassification.VerifiedAutomation, delivered.Source.Principal); Assert.Equal("background-job", delivered.Source.Provenance.SourceKind?.Value); Assert.NotNull(delivered.Source.BackgroundJobId); - Assert.StartsWith("bg-job:", delivered.Source.BackgroundJobId); + Assert.StartsWith("bg-job:", delivered.Source.BackgroundJobId!.Value.Value); await AwaitAssertAsync(() => { diff --git a/src/Netclaw.Actors.Tests/Jobs/BackgroundJobManagerActorTests.cs b/src/Netclaw.Actors.Tests/Jobs/BackgroundJobManagerActorTests.cs index a5221106b..3c097df6d 100644 --- a/src/Netclaw.Actors.Tests/Jobs/BackgroundJobManagerActorTests.cs +++ b/src/Netclaw.Actors.Tests/Jobs/BackgroundJobManagerActorTests.cs @@ -253,7 +253,7 @@ await manager.Ask( Assert.Contains(logPath, delivery.Content); Assert.Contains("Server running on", delivery.Content); Assert.Equal(TrustAudience.Personal, delivery.Source.Audience); - Assert.Equal($"bg-job:{orphanId.Value}", delivery.Source.BackgroundJobId); + Assert.Equal(new BackgroundJobId($"bg-job:{orphanId.Value}"), delivery.Source.BackgroundJobId); } [Fact] diff --git a/src/Netclaw.Actors.Tests/Protocol/SerializationRoundTripTests.cs b/src/Netclaw.Actors.Tests/Protocol/SerializationRoundTripTests.cs index 3a447877b..096d1f19b 100644 --- a/src/Netclaw.Actors.Tests/Protocol/SerializationRoundTripTests.cs +++ b/src/Netclaw.Actors.Tests/Protocol/SerializationRoundTripTests.cs @@ -9,8 +9,10 @@ using Google.Protobuf; using Netclaw.Actors.Channels; using Netclaw.Actors.Hosting; +using Netclaw.Actors.Jobs; using Netclaw.Actors.Protocol; using Netclaw.Actors.Reminders; +using Netclaw.Actors.Serialization; using Netclaw.Actors.Sessions; using Netclaw.Tools; using Xunit; @@ -130,6 +132,71 @@ public void TurnRecorded_round_trips() Assert.Equal(original.RecordedAtMs, result.RecordedAtMs); } + [Fact] + public void TurnRecorded_round_trips_preserving_value_object_source_ids() + { + var original = new TurnRecorded + { + SessionId = new SessionId("C99999/1708531200.000100"), + UserMessage = new SerializableChatMessage { Role = ChatRole.User, Content = "check PR" }, + AssistantReply = new SerializableChatMessage { Role = ChatRole.Assistant, Content = "merged" }, + RecordedAtMs = 1_700_000_000_000, + SourceReminderId = new ReminderId("check-pr:1712000000000"), + SourceBackgroundJobId = new BackgroundJobId("bg-job:abc123") + }; + + var result = RoundTrip(original); + + Assert.Equal(original.SourceReminderId, result.SourceReminderId); + Assert.Equal(original.SourceBackgroundJobId, result.SourceBackgroundJobId); + } + + [Fact] + public void TurnRecorded_round_trips_with_null_source_ids() + { + var original = new TurnRecorded + { + SessionId = new SessionId("C99999/1708531200.000100"), + UserMessage = new SerializableChatMessage { Role = ChatRole.User, Content = "hi" }, + AssistantReply = new SerializableChatMessage { Role = ChatRole.Assistant, Content = "hello" }, + RecordedAtMs = 1_700_000_000_000 + }; + + var result = RoundTrip(original); + + Assert.Null(result.SourceReminderId); + Assert.Null(result.SourceBackgroundJobId); + } + + [Fact] + public void TurnRecorded_value_object_source_ids_use_bare_string_proto_fields() + { + // Wire-compat: the reminder/background-job value objects map to the SAME + // bare-string proto fields the pre-value-object code used, so old journals + // deserialize unchanged and new journals are byte-identical. Proven at the + // proto-mapper boundary in both directions. + var evt = new TurnRecorded + { + SessionId = new SessionId("C99999/1708531200.000100"), + UserMessage = new SerializableChatMessage { Role = ChatRole.User, Content = "x" }, + AssistantReply = new SerializableChatMessage { Role = ChatRole.Assistant, Content = "y" }, + RecordedAtMs = 1_700_000_000_000, + SourceReminderId = new ReminderId("check-pr:1712000000000"), + SourceBackgroundJobId = new BackgroundJobId("bg-job:abc123") + }; + + // Forward: value object -> bare string on the wire (no nested object). + var proto = NetclawProtoMapper.ToProto(evt); + Assert.Equal("check-pr:1712000000000", proto.SourceReminderId); + Assert.Equal("bg-job:abc123", proto.SourceBackgroundJobId); + + // Reverse: an "old" proto carrying bare strings deserializes into the + // value-object-typed event. + var restored = NetclawProtoMapper.FromProto(proto); + Assert.Equal(new ReminderId("check-pr:1712000000000"), restored.SourceReminderId); + Assert.Equal(new BackgroundJobId("bg-job:abc123"), restored.SourceBackgroundJobId); + } + [Fact] public void SessionCompacted_round_trips_with_messages() { diff --git a/src/Netclaw.Actors.Tests/Reminders/ReminderManagerActorTests.cs b/src/Netclaw.Actors.Tests/Reminders/ReminderManagerActorTests.cs index d17712330..0369c2da9 100644 --- a/src/Netclaw.Actors.Tests/Reminders/ReminderManagerActorTests.cs +++ b/src/Netclaw.Actors.Tests/Reminders/ReminderManagerActorTests.cs @@ -510,7 +510,7 @@ public async Task Mode_B_reminder_dispatches_to_resolved_gateway_and_completes_o Assert.Equal(TrustAudience.Team, delivered.Source.Audience); Assert.Equal(TrustBoundary.TrustedInstance, delivered.Source.Boundary); Assert.NotNull(delivered.Source.ReminderId); - Assert.StartsWith("mode-b-anchor:", delivered.Source.ReminderId); + Assert.StartsWith("mode-b-anchor:", delivered.Source.ReminderId!.Value.Value); Assert.Equal(PrincipalClassification.VerifiedAutomation, delivered.Source.Principal); Assert.Equal("reminder", delivered.Source.Provenance.SourceKind?.Value); @@ -659,7 +659,7 @@ public async Task CurrentSession_delivery_required_fails_fast_on_explicit_delive // Channel reports the post failed — execution must report failure // (so Akka.Reminders redelivers) without acking the envelope. delivered.Source.DeliveryObserver!.Tell(new ReminderDeliveryResult( - delivered.Source.ReminderId!, + delivered.Source.ReminderId!.Value.Value, ChannelType.Slack, Delivered: false, FailureReason: "channel API down")); @@ -710,7 +710,7 @@ public async Task CurrentSession_delivery_key_is_built_from_envelope_fire_time() var delivered = await gatewayProbe.ExpectMsgAsync( TimeSpan.FromSeconds(5), cancellationToken: TestContext.Current.CancellationToken); - Assert.Equal($"{definition.Id}:{fireTime.ToUnixTimeMilliseconds()}", delivered.Source.ReminderId); + Assert.Equal(new ReminderId($"{definition.Id}:{fireTime.ToUnixTimeMilliseconds()}"), delivered.Source.ReminderId); } [Fact] @@ -740,7 +740,7 @@ public async Task CurrentSession_delivery_required_succeeds_when_delivery_is_obs Assert.NotNull(delivered.Source.DeliveryObserver); delivered.Source.DeliveryObserver!.Tell(new ReminderDeliveryResult( - delivered.Source.ReminderId!, + delivered.Source.ReminderId!.Value.Value, ChannelType.Slack, Delivered: true, ObservedAtMs: TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds())); @@ -1024,7 +1024,7 @@ public async Task CurrentSession_discord_delivery_required_succeeds_when_deliver Assert.NotNull(delivered.Source.DeliveryObserver); delivered.Source.DeliveryObserver!.Tell(new ReminderDeliveryResult( - delivered.Source.ReminderId!, + delivered.Source.ReminderId!.Value.Value, ChannelType.Discord, Delivered: true, ObservedAtMs: TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds())); diff --git a/src/Netclaw.Actors.Tests/Sessions/BackgroundJobSessionStateTests.cs b/src/Netclaw.Actors.Tests/Sessions/BackgroundJobSessionStateTests.cs index df9fcf43c..8f0f05341 100644 --- a/src/Netclaw.Actors.Tests/Sessions/BackgroundJobSessionStateTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/BackgroundJobSessionStateTests.cs @@ -66,13 +66,13 @@ public void TurnRecorded_WithSourceBackgroundJobId_DedupAndRemovesActive() UserMessage = new SerializableChatMessage { Role = ChatRole.User, Content = "result" }, AssistantReply = new SerializableChatMessage { Role = ChatRole.Assistant, Content = "ok" }, RecordedAtMs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), - SourceBackgroundJobId = jobKey + SourceBackgroundJobId = new BackgroundJobId(jobKey) }; state = state.Apply(evt); Assert.Empty(state.ActiveBackgroundJobs); - Assert.Contains(jobKey, state.ProcessedBackgroundJobIds); + Assert.Contains(new BackgroundJobId(jobKey), state.ProcessedBackgroundJobIds); } [Fact] @@ -209,7 +209,7 @@ public void Compaction_Preserves_ActiveJobsAndDedupSet() var processedKey = "bg-job:already-done"; state = state with { - ProcessedBackgroundJobIds = state.ProcessedBackgroundJobIds.Add(processedKey) + ProcessedBackgroundJobIds = state.ProcessedBackgroundJobIds.Add(new BackgroundJobId(processedKey)) }; var compactedEvt = new SessionCompacted @@ -225,6 +225,6 @@ public void Compaction_Preserves_ActiveJobsAndDedupSet() Assert.Single(state.ActiveBackgroundJobs); Assert.True(state.ActiveBackgroundJobs.ContainsKey(jobKey)); - Assert.Contains(processedKey, state.ProcessedBackgroundJobIds); + Assert.Contains(new BackgroundJobId(processedKey), state.ProcessedBackgroundJobIds); } } diff --git a/src/Netclaw.Actors.Tests/Sessions/LlmSessionIntegrationTests.cs b/src/Netclaw.Actors.Tests/Sessions/LlmSessionIntegrationTests.cs index 3a857737a..9da7e4c09 100644 --- a/src/Netclaw.Actors.Tests/Sessions/LlmSessionIntegrationTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/LlmSessionIntegrationTests.cs @@ -15,6 +15,7 @@ using Netclaw.Actors.Memory; using Netclaw.Actors.Channels; using Netclaw.Actors.Protocol; +using Netclaw.Actors.Reminders; using Netclaw.Actors.Sessions; using Netclaw.Actors.Tools; using Xunit; @@ -1660,7 +1661,7 @@ await sessionManager.Ask(new SendUserMessage SourceKind = new Netclaw.Actors.Channels.SourceKind("reminder") }, ReceivedAt = _timeProvider.GetUtcNow(), - ReminderId = reminderId + ReminderId = new ReminderId(reminderId) }; } diff --git a/src/Netclaw.Actors.Tests/Sessions/SessionStateTests.cs b/src/Netclaw.Actors.Tests/Sessions/SessionStateTests.cs index c8820c888..b7357a12a 100644 --- a/src/Netclaw.Actors.Tests/Sessions/SessionStateTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/SessionStateTests.cs @@ -5,6 +5,7 @@ // ----------------------------------------------------------------------- using System.Collections.Immutable; using Netclaw.Actors.Protocol; +using Netclaw.Actors.Reminders; using Netclaw.Actors.Sessions; using Xunit; using static Netclaw.Actors.Sessions.SessionProtocol; @@ -525,12 +526,12 @@ public void Apply_TurnRecorded_folds_SourceReminderId_into_ProcessedReminderIds( SessionId = TestSessionId, UserMessage = new SerializableChatMessage { Role = ChatRole.User, Content = "check PR" }, AssistantReply = new SerializableChatMessage { Role = ChatRole.Assistant, Content = "merged" }, - SourceReminderId = "check-pr:1712000000000" + SourceReminderId = new ReminderId("check-pr:1712000000000") }; var next = state.Apply(evt); - Assert.Contains("check-pr:1712000000000", next.ProcessedReminderIds); + Assert.Contains(new ReminderId("check-pr:1712000000000"), next.ProcessedReminderIds); Assert.Single(next.ProcessedReminderIds); } @@ -559,7 +560,7 @@ public void Apply_TurnRecorded_replay_builds_cumulative_dedup_set() SessionId = TestSessionId, UserMessage = new SerializableChatMessage { Role = ChatRole.User, Content = "r1" }, AssistantReply = new SerializableChatMessage { Role = ChatRole.Assistant, Content = "ok" }, - SourceReminderId = "r1:100" + SourceReminderId = new ReminderId("r1:100") }); state = state.Apply(new TurnRecorded { @@ -573,12 +574,12 @@ public void Apply_TurnRecorded_replay_builds_cumulative_dedup_set() SessionId = TestSessionId, UserMessage = new SerializableChatMessage { Role = ChatRole.User, Content = "r2" }, AssistantReply = new SerializableChatMessage { Role = ChatRole.Assistant, Content = "ok" }, - SourceReminderId = "r2:200" + SourceReminderId = new ReminderId("r2:200") }); Assert.Equal(2, state.ProcessedReminderIds.Count); - Assert.Contains("r1:100", state.ProcessedReminderIds); - Assert.Contains("r2:200", state.ProcessedReminderIds); + Assert.Contains(new ReminderId("r1:100"), state.ProcessedReminderIds); + Assert.Contains(new ReminderId("r2:200"), state.ProcessedReminderIds); } [Fact] @@ -590,7 +591,7 @@ public void Apply_SessionCompacted_preserves_ProcessedReminderIds() SessionId = TestSessionId, UserMessage = new SerializableChatMessage { Role = ChatRole.User, Content = "r1" }, AssistantReply = new SerializableChatMessage { Role = ChatRole.Assistant, Content = "ok" }, - SourceReminderId = "preserved:1" + SourceReminderId = new ReminderId("preserved:1") }); var compacted = state.Apply(new SessionCompacted @@ -602,7 +603,7 @@ public void Apply_SessionCompacted_preserves_ProcessedReminderIds() ] }); - Assert.Contains("preserved:1", compacted.ProcessedReminderIds); + Assert.Contains(new ReminderId("preserved:1"), compacted.ProcessedReminderIds); } [Fact] @@ -619,7 +620,7 @@ public void ProcessedReminderIds_is_not_persisted_in_snapshot() SessionId = TestSessionId, UserMessage = new SerializableChatMessage { Role = ChatRole.User, Content = "r1" }, AssistantReply = new SerializableChatMessage { Role = ChatRole.Assistant, Content = "ok" }, - SourceReminderId = "lost-on-snapshot:1" + SourceReminderId = new ReminderId("lost-on-snapshot:1") }); Assert.NotEmpty(state.ProcessedReminderIds); diff --git a/src/Netclaw.Actors.Tests/Sessions/SubAgentSpawnIntegrationTests.cs b/src/Netclaw.Actors.Tests/Sessions/SubAgentSpawnIntegrationTests.cs index c57f84a32..974515284 100644 --- a/src/Netclaw.Actors.Tests/Sessions/SubAgentSpawnIntegrationTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/SubAgentSpawnIntegrationTests.cs @@ -10,6 +10,7 @@ using Netclaw.Actors.Channels; using Netclaw.Actors.Hosting; using Netclaw.Actors.Protocol; +using Netclaw.Actors.Reminders; using Netclaw.Actors.Skills; using Netclaw.Actors.SubAgents; using Netclaw.Actors.Sessions; @@ -743,7 +744,7 @@ private static MessageSource BuildReminderSource(string? reminderId = null) Principal = PrincipalClassification.VerifiedAutomation, Provenance = new SourceProvenance(TransportAuthenticity.LocalProcess, PayloadTaint.Trusted), ReceivedAt = DateTimeOffset.UtcNow, - ReminderId = reminderId + ReminderId = reminderId is null ? null : new ReminderId(reminderId) }; } diff --git a/src/Netclaw.Actors/Channels/ChannelInput.cs b/src/Netclaw.Actors/Channels/ChannelInput.cs index 506e199dc..731e3eb02 100644 --- a/src/Netclaw.Actors/Channels/ChannelInput.cs +++ b/src/Netclaw.Actors/Channels/ChannelInput.cs @@ -5,6 +5,7 @@ // ----------------------------------------------------------------------- using Akka.Actor; using Microsoft.Extensions.AI; +using Netclaw.Actors.Reminders; using Netclaw.Configuration; using Netclaw.Tools; @@ -141,7 +142,7 @@ public bool HasAdoptedContext /// by /// . Null for regular inbound ingress. /// - public string? ReminderId { get; init; } + public ReminderId? ReminderId { get; init; } /// /// Ephemeral ack reply target. Set by channel leaf actors when handling diff --git a/src/Netclaw.Actors/Channels/MessageSource.cs b/src/Netclaw.Actors/Channels/MessageSource.cs index 96753dc30..e02f2058e 100644 --- a/src/Netclaw.Actors/Channels/MessageSource.cs +++ b/src/Netclaw.Actors/Channels/MessageSource.cs @@ -4,6 +4,8 @@ // // ----------------------------------------------------------------------- using Akka.Actor; +using Netclaw.Actors.Jobs; +using Netclaw.Actors.Reminders; using Netclaw.Configuration; using Netclaw.Tools; @@ -147,13 +149,13 @@ public bool HasAdoptedContext /// is recorded. This field is runtime-only — /// is never serialized. /// - public string? ReminderId { get; init; } + public ReminderId? ReminderId { get; init; } /// /// Ephemeral dedup key for background-job-originated deliveries. /// Format is "bg-job:{jobId}". Null for regular user messages. /// - public string? BackgroundJobId { get; init; } + public BackgroundJobId? BackgroundJobId { get; init; } /// /// Optional reply target for ack-gated trusted deliveries. When set, diff --git a/src/Netclaw.Actors/Jobs/BackgroundJobManagerActor.cs b/src/Netclaw.Actors/Jobs/BackgroundJobManagerActor.cs index 8b489c7a2..29e4a8018 100644 --- a/src/Netclaw.Actors/Jobs/BackgroundJobManagerActor.cs +++ b/src/Netclaw.Actors/Jobs/BackgroundJobManagerActor.cs @@ -431,7 +431,7 @@ private void DeliverResultToSession(BackgroundJobCompleted completed, Background SourceKind = new SourceKind(BackgroundJobManagerActor.SourceKind) }, ReceivedAt = _timeProvider.GetUtcNow(), - BackgroundJobId = jobDeliveryKey + BackgroundJobId = new BackgroundJobId(jobDeliveryKey) }; var deliverMsg = new DeliverTrustedSessionTurn(sessionId, content, source); diff --git a/src/Netclaw.Actors/Protocol/SessionOutputDtoMapper.cs b/src/Netclaw.Actors/Protocol/SessionOutputDtoMapper.cs index 29da8a9a3..a5e6b8586 100644 --- a/src/Netclaw.Actors/Protocol/SessionOutputDtoMapper.cs +++ b/src/Netclaw.Actors/Protocol/SessionOutputDtoMapper.cs @@ -3,6 +3,7 @@ // Copyright (C) 2026 - 2026 Petabridge, LLC // // ----------------------------------------------------------------------- +using Netclaw.Actors.Reminders; using Netclaw.Media; using static Netclaw.Actors.Sessions.SessionProtocol; @@ -91,7 +92,7 @@ public static class SessionOutputDtoMapper TimestampMs = msg.TimestampMs, TurnNumber = msg.TurnNumber, TurnOutcome = msg.Outcome.ToString().ToLowerInvariant(), - SourceReminderId = msg.SourceReminderId + SourceReminderId = msg.SourceReminderId?.Value }, SessionTitleOutput msg => new SessionOutputDto @@ -268,7 +269,7 @@ public static SessionOutput FromDto(SessionOutputDto dto) Outcome = Enum.TryParse(dto.TurnOutcome, ignoreCase: true, out var outcome) ? outcome : TurnOutcome.Completed, - SourceReminderId = dto.SourceReminderId + SourceReminderId = dto.SourceReminderId is null ? null : new ReminderId(dto.SourceReminderId) }, SessionOutputTypes.SessionTitle => new SessionTitleOutput(dto.Title ?? string.Empty) { diff --git a/src/Netclaw.Actors/Reminders/ReminderExecutionActor.cs b/src/Netclaw.Actors/Reminders/ReminderExecutionActor.cs index 5abe109f2..a9b3c6507 100644 --- a/src/Netclaw.Actors/Reminders/ReminderExecutionActor.cs +++ b/src/Netclaw.Actors/Reminders/ReminderExecutionActor.cs @@ -229,7 +229,7 @@ private async Task InitializeCurrentSessionAsync() SourceKind = new SourceKind("reminder") }, ReceivedAt = _dispatchedAt, - ReminderId = reminderDeliveryKey, + ReminderId = new ReminderId(reminderDeliveryKey), // Only reminders that gate on delivery need a confirmation // channel. The binding actor tells this ref a // ReminderDeliveryResult on turn completion; leaving it null diff --git a/src/Netclaw.Actors/Serialization/NetclawProtoMapper.cs b/src/Netclaw.Actors/Serialization/NetclawProtoMapper.cs index 5eabd3d30..4fe58285b 100644 --- a/src/Netclaw.Actors/Serialization/NetclawProtoMapper.cs +++ b/src/Netclaw.Actors/Serialization/NetclawProtoMapper.cs @@ -159,10 +159,12 @@ internal static Proto.TurnRecordedProto ToProto(TurnRecorded evt) AssistantReply = ToProto(evt.AssistantReply), RecordedAtMs = evt.RecordedAtMs }; - if (evt.SourceReminderId is not null) - proto.SourceReminderId = evt.SourceReminderId; - if (evt.SourceBackgroundJobId is not null) - proto.SourceBackgroundJobId = evt.SourceBackgroundJobId; + // Value objects map to the existing string proto fields, so the on-disk + // form is byte-identical to the pre-value-object representation. + if (evt.SourceReminderId is { } reminderId) + proto.SourceReminderId = reminderId.Value; + if (evt.SourceBackgroundJobId is { } backgroundJobId) + proto.SourceBackgroundJobId = backgroundJobId.Value; return proto; } @@ -172,8 +174,8 @@ internal static Proto.TurnRecordedProto ToProto(TurnRecorded evt) UserMessage = FromProto(proto.UserMessage), AssistantReply = FromProto(proto.AssistantReply), RecordedAtMs = proto.RecordedAtMs, - SourceReminderId = proto.HasSourceReminderId ? proto.SourceReminderId : null, - SourceBackgroundJobId = proto.HasSourceBackgroundJobId ? proto.SourceBackgroundJobId : null + SourceReminderId = proto.HasSourceReminderId ? new ReminderId(proto.SourceReminderId) : (ReminderId?)null, + SourceBackgroundJobId = proto.HasSourceBackgroundJobId ? new BackgroundJobId(proto.SourceBackgroundJobId) : (BackgroundJobId?)null }; // ── SessionTitleSet ── diff --git a/src/Netclaw.Actors/Sessions/Handlers/InFlightTurnDedup.cs b/src/Netclaw.Actors/Sessions/Handlers/InFlightTurnDedup.cs new file mode 100644 index 000000000..3e36e8a15 --- /dev/null +++ b/src/Netclaw.Actors/Sessions/Handlers/InFlightTurnDedup.cs @@ -0,0 +1,54 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- + +using Netclaw.Actors.Jobs; +using Netclaw.Actors.Reminders; + +namespace Netclaw.Actors.Sessions.Handlers; + +/// +/// Tracks reminder- and background-job-originated turns that are accepted but +/// not yet recorded ("in flight"), so an Akka.Reminders redelivery (or a +/// duplicate job result) is deduplicated before the persisted ledgers +/// (SessionState.ProcessedReminderIds / ProcessedBackgroundJobIds) +/// catch it. Transient and actor-owned: never persisted, rebuilt from journal +/// replay on recovery. Mirrors . +/// +internal sealed class InFlightTurnDedup +{ + private readonly HashSet _reminders = []; + private readonly HashSet _backgroundJobs = []; + + public bool IsReminderInFlight(ReminderId? reminderId) => + reminderId is { } id && _reminders.Contains(id); + + public void ReserveReminder(ReminderId? reminderId) + { + if (reminderId is { } id && !string.IsNullOrEmpty(id.Value)) + _reminders.Add(id); + } + + public void CompleteReminder(ReminderId? reminderId) + { + if (reminderId is { } id) + _reminders.Remove(id); + } + + public bool IsBackgroundJobInFlight(BackgroundJobId? backgroundJobId) => + backgroundJobId is { } id && _backgroundJobs.Contains(id); + + public void ReserveBackgroundJob(BackgroundJobId? backgroundJobId) + { + if (backgroundJobId is { } id && !string.IsNullOrEmpty(id.Value)) + _backgroundJobs.Add(id); + } + + public void CompleteBackgroundJob(BackgroundJobId? backgroundJobId) + { + if (backgroundJobId is { } id) + _backgroundJobs.Remove(id); + } +} diff --git a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs index b302d8f2f..82ed13662 100644 --- a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs +++ b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs @@ -12,8 +12,10 @@ using Netclaw.Actors.Hosting; using Microsoft.Extensions.AI; using Netclaw.Actors.Channels; +using Netclaw.Actors.Jobs; using Netclaw.Actors.Memory; using Netclaw.Actors.Protocol; +using Netclaw.Actors.Reminders; using Netclaw.Actors.Skills; using Netclaw.Actors.SubAgents; using Netclaw.Actors.Sessions.Handlers; @@ -72,8 +74,8 @@ public sealed class LlmSessionActor : ReceivePersistentActor, IWithTimers // Transient state (not persisted) private readonly List _buffer = []; - private readonly HashSet _inFlightReminderIds = new(StringComparer.Ordinal); - private readonly HashSet _inFlightBackgroundJobIds = new(StringComparer.Ordinal); + // In-flight reminder/background-job dedup (transient; rebuilt from journal on recovery). + private readonly InFlightTurnDedup _inFlightDedup = new(); private readonly SessionSubscriberManager _subscribers = new(); private readonly List _availableTools = []; private readonly Dictionary _pendingToolInteractions = new(StringComparer.Ordinal); @@ -2047,13 +2049,13 @@ private void HandleTextResponse( Persist(turnEvent, evt => { - CompleteReminderInFlight(evt.SourceReminderId); - CompleteBackgroundJobInFlight(evt.SourceBackgroundJobId); + _inFlightDedup.CompleteReminder(evt.SourceReminderId); + _inFlightDedup.CompleteBackgroundJob(evt.SourceBackgroundJobId); var processed = _state.ProcessedReminderIds; - if (!string.IsNullOrEmpty(evt.SourceReminderId)) + if (evt.SourceReminderId is { } reminderId && !string.IsNullOrEmpty(reminderId.Value)) { - processed = processed.Add(evt.SourceReminderId); + processed = processed.Add(reminderId); } _state = (_state with @@ -2196,8 +2198,8 @@ private void HandleIncomingUserMessage(SendUserMessage cmd) return; } - ReserveInFlightReminderId(reminderId); - ReserveInFlightBackgroundJobId(bgJobId); + _inFlightDedup.ReserveReminder(reminderId); + _inFlightDedup.ReserveBackgroundJob(bgJobId); // A new inbound message while a tool batch is still parked on an // approval gate means the user abandoned that approval. This is only @@ -2326,66 +2328,38 @@ private void ContinueIncomingUserMessage(SendUserMessage cmd) TransitionTo(SessionPhase.Processing); } - private bool IsReminderDedupHit(string? reminderId, bool includeBuffered) + private bool IsReminderDedupHit(ReminderId? reminderId, bool includeBuffered) { - if (string.IsNullOrEmpty(reminderId)) + if (reminderId is not { } id || string.IsNullOrEmpty(id.Value)) return false; - if (_state.ProcessedReminderIds.Contains(reminderId)) + if (_state.ProcessedReminderIds.Contains(id)) return true; - if (_inFlightReminderIds.Contains(reminderId)) + if (_inFlightDedup.IsReminderInFlight(id)) return true; if (!includeBuffered) return false; - return _buffer.Any(buffered => - !string.IsNullOrEmpty(buffered.Source?.ReminderId) - && string.Equals(buffered.Source!.ReminderId, reminderId, StringComparison.Ordinal)); + return _buffer.Any(buffered => buffered.Source?.ReminderId == id); } - private void ReserveInFlightReminderId(string? reminderId) + private bool IsBackgroundJobDedupHit(BackgroundJobId? bgJobId, bool includeBuffered) { - if (!string.IsNullOrEmpty(reminderId)) - _inFlightReminderIds.Add(reminderId); - } - - private void CompleteReminderInFlight(string? reminderId) - { - if (!string.IsNullOrEmpty(reminderId)) - _inFlightReminderIds.Remove(reminderId); - } - - private bool IsBackgroundJobDedupHit(string? bgJobId, bool includeBuffered) - { - if (string.IsNullOrEmpty(bgJobId)) + if (bgJobId is not { } id || string.IsNullOrEmpty(id.Value)) return false; - if (_state.ProcessedBackgroundJobIds.Contains(bgJobId)) + if (_state.ProcessedBackgroundJobIds.Contains(id)) return true; - if (_inFlightBackgroundJobIds.Contains(bgJobId)) + if (_inFlightDedup.IsBackgroundJobInFlight(id)) return true; if (!includeBuffered) return false; - return _buffer.Any(buffered => - !string.IsNullOrEmpty(buffered.Source?.BackgroundJobId) - && string.Equals(buffered.Source!.BackgroundJobId, bgJobId, StringComparison.Ordinal)); - } - - private void ReserveInFlightBackgroundJobId(string? bgJobId) - { - if (!string.IsNullOrEmpty(bgJobId)) - _inFlightBackgroundJobIds.Add(bgJobId); - } - - private void CompleteBackgroundJobInFlight(string? bgJobId) - { - if (!string.IsNullOrEmpty(bgJobId)) - _inFlightBackgroundJobIds.Remove(bgJobId); + return _buffer.Any(buffered => buffered.Source?.BackgroundJobId == id); } private bool ShouldCompact() @@ -3133,13 +3107,13 @@ private void HandleRoutedSkillExecutionCompleted(RoutedSkillExecutionCompleted m Persist(turnEvent, evt => { - CompleteReminderInFlight(evt.SourceReminderId); - CompleteBackgroundJobInFlight(evt.SourceBackgroundJobId); + _inFlightDedup.CompleteReminder(evt.SourceReminderId); + _inFlightDedup.CompleteBackgroundJob(evt.SourceBackgroundJobId); var processed = _state.ProcessedReminderIds; - if (!string.IsNullOrEmpty(evt.SourceReminderId)) + if (evt.SourceReminderId is { } reminderId && !string.IsNullOrEmpty(reminderId.Value)) { - processed = processed.Add(evt.SourceReminderId); + processed = processed.Add(reminderId); } _state = (_state with @@ -4244,8 +4218,8 @@ private ToolBatchAbandoned BuildToolBatchAbandonedEvent(string resultContent) private void FailCurrentTurn(string errorMessage, Exception cause, ErrorCategory category = ErrorCategory.Unknown) { - CompleteReminderInFlight(_currentTurnSource?.ReminderId); - CompleteBackgroundJobInFlight(_currentTurnSource?.BackgroundJobId); + _inFlightDedup.CompleteReminder(_currentTurnSource?.ReminderId); + _inFlightDedup.CompleteBackgroundJob(_currentTurnSource?.BackgroundJobId); CancelAndDisposeToolExecutionCts(); _deliveryRetry.Clear(); _pendingToolInteractions.Clear(); diff --git a/src/Netclaw.Actors/Sessions/SessionProtocol.Events.cs b/src/Netclaw.Actors/Sessions/SessionProtocol.Events.cs index 82fcce0b5..660e962a3 100644 --- a/src/Netclaw.Actors/Sessions/SessionProtocol.Events.cs +++ b/src/Netclaw.Actors/Sessions/SessionProtocol.Events.cs @@ -3,7 +3,9 @@ // Copyright (C) 2026 - 2026 Petabridge, LLC // // ----------------------------------------------------------------------- +using Netclaw.Actors.Jobs; using Netclaw.Actors.Protocol; +using Netclaw.Actors.Reminders; using Netclaw.Actors.Serialization; using Netclaw.Configuration; using Netclaw.Security; @@ -37,7 +39,7 @@ public sealed record TurnRecorded : ISessionEvent /// () from /// event replay. /// - public string? SourceReminderId { get; init; } + public ReminderId? SourceReminderId { get; init; } /// /// Populated when this turn originated from a background job result delivery. @@ -45,7 +47,7 @@ public sealed record TurnRecorded : ISessionEvent /// by the /// background job manager. Null for regular user turns. /// - public string? SourceBackgroundJobId { get; init; } + public BackgroundJobId? SourceBackgroundJobId { get; init; } public DateTimeOffset RecordedAt => DateTimeOffset.FromUnixTimeMilliseconds(RecordedAtMs); diff --git a/src/Netclaw.Actors/Sessions/SessionProtocol.Outputs.cs b/src/Netclaw.Actors/Sessions/SessionProtocol.Outputs.cs index 0d58a28dd..a5f304f6f 100644 --- a/src/Netclaw.Actors/Sessions/SessionProtocol.Outputs.cs +++ b/src/Netclaw.Actors/Sessions/SessionProtocol.Outputs.cs @@ -5,6 +5,7 @@ // ----------------------------------------------------------------------- using Akka.Actor; using Netclaw.Actors.Protocol; +using Netclaw.Actors.Reminders; using Netclaw.Configuration; using Netclaw.Media; using Netclaw.Security; @@ -167,7 +168,7 @@ public sealed record TurnCompleted : SessionOutput /// Reminder delivery key ({reminderId}:{fireTimestampMs}) for reminder-sourced turns. /// Null for non-reminder turns. /// - public string? SourceReminderId { get; init; } + public ReminderId? SourceReminderId { get; init; } } /// diff --git a/src/Netclaw.Actors/Sessions/SessionState.cs b/src/Netclaw.Actors/Sessions/SessionState.cs index 48958f0bd..69dff1336 100644 --- a/src/Netclaw.Actors/Sessions/SessionState.cs +++ b/src/Netclaw.Actors/Sessions/SessionState.cs @@ -6,6 +6,7 @@ using System.Collections.Immutable; using Netclaw.Actors.Jobs; using Netclaw.Actors.Protocol; +using Netclaw.Actors.Reminders; using static Netclaw.Actors.Sessions.SessionProtocol; namespace Netclaw.Actors.Sessions; @@ -70,8 +71,8 @@ public sealed record AdoptedContextAuditMessage( /// boundaries are an explicitly accepted tradeoff; see /// reminder-session-reentry design doc D2. /// - public IImmutableSet ProcessedReminderIds { get; init; } = - ImmutableHashSet.Empty; + public IImmutableSet ProcessedReminderIds { get; init; } = + ImmutableHashSet.Empty; /// /// Background jobs this session is waiting on. Persisted to snapshot @@ -88,16 +89,16 @@ public sealed record AdoptedContextAuditMessage( /// Same pattern as — not persisted to /// snapshot, rebuilds from event replay. /// - public IImmutableSet ProcessedBackgroundJobIds { get; init; } = - ImmutableHashSet.Empty; + public IImmutableSet ProcessedBackgroundJobIds { get; init; } = + ImmutableHashSet.Empty; // ── Event application (pure functions) ── public SessionState Apply(TurnRecorded evt) { var processedReminders = ProcessedReminderIds; - if (!string.IsNullOrEmpty(evt.SourceReminderId)) - processedReminders = processedReminders.Add(evt.SourceReminderId); + if (evt.SourceReminderId is { } reminderId && !string.IsNullOrEmpty(reminderId.Value)) + processedReminders = processedReminders.Add(reminderId); // Background-job dedup/remove/prune is delegated to the single shared // helper so the replay path here and the live turn-completion path in @@ -118,14 +119,14 @@ public SessionState Apply(TurnRecorded evt) /// surfaced in this turn's context block (so the agent learns of a reap /// exactly once instead of on every turn forever). /// - public SessionState CompleteTurnBackgroundJobBookkeeping(string? sourceBackgroundJobId) + public SessionState CompleteTurnBackgroundJobBookkeeping(BackgroundJobId? sourceBackgroundJobId) { var processedJobs = ProcessedBackgroundJobIds; var activeJobs = ActiveBackgroundJobs; - if (!string.IsNullOrEmpty(sourceBackgroundJobId)) + if (sourceBackgroundJobId is { } jobId && !string.IsNullOrEmpty(jobId.Value)) { - processedJobs = processedJobs.Add(sourceBackgroundJobId); - activeJobs = activeJobs.Remove(sourceBackgroundJobId); + processedJobs = processedJobs.Add(jobId); + activeJobs = activeJobs.Remove(jobId.Value); } activeJobs = PruneReaped(activeJobs); diff --git a/src/Netclaw.Channels.Discord/DiscordSessionBindingActor.cs b/src/Netclaw.Channels.Discord/DiscordSessionBindingActor.cs index 7b04a87e9..24fd8f321 100644 --- a/src/Netclaw.Channels.Discord/DiscordSessionBindingActor.cs +++ b/src/Netclaw.Channels.Discord/DiscordSessionBindingActor.cs @@ -75,7 +75,7 @@ internal sealed class DiscordSessionBindingActor : ReceivePersistentActor, IWith // told a ReminderDeliveryResult on its turn's TurnCompleted and removed. // Keyed (not a single field) because multiple reminders can target the // same session concurrently — a single field would be clobbered. - private readonly Dictionary _reminderDeliveryObservers = new(StringComparer.Ordinal); + private readonly Dictionary _reminderDeliveryObservers = new(); private Netclaw.Actors.Protocol.TurnNumber _turnNumber; private string? _lastSetThreadName; private ulong? _cursorSnowflake; @@ -1065,8 +1065,9 @@ private async Task HandleTrustedReminderAsync(DeliverTrustedSessionTurn message) // second concurrent reminder to this session can't overwrite the // first's observer before its turn reaches TurnCompleted. if (message.Source.DeliveryObserver is { } deliveryObserver - && !string.IsNullOrWhiteSpace(message.Source.ReminderId)) - _reminderDeliveryObservers[message.Source.ReminderId] = deliveryObserver; + && message.Source.ReminderId is { } reminderKey + && !string.IsNullOrWhiteSpace(reminderKey.Value)) + _reminderDeliveryObservers[reminderKey] = deliveryObserver; try { @@ -1194,11 +1195,12 @@ private async Task HandleOutputReceivedAsync(OutputReceived msg) AdvanceCursor(pendingSnowflake); _pendingCursorSnowflake = null; - if (!string.IsNullOrWhiteSpace(completed.SourceReminderId) - && _reminderDeliveryObservers.Remove(completed.SourceReminderId, out var reminderObserver)) + if (completed.SourceReminderId is { } sourceReminderKey + && !string.IsNullOrWhiteSpace(sourceReminderKey.Value) + && _reminderDeliveryObservers.Remove(sourceReminderKey, out var reminderObserver)) { reminderObserver.Tell(new ReminderDeliveryResult( - completed.SourceReminderId, + sourceReminderKey.Value, ChannelType.Discord, Delivered: _deliveredThisTurn, FailureReason: _deliveredThisTurn ? null : "Discord post did not succeed", @@ -1437,7 +1439,7 @@ private void FailPendingReminderDeliveries(string reason) return; foreach (var (key, observer) in _reminderDeliveryObservers) - observer.Tell(new ReminderDeliveryResult(key, ChannelType.Discord, Delivered: false, FailureReason: reason)); + observer.Tell(new ReminderDeliveryResult(key.Value, ChannelType.Discord, Delivered: false, FailureReason: reason)); _reminderDeliveryObservers.Clear(); } diff --git a/src/Netclaw.Channels.Mattermost/MattermostSessionBindingActor.cs b/src/Netclaw.Channels.Mattermost/MattermostSessionBindingActor.cs index 4e91a3348..120cf2ae1 100644 --- a/src/Netclaw.Channels.Mattermost/MattermostSessionBindingActor.cs +++ b/src/Netclaw.Channels.Mattermost/MattermostSessionBindingActor.cs @@ -74,7 +74,7 @@ internal sealed class MattermostSessionBindingActor : ReceivePersistentActor, IW // told a ReminderDeliveryResult on its turn's TurnCompleted and removed. // Keyed (not a single field) because multiple reminders can target the // same session concurrently — a single field would be clobbered. - private readonly Dictionary _reminderDeliveryObservers = new(StringComparer.Ordinal); + private readonly Dictionary _reminderDeliveryObservers = new(); private TurnNumber _turnNumber; private string? _cursorPostId; private string? _pendingCursorPostId; @@ -1043,8 +1043,9 @@ private async Task HandleTrustedReminderAsync(DeliverTrustedSessionTurn message) // second concurrent reminder to this session can't overwrite the // first's observer before its turn reaches TurnCompleted. if (message.Source.DeliveryObserver is { } deliveryObserver - && !string.IsNullOrWhiteSpace(message.Source.ReminderId)) - _reminderDeliveryObservers[message.Source.ReminderId] = deliveryObserver; + && message.Source.ReminderId is { } reminderKey + && !string.IsNullOrWhiteSpace(reminderKey.Value)) + _reminderDeliveryObservers[reminderKey] = deliveryObserver; try { @@ -1165,11 +1166,12 @@ private async Task HandleOutputReceivedAsync(OutputReceived msg) AdvanceCursor(pendingCursor); _pendingCursorPostId = null; - if (!string.IsNullOrWhiteSpace(completed.SourceReminderId) - && _reminderDeliveryObservers.Remove(completed.SourceReminderId, out var reminderObserver)) + if (completed.SourceReminderId is { } sourceReminderKey + && !string.IsNullOrWhiteSpace(sourceReminderKey.Value) + && _reminderDeliveryObservers.Remove(sourceReminderKey, out var reminderObserver)) { reminderObserver.Tell(new ReminderDeliveryResult( - completed.SourceReminderId, + sourceReminderKey.Value, ChannelType.Mattermost, Delivered: _deliveredThisTurn, FailureReason: _deliveredThisTurn ? null : "Mattermost post did not succeed", @@ -1377,7 +1379,7 @@ private void FailPendingReminderDeliveries(string reason) return; foreach (var (key, observer) in _reminderDeliveryObservers) - observer.Tell(new ReminderDeliveryResult(key, ChannelType.Mattermost, Delivered: false, FailureReason: reason)); + observer.Tell(new ReminderDeliveryResult(key.Value, ChannelType.Mattermost, Delivered: false, FailureReason: reason)); _reminderDeliveryObservers.Clear(); } diff --git a/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs b/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs index 8e5943be4..e0e74b60e 100644 --- a/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs +++ b/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs @@ -45,7 +45,7 @@ internal sealed class SlackThreadBindingActor : ReceivePersistentActor, IWithTim // told a ReminderDeliveryResult on its turn's TurnCompleted and removed. // Keyed (not a single field) because multiple reminders can target the // same session concurrently — a single field would be clobbered. - private readonly Dictionary _reminderDeliveryObservers = new(StringComparer.Ordinal); + private readonly Dictionary _reminderDeliveryObservers = new(); private readonly List _pendingApprovalRequests = []; // Gates the text-approval cold path (TryHandleColdTextApprovalResponseAsync). @@ -285,8 +285,9 @@ private async Task HandleTrustedReminderAsync(DeliverTrustedSessionTurn message) // second concurrent reminder to this session can't overwrite the // first's observer before its turn reaches TurnCompleted. if (message.Source.DeliveryObserver is { } deliveryObserver - && !string.IsNullOrWhiteSpace(message.Source.ReminderId)) - _reminderDeliveryObservers[message.Source.ReminderId] = deliveryObserver; + && message.Source.ReminderId is { } reminderKey + && !string.IsNullOrWhiteSpace(reminderKey.Value)) + _reminderDeliveryObservers[reminderKey] = deliveryObserver; try { @@ -1112,12 +1113,13 @@ private async Task HandleOutputAsync(ThreadOutput threadOutput) AdvanceCursor(pendingTs); _pendingCursorTs = null; - if (!string.IsNullOrWhiteSpace(completed.SourceReminderId) - && _reminderDeliveryObservers.Remove(completed.SourceReminderId, out var reminderObserver)) + if (completed.SourceReminderId is { } sourceReminderKey + && !string.IsNullOrWhiteSpace(sourceReminderKey.Value) + && _reminderDeliveryObservers.Remove(sourceReminderKey, out var reminderObserver)) { var delivered = _postedThisTurn || _uploadedFileThisTurn; reminderObserver.Tell(new ReminderDeliveryResult( - completed.SourceReminderId, + sourceReminderKey.Value, ChannelType.Slack, Delivered: delivered, FailureReason: delivered ? null : "Slack post did not succeed", @@ -1469,7 +1471,7 @@ private void FailPendingReminderDeliveries(string reason) return; foreach (var (key, observer) in _reminderDeliveryObservers) - observer.Tell(new ReminderDeliveryResult(key, ChannelType.Slack, Delivered: false, FailureReason: reason)); + observer.Tell(new ReminderDeliveryResult(key.Value, ChannelType.Slack, Delivered: false, FailureReason: reason)); _reminderDeliveryObservers.Clear(); } diff --git a/src/Netclaw.Daemon/Gateway/SignalRSessionActor.cs b/src/Netclaw.Daemon/Gateway/SignalRSessionActor.cs index a5bfaf4d8..2e6d235b1 100644 --- a/src/Netclaw.Daemon/Gateway/SignalRSessionActor.cs +++ b/src/Netclaw.Daemon/Gateway/SignalRSessionActor.cs @@ -38,7 +38,7 @@ internal sealed class SignalRSessionActor : ReceiveActor, IWithUnboundedStash, I // told a ReminderDeliveryResult on its turn's TurnCompleted and removed. // Keyed (not a single field) because multiple reminders can target the // same session concurrently — a single field would be clobbered. - private readonly Dictionary _reminderDeliveryObservers = new(StringComparer.Ordinal); + private readonly Dictionary _reminderDeliveryObservers = new(); private static readonly TimeSpan PipelineInitTimeout = TimeSpan.FromSeconds(15); private static readonly object ReinitializeTimerKey = new(); @@ -220,8 +220,9 @@ await _handle.ReinitializeAsync( // second concurrent reminder to this session can't overwrite the // first's observer before its turn reaches TurnCompleted. if (msg.Source.DeliveryObserver is { } deliveryObserver - && !string.IsNullOrWhiteSpace(msg.Source.ReminderId)) - _reminderDeliveryObservers[msg.Source.ReminderId] = deliveryObserver; + && msg.Source.ReminderId is { } reminderKey + && !string.IsNullOrWhiteSpace(reminderKey.Value)) + _reminderDeliveryObservers[reminderKey] = deliveryObserver; try { @@ -315,11 +316,12 @@ private async Task HandleOutputReceivedAsync(OutputReceived msg) /// private void ReportReminderDeliveryResult(TurnCompleted completed, bool delivered) { - if (!string.IsNullOrWhiteSpace(completed.SourceReminderId) - && _reminderDeliveryObservers.Remove(completed.SourceReminderId!, out var observer)) + if (completed.SourceReminderId is { } sourceReminderKey + && !string.IsNullOrWhiteSpace(sourceReminderKey.Value) + && _reminderDeliveryObservers.Remove(sourceReminderKey, out var observer)) { observer.Tell(new ReminderDeliveryResult( - completed.SourceReminderId!, + sourceReminderKey.Value, _channelType, Delivered: delivered, FailureReason: delivered ? null : "SignalR client did not receive the reply", @@ -339,7 +341,7 @@ private void FailPendingReminderDeliveries(string reason) return; foreach (var (key, observer) in _reminderDeliveryObservers) - observer.Tell(new ReminderDeliveryResult(key, _channelType, Delivered: false, FailureReason: reason)); + observer.Tell(new ReminderDeliveryResult(key.Value, _channelType, Delivered: false, FailureReason: reason)); _reminderDeliveryObservers.Clear(); } From 25a9a70e7b2ae25932c51683eaf79007f57f1fe3 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Thu, 25 Jun 2026 11:39:09 -0500 Subject: [PATCH 2/5] refactor(sessions): extract ModelInputMediaBuffer handler Move the transient model-input media buffer out of LlmSessionActor into a new Sessions/Handlers/ModelInputMediaBuffer. Accumulation, drain-on-batch-complete, and reset are behavior-preserving; AddModelInputMediaNudge (which mutates _state) stays in the actor. DrainSnapshot returns the buffered references and clears in one call, collapsing the prior nudge-then-clear pairs. --- .../Handlers/ModelInputMediaBuffer.cs | 36 +++++++++++++++++++ .../Sessions/LlmSessionActor.cs | 17 +++++---- 2 files changed, 44 insertions(+), 9 deletions(-) create mode 100644 src/Netclaw.Actors/Sessions/Handlers/ModelInputMediaBuffer.cs diff --git a/src/Netclaw.Actors/Sessions/Handlers/ModelInputMediaBuffer.cs b/src/Netclaw.Actors/Sessions/Handlers/ModelInputMediaBuffer.cs new file mode 100644 index 000000000..36d6e5240 --- /dev/null +++ b/src/Netclaw.Actors/Sessions/Handlers/ModelInputMediaBuffer.cs @@ -0,0 +1,36 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- + +using Netclaw.Actors.Protocol; + +namespace Netclaw.Actors.Sessions.Handlers; + +/// +/// Buffers media references that tools load for model-visible inspection during +/// a streamed tool batch. References accumulate per tool result and are drained +/// into a single system nudge once the batch completes, or cleared when the turn +/// fails or its tool-batch tracking resets. Transient and actor-owned: never +/// persisted, rebuilt implicitly per batch. +/// +internal sealed class ModelInputMediaBuffer +{ + private readonly List _pending = []; + + public void Add(IEnumerable references) => _pending.AddRange(references); + + /// Returns the buffered references and clears the buffer. + public IReadOnlyList DrainSnapshot() + { + if (_pending.Count == 0) + return []; + + var snapshot = _pending.ToArray(); + _pending.Clear(); + return snapshot; + } + + public void Clear() => _pending.Clear(); +} diff --git a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs index 82ed13662..ae16f8a2f 100644 --- a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs +++ b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs @@ -83,7 +83,9 @@ public sealed class LlmSessionActor : ReceivePersistentActor, IWithTimers // Live-only coordination for the currently executing streamed tool batch. // Durable recovery derives unanswered calls from _state.History. private readonly ActiveToolBatchTracker _activeToolBatch = new(); - private readonly List _pendingModelInputMediaReferences = []; + // Media loaded by tools for model-visible inspection during a streamed tool + // batch; drained into a system nudge when the batch completes. + private readonly ModelInputMediaBuffer _mediaBuffer = new(); private MessageSource? _currentTurnSource; private TurnContext? _currentTurnContext; private bool _processingStateActive; @@ -558,7 +560,7 @@ private void Processing() { _watchdog.Stop(Timers); CancelAndDisposeToolExecutionCts(); - _pendingModelInputMediaReferences.Clear(); + _mediaBuffer.Clear(); TurnLog().Error(msg.Cause, "turn_tool_execution_failed"); const string errorMessage = "I encountered an error executing a tool. Please try again."; @@ -3277,13 +3279,11 @@ private void ApplyToolCallRecorded(ToolCallRecorded evt) if (!alreadyRecorded) { _state = _state with { History = _state.History.Add(evt.ToolResult) }; - if (evt.ToolResult.MediaReferences.Count > 0) - _pendingModelInputMediaReferences.AddRange(evt.ToolResult.MediaReferences); + _mediaBuffer.Add(evt.ToolResult.MediaReferences); if (_activeToolBatch.HasAllResults) { - AddModelInputMediaNudge(_pendingModelInputMediaReferences); - _pendingModelInputMediaReferences.Clear(); + AddModelInputMediaNudge(_mediaBuffer.DrainSnapshot()); } } } @@ -3433,7 +3433,7 @@ private void ApplyToolBatchAbandoned(ToolBatchAbandoned evt) private void ClearActiveToolBatchTracking() { _activeToolBatch.Clear(); - _pendingModelInputMediaReferences.Clear(); + _mediaBuffer.Clear(); } private void MaybeSnapshot() @@ -4494,8 +4494,7 @@ private void TryCompleteStreamedToolBatch() private void CompleteToolBatch(int resultCount) { - AddModelInputMediaNudge(_pendingModelInputMediaReferences); - _pendingModelInputMediaReferences.Clear(); + AddModelInputMediaNudge(_mediaBuffer.DrainSnapshot()); var budgetStatus = _turnState.RecordToolCompletion(resultCount, _config.MaxToolIterationsPerTurn); From fdbd8afc8d8c61c9439b1bc86a71962febac4e7a Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Thu, 25 Jun 2026 12:01:09 -0500 Subject: [PATCH 3/5] refactor(sessions): extract SessionPhaseMachine handler Move the explicit SessionPhase field and legal-transition validation out of LlmSessionActor into a new Sessions/Handlers/SessionPhaseMachine. TransitionTo now delegates validation to _phase.TryTransition, but the Become() switch, processing-state emission, and observer notification stay in the actor so phase tracking and behavior remain co-located (no drift). Drops the now-redundant IsLegalTransition wrapper. --- .../Sessions/Handlers/SessionPhaseMachine.cs | 35 +++++++++++++++++++ .../Sessions/LlmSessionActor.cs | 17 ++++----- 2 files changed, 41 insertions(+), 11 deletions(-) create mode 100644 src/Netclaw.Actors/Sessions/Handlers/SessionPhaseMachine.cs diff --git a/src/Netclaw.Actors/Sessions/Handlers/SessionPhaseMachine.cs b/src/Netclaw.Actors/Sessions/Handlers/SessionPhaseMachine.cs new file mode 100644 index 000000000..3f3d6dc5d --- /dev/null +++ b/src/Netclaw.Actors/Sessions/Handlers/SessionPhaseMachine.cs @@ -0,0 +1,35 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- + +namespace Netclaw.Actors.Sessions.Handlers; + +/// +/// Owns the session's explicit and enforces legal +/// transitions (per ). This is the metadata +/// + validation layer only — the actor still drives the matching Become() +/// behavior so phase tracking and behavior stay co-located there. Transient and +/// actor-owned; starts at . +/// +internal sealed class SessionPhaseMachine +{ + public SessionPhase Current { get; private set; } = SessionPhase.Recovering; + + /// + /// Attempts a validated transition to . On success, + /// advances and reports the prior phase via + /// ; on an illegal transition returns false and + /// leaves unchanged. + /// + public bool TryTransition(SessionPhase target, out SessionPhase from) + { + from = Current; + if (!SessionPhaseTransitions.IsLegal(Current, target)) + return false; + + Current = target; + return true; + } +} diff --git a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs index ae16f8a2f..acc1bc1e7 100644 --- a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs +++ b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs @@ -201,7 +201,7 @@ public sealed class LlmSessionActor : ReceivePersistentActor, IWithTimers private SessionState _state = SessionState.Empty; // Explicit state machine phase (metadata + validation layer over Become()) - private SessionPhase _currentPhase = SessionPhase.Recovering; + private readonly SessionPhaseMachine _phase = new(); public override string PersistenceId { get; } public ITimerScheduler Timers { get; set; } = null!; @@ -342,12 +342,10 @@ public LlmSessionActor( /// private void TransitionTo(SessionPhase target) { - if (!IsLegalTransition(_currentPhase, target)) + if (!_phase.TryTransition(target, out var from)) throw new InvalidOperationException( - $"Illegal session phase transition: {_currentPhase} → {target}"); + $"Illegal session phase transition: {_phase.Current} → {target}"); - var from = _currentPhase; - _currentPhase = target; _log.Info("session_phase_transition from={From} to={To}", from, target); EmitProcessingStateForPhase(target); @@ -373,9 +371,6 @@ private void TransitionTo(SessionPhase target) } } - private static bool IsLegalTransition(SessionPhase from, SessionPhase to) - => SessionPhaseTransitions.IsLegal(from, to); - private void EmitProcessingStateForPhase(SessionPhase phase) { var isProcessing = phase is SessionPhase.Processing or SessionPhase.Compacting; @@ -2475,7 +2470,7 @@ private void CommandSubscriptionMessages() // If replay found an approval decision but no durable tool result, // do not replay the approved side effect after restart. Close the // orphaned tool_use blocks so the next user turn has valid history. - if (_currentPhase == SessionPhase.Ready) + if (_phase.Current == SessionPhase.Ready) { if (!AbandonResolvedToolBatchAfterRecovery()) AbandonInterruptedToolBatchAfterRecovery(); @@ -3360,7 +3355,7 @@ private void ApplyToolApprovalRequested(ToolApprovalRequested evt, bool persistA _resolvedToolApprovals.Remove(evt.CallId); if (persistApprovalState && turnContext is not null) - RecordWaitingApprovalState(turnContext, evt.CallId, recovered: _currentPhase == SessionPhase.Recovering); + RecordWaitingApprovalState(turnContext, evt.CallId, recovered: _phase.Current == SessionPhase.Recovering); else if (persistApprovalState && restoreFailure is not null) _log.Warning( "Approval request {CallId} could not restore turn context: {Reason}", @@ -4672,7 +4667,7 @@ private void RequestRestartDrain() _restartDrainRequested = true; _restartDrainReplyTo = Sender; - if (_currentPhase == SessionPhase.Ready) + if (_phase.Current == SessionPhase.Ready) TransitionTo(SessionPhase.Passivating); } From 44b91e05a1d626704d3f981d21be8b4e4ed50e84 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Thu, 25 Jun 2026 12:08:42 -0500 Subject: [PATCH 4/5] refactor(sessions): fold available-tools list into DiscoveredToolCache DiscoveredToolCache now owns the exposed tool list and base-tool count, removing the external-state-passing where LlmSessionActor handed its list to EvictAll/PrepareForNewTurn on every call. The actor seeds base tools once (SeedBaseTools), reads the list via AvailableTools, and adds discovered tools via AddIfMissing (returns whether it added, so the actor keeps its log line). Tool-access-policy orchestration (TryActivateDiscoveredTool, ResolveExposedToolsForCurrentTurn) stays in the actor: dragging the registry, access policy, and trust context into the cache would be plumbing without a win. --- .../Tools/DiscoveredToolCacheTests.cs | 46 +++++---- .../Sessions/Handlers/DiscoveredToolCache.cs | 94 ++++++++++++------- .../Sessions/LlmSessionActor.cs | 33 +++---- 3 files changed, 90 insertions(+), 83 deletions(-) diff --git a/src/Netclaw.Actors.Tests/Tools/DiscoveredToolCacheTests.cs b/src/Netclaw.Actors.Tests/Tools/DiscoveredToolCacheTests.cs index 44ed8e768..4e318c871 100644 --- a/src/Netclaw.Actors.Tests/Tools/DiscoveredToolCacheTests.cs +++ b/src/Netclaw.Actors.Tests/Tools/DiscoveredToolCacheTests.cs @@ -1,4 +1,4 @@ -// ----------------------------------------------------------------------- +// ----------------------------------------------------------------------- // // Copyright (C) 2026 - 2026 Petabridge, LLC // @@ -17,22 +17,21 @@ public void EvictAll_ClearsAllDiscoveredTools() { var registry = new ToolRegistry(); var cache = new DiscoveredToolCache(); - var availableTools = new List(); // Register and load 3 MCP tools - var tool1 = RegisterAndRemember(registry, cache, availableTools, "server", "tool_a", retentionTurns: 3, maxCount: 12); - var tool2 = RegisterAndRemember(registry, cache, availableTools, "server", "tool_b", retentionTurns: 3, maxCount: 12); - var tool3 = RegisterAndRemember(registry, cache, availableTools, "server", "tool_c", retentionTurns: 3, maxCount: 12); + RegisterAndRemember(registry, cache, "server", "tool_a", retentionTurns: 3, maxCount: 12); + RegisterAndRemember(registry, cache, "server", "tool_b", retentionTurns: 3, maxCount: 12); + RegisterAndRemember(registry, cache, "server", "tool_c", retentionTurns: 3, maxCount: 12); - Assert.Equal(3, availableTools.Count); + Assert.Equal(3, cache.AvailableTools.Count); Assert.True(cache.HasTool("server/tool_a")); Assert.True(cache.HasTool("server/tool_b")); Assert.True(cache.HasTool("server/tool_c")); // Evict all — simulates compaction reset - cache.EvictAll(availableTools, baseToolCount: 0); + cache.EvictAll(); - Assert.Empty(availableTools); + Assert.Empty(cache.AvailableTools); Assert.False(cache.HasTool("server/tool_a")); Assert.False(cache.HasTool("server/tool_b")); Assert.False(cache.HasTool("server/tool_c")); @@ -47,18 +46,17 @@ public void EvictAll_PreservesBaseTools() // Simulate 2 base tools + 1 discovered tool var baseTool1 = AIFunctionFactory.Create(() => "result", "search_tools"); var baseTool2 = AIFunctionFactory.Create(() => "result", "load_tool"); - var availableTools = new List { baseTool1, baseTool2 }; - var baseToolCount = 2; + cache.SeedBaseTools([baseTool1, baseTool2]); - RegisterAndRemember(registry, cache, availableTools, "notion", "search", retentionTurns: 3, maxCount: 12); + RegisterAndRemember(registry, cache, "notion", "search", retentionTurns: 3, maxCount: 12); - Assert.Equal(3, availableTools.Count); + Assert.Equal(3, cache.AvailableTools.Count); - cache.EvictAll(availableTools, baseToolCount); + cache.EvictAll(); - Assert.Equal(2, availableTools.Count); - Assert.Contains(baseTool1, availableTools); - Assert.Contains(baseTool2, availableTools); + Assert.Equal(2, cache.AvailableTools.Count); + Assert.Contains(baseTool1, cache.AvailableTools); + Assert.Contains(baseTool2, cache.AvailableTools); } [Fact] @@ -66,23 +64,21 @@ public void PrepareForNewTurn_AfterEvictAll_DoesNotRestoreEvictedTools() { var registry = new ToolRegistry(); var cache = new DiscoveredToolCache(); - var availableTools = new List(); - RegisterAndRemember(registry, cache, availableTools, "notion", "search", retentionTurns: 5, maxCount: 12); - Assert.Single(availableTools); + RegisterAndRemember(registry, cache, "notion", "search", retentionTurns: 5, maxCount: 12); + Assert.Single(cache.AvailableTools); - cache.EvictAll(availableTools, baseToolCount: 0); - Assert.Empty(availableTools); + cache.EvictAll(); + Assert.Empty(cache.AvailableTools); // Next turn — evicted tools should NOT come back - cache.PrepareForNewTurn(availableTools, baseToolCount: 0, retentionTurns: 5, maxCount: 12, registry); - Assert.Empty(availableTools); + cache.PrepareForNewTurn(retentionTurns: 5, maxCount: 12, registry); + Assert.Empty(cache.AvailableTools); } private static McpToolAdapter RegisterAndRemember( ToolRegistry registry, DiscoveredToolCache cache, - List availableTools, string serverName, string toolName, int retentionTurns, @@ -92,7 +88,7 @@ private static McpToolAdapter RegisterAndRemember( var adapter = new McpToolAdapter(fake, serverName, toolName); registry.Register(adapter); cache.Remember(adapter.Name, adapter, retentionTurns, maxCount); - availableTools.Add(adapter.ToAITool()); + cache.AddIfMissing(adapter.ToAITool()); return adapter; } } diff --git a/src/Netclaw.Actors/Sessions/Handlers/DiscoveredToolCache.cs b/src/Netclaw.Actors/Sessions/Handlers/DiscoveredToolCache.cs index e1e97db29..100c8afb7 100644 --- a/src/Netclaw.Actors/Sessions/Handlers/DiscoveredToolCache.cs +++ b/src/Netclaw.Actors/Sessions/Handlers/DiscoveredToolCache.cs @@ -1,4 +1,4 @@ -// ----------------------------------------------------------------------- +// ----------------------------------------------------------------------- // // Copyright (C) 2026 - 2026 Petabridge, LLC // @@ -10,29 +10,45 @@ namespace Netclaw.Actors.Sessions.Handlers; /// -/// Tracks MCP tools discovered via search_tools across turns, managing -/// lease-based retention and eviction. +/// Owns the session's exposed tool list — the always-loaded base tools plus the +/// MCP tools discovered via search_tools — and manages lease-based retention and +/// eviction of the discovered set across turns. Because the cache owns the list +/// it rebuilds, callers seed the base tools once and then drive +/// / / +/// without passing the list around. Transient and actor-owned; never persisted. /// internal sealed class DiscoveredToolCache { + private readonly List _availableTools = []; private readonly List _order = []; private readonly Dictionary _leases = new(StringComparer.Ordinal); + private int _baseToolCount; + + /// + /// The tools currently exposed to the model this turn: the base tools + /// followed by any discovered tools with an active lease. + /// + public IReadOnlyList AvailableTools => _availableTools; /// - /// Prepare the tool cache for a new turn: decrement leases, evict expired tools, + /// Seed the always-loaded base tools once at session start. Everything added + /// beyond this set is a discovered tool subject to lease-based eviction. + /// + public void SeedBaseTools(IReadOnlyList alwaysLoadedTools) + { + _availableTools.Clear(); + _availableTools.AddRange(alwaysLoadedTools); + _baseToolCount = _availableTools.Count; + } + + /// + /// Prepare the tool set for a new turn: decrement leases, evict expired tools, /// and rebuild the available tools list from the cache. /// - /// The mutable tools list owned by the actor. - /// Count of always-loaded tools (dynamic tools start after this index). /// Configured retention turns (0 or negative disables caching). /// Maximum discovered tools to retain. /// Tool registry for resolving tool instances. - public void PrepareForNewTurn( - List availableTools, - int baseToolCount, - int retentionTurns, - int maxCount, - ToolRegistry? registry) + public void PrepareForNewTurn(int retentionTurns, int maxCount, ToolRegistry? registry) { if (registry is null) return; @@ -41,13 +57,13 @@ public void PrepareForNewTurn( { _leases.Clear(); _order.Clear(); - TrimToBase(availableTools, baseToolCount); + TrimToBase(); return; } if (_leases.Count == 0) { - TrimToBase(availableTools, baseToolCount); + TrimToBase(); return; } @@ -66,7 +82,7 @@ public void PrepareForNewTurn( _order.RemoveAll(name => !_leases.ContainsKey(name)); } - RebuildFromCache(availableTools, baseToolCount, registry); + RebuildFromCache(registry); // Lease countdown happens after this turn's tool set is prepared, // so a lease value of N keeps tools available for N future turns. @@ -104,14 +120,15 @@ public void Remember(string toolName, INetclawTool tool, int leaseTurns, int max } /// - /// Evict all discovered tools and trim the available tools list back to base tools. - /// Used when an LLM call fails to prevent a bad tool set from poisoning subsequent turns. + /// Evict all discovered tools and trim the available tools list back to the + /// base tools. Used when an LLM call fails to prevent a bad tool set from + /// poisoning subsequent turns. /// - public void EvictAll(List availableTools, int baseToolCount) + public void EvictAll() { _leases.Clear(); _order.Clear(); - TrimToBase(availableTools, baseToolCount); + TrimToBase(); } /// @@ -122,12 +139,28 @@ public bool HasTool(string toolName) return _leases.TryGetValue(toolName, out var lease) && lease > 0; } + /// + /// Add a tool to the exposed list when no with the + /// same name is already present. Returns true if it was added. + /// + public bool AddIfMissing(AITool aiTool) + { + if (_availableTools.Any(existing => + existing is AIFunction ef && aiTool is AIFunction nf && ef.Name == nf.Name)) + { + return false; + } + + _availableTools.Add(aiTool); + return true; + } + /// /// Rebuild the available tools list from the discovered tool cache. /// - private void RebuildFromCache(List availableTools, int baseToolCount, ToolRegistry registry) + private void RebuildFromCache(ToolRegistry registry) { - TrimToBase(availableTools, baseToolCount); + TrimToBase(); foreach (var toolName in _order) { @@ -138,24 +171,13 @@ private void RebuildFromCache(List availableTools, int baseToolCount, To if (tool is null) continue; - AddIfMissing(availableTools, toolName, tool.ToAITool()); + AddIfMissing(tool.ToAITool()); } } - private static void TrimToBase(List availableTools, int baseToolCount) + private void TrimToBase() { - if (availableTools.Count > baseToolCount) - availableTools.RemoveRange(baseToolCount, availableTools.Count - baseToolCount); - } - - private static void AddIfMissing(List availableTools, string toolName, AITool aiTool) - { - if (availableTools.Any(existing => - existing is AIFunction ef && aiTool is AIFunction nf && ef.Name == nf.Name)) - { - return; - } - - availableTools.Add(aiTool); + if (_availableTools.Count > _baseToolCount) + _availableTools.RemoveRange(_baseToolCount, _availableTools.Count - _baseToolCount); } } diff --git a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs index acc1bc1e7..aa34da912 100644 --- a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs +++ b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs @@ -77,7 +77,6 @@ public sealed class LlmSessionActor : ReceivePersistentActor, IWithTimers // In-flight reminder/background-job dedup (transient; rebuilt from journal on recovery). private readonly InFlightTurnDedup _inFlightDedup = new(); private readonly SessionSubscriberManager _subscribers = new(); - private readonly List _availableTools = []; private readonly Dictionary _pendingToolInteractions = new(StringComparer.Ordinal); private readonly Dictionary _resolvedToolApprovals = new(StringComparer.Ordinal); // Live-only coordination for the currently executing streamed tool batch. @@ -93,7 +92,7 @@ public sealed class LlmSessionActor : ReceivePersistentActor, IWithTimers private readonly ToolRegistry? _fullRegistry; private readonly ToolAccessPolicy? _toolAccessPolicy; private readonly TrustContextDeriver? _trustContextDeriver; - private int _baseToolCount; // count of always-loaded tools; dynamic tools appended after this + // Owns the exposed tool list (base + discovered) and lease-based eviction. private readonly DiscoveredToolCache _discoveredToolCache = new(); // Last observed input token count from LLM response (for compaction trigger) @@ -255,9 +254,8 @@ public LlmSessionActor( _fullRegistry = tools?.ToolRegistry; if (_fullRegistry is not null) { - _availableTools.AddRange(_fullRegistry.GetAlwaysLoadedTools()); + _discoveredToolCache.SeedBaseTools(_fullRegistry.GetAlwaysLoadedTools()); } - _baseToolCount = _availableTools.Count; // ── Recovery handlers ── Recover(evt => @@ -638,7 +636,7 @@ private void Processing() // Evict discovered tools to prevent a poisoned tool set from cascading // across turns (e.g., oversized Notion schemas causing repeated 502s). - _discoveredToolCache.EvictAll(_availableTools, _baseToolCount); + _discoveredToolCache.EvictAll(); TurnLog().Info("turn_discovered_tools_evicted — tool list reset to base tools after LLM call failure"); var errorMessage = ExtractLlmErrorMessage(msg.Cause); @@ -1235,7 +1233,7 @@ private void HandleCompactionWorkCompleted(CompactionWorkCompleted msg) _lastInputTokenCount = 0; _startupContextInjected = false; _recallManager.ResetForCompaction(); - _discoveredToolCache.EvictAll(_availableTools, _baseToolCount); + _discoveredToolCache.EvictAll(); EnqueueCheckpointFireAndForget(new MemoryCheckpointRequest( SessionId: _sessionId, @@ -2282,7 +2280,6 @@ private void ContinueIncomingUserMessage(SendUserMessage cmd) _turnState.ResetForNewTurn(); _discoveredToolCache.PrepareForNewTurn( - _availableTools, _baseToolCount, _config.Tuning.DiscoveredToolRetentionTurns, _config.Tuning.DiscoveredToolMaxCount, _fullRegistry); @@ -2780,10 +2777,11 @@ private string CurrentMemoryBoundary() private IReadOnlyList ResolveExposedToolsForCurrentTurn() { - if (_toolAccessPolicy is null || _fullRegistry is null || _availableTools.Count == 0) - return _availableTools; + var availableTools = _discoveredToolCache.AvailableTools; + if (_toolAccessPolicy is null || _fullRegistry is null || availableTools.Count == 0) + return availableTools; - return _toolAccessPolicy.FilterExposedTools(_availableTools, _fullRegistry, _currentTrustContext); + return _toolAccessPolicy.FilterExposedTools(availableTools, _fullRegistry, _currentTrustContext); } /// @@ -3145,7 +3143,7 @@ private void HandleRoutedSkillExecutionCompleted(RoutedSkillExecutionCompleted m private bool HasFileReadGranted() { - foreach (var tool in _availableTools) + foreach (var tool in _discoveredToolCache.AvailableTools) { if (tool is AIFunction fn && string.Equals(fn.Name, FileReadTool.ToolName, StringComparison.Ordinal)) return true; @@ -3200,20 +3198,11 @@ private bool TryActivateDiscoveredTool(string toolName) _discoveredToolCache.Remember(canonicalName, tool, _config.Tuning.DiscoveredToolRetentionTurns, _config.Tuning.DiscoveredToolMaxCount); - AddAvailableToolIfMissing(canonicalName, tool.ToAITool()); + if (_discoveredToolCache.AddIfMissing(tool.ToAITool())) + _log.Info("Dynamically loaded tool '{ToolName}' into session", canonicalName); return true; } - private void AddAvailableToolIfMissing(string toolName, AITool aiTool) - { - if (_availableTools.Any(existing => - existing is AIFunction ef && aiTool is AIFunction nf && ef.Name == nf.Name)) - return; - - _availableTools.Add(aiTool); - _log.Info("Dynamically loaded tool '{ToolName}' into session", toolName); - } - private SessionSnapshot BuildSnapshot() { return _state.ToSnapshot() with From 296e0eb5f47154db9ca05781948c376943f63dec Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Thu, 25 Jun 2026 13:25:32 -0500 Subject: [PATCH 5/5] refactor(sessions): address code-review findings Cleanup follow-up to the LlmSessionActor decomposition (no behavior change): 1. Repoint three stale comments that named the deleted _pendingModelInputMediaReferences field at the now-correct ModelInputMediaBuffer / DrainSnapshot mechanism (SessionState.BuildNudgeMessage + two regression-test comments). 2. ModelInputMediaBuffer.DrainSnapshot now hands off its backing list and adopts a fresh one (swap) instead of ToArray-copying; BuildNudgeMessage's existing defensive copy remains the single copy, restoring the pre-refactor 1-copy cost. 3. InFlightTurnDedup collapses its duplicated reminder/job trios into one generic InFlightSet, so the reserve-ignores-empty / complete-doesn't guard asymmetry lives in one place and can't drift. 4. ReminderDeliveryResult.ReminderDeliveryKey is now ReminderId (was string), closing the last stringly-typed seam on the reminder-delivery path: the four binding actors stop unwrapping with .Value, ReminderExecutionActor compares typed keys, and tests drop the .Value.Value double-unwrap. The message is INoSerializationVerificationNeeded (transient, never journaled), so this has zero persistence/wire impact. --- .../Contracts/SessionBindingContractTests.cs | 8 +-- .../Reminders/ReminderManagerActorTests.cs | 6 +- .../Sessions/LlmSessionImageDeliveryTests.cs | 13 +++-- .../Sessions/SessionStateTests.cs | 12 ++-- .../Reminders/ReminderExecutionActor.cs | 16 +++--- .../Reminders/ReminderProtocol.cs | 2 +- .../Sessions/Handlers/InFlightTurnDedup.cs | 55 ++++++++++--------- .../Handlers/ModelInputMediaBuffer.cs | 15 +++-- src/Netclaw.Actors/Sessions/SessionState.cs | 10 ++-- .../DiscordSessionBindingActor.cs | 4 +- .../MattermostSessionBindingActor.cs | 4 +- .../SlackThreadBindingActor.cs | 4 +- .../Gateway/SignalRSessionActor.cs | 4 +- 13 files changed, 80 insertions(+), 73 deletions(-) diff --git a/src/Netclaw.Actors.Tests/Channels/Contracts/SessionBindingContractTests.cs b/src/Netclaw.Actors.Tests/Channels/Contracts/SessionBindingContractTests.cs index 36cd420a6..5698d0657 100644 --- a/src/Netclaw.Actors.Tests/Channels/Contracts/SessionBindingContractTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/Contracts/SessionBindingContractTests.cs @@ -286,7 +286,7 @@ public async Task Reminder_delivery_reports_success_when_post_succeeds() var result = await observer.ExpectMsgAsync( TimeSpan.FromSeconds(5), cancellationToken: ct); - Assert.Equal(reminderKey, result.ReminderDeliveryKey); + Assert.Equal(new ReminderId(reminderKey), result.ReminderDeliveryKey); Assert.Equal(ExpectedChannelType, result.ChannelType); Assert.True(result.Delivered); } @@ -316,7 +316,7 @@ public async Task Reminder_delivery_reports_failure_when_post_throws() var result = await observer.ExpectMsgAsync( TimeSpan.FromSeconds(5), cancellationToken: ct); - Assert.Equal(reminderKey, result.ReminderDeliveryKey); + Assert.Equal(new ReminderId(reminderKey), result.ReminderDeliveryKey); Assert.Equal(ExpectedChannelType, result.ChannelType); Assert.False(result.Delivered); @@ -355,11 +355,11 @@ public async Task Concurrent_reminders_to_same_session_each_get_their_own_result var resultA = await observerA.ExpectMsgAsync( TimeSpan.FromSeconds(5), cancellationToken: ct); - Assert.Equal(keyA, resultA.ReminderDeliveryKey); + Assert.Equal(new ReminderId(keyA), resultA.ReminderDeliveryKey); var resultB = await observerB.ExpectMsgAsync( TimeSpan.FromSeconds(5), cancellationToken: ct); - Assert.Equal(keyB, resultB.ReminderDeliveryKey); + Assert.Equal(new ReminderId(keyB), resultB.ReminderDeliveryKey); } // Regression for the misleading-fallback bug: when the real content post diff --git a/src/Netclaw.Actors.Tests/Reminders/ReminderManagerActorTests.cs b/src/Netclaw.Actors.Tests/Reminders/ReminderManagerActorTests.cs index 0369c2da9..b10cc9d86 100644 --- a/src/Netclaw.Actors.Tests/Reminders/ReminderManagerActorTests.cs +++ b/src/Netclaw.Actors.Tests/Reminders/ReminderManagerActorTests.cs @@ -659,7 +659,7 @@ public async Task CurrentSession_delivery_required_fails_fast_on_explicit_delive // Channel reports the post failed — execution must report failure // (so Akka.Reminders redelivers) without acking the envelope. delivered.Source.DeliveryObserver!.Tell(new ReminderDeliveryResult( - delivered.Source.ReminderId!.Value.Value, + delivered.Source.ReminderId!.Value, ChannelType.Slack, Delivered: false, FailureReason: "channel API down")); @@ -740,7 +740,7 @@ public async Task CurrentSession_delivery_required_succeeds_when_delivery_is_obs Assert.NotNull(delivered.Source.DeliveryObserver); delivered.Source.DeliveryObserver!.Tell(new ReminderDeliveryResult( - delivered.Source.ReminderId!.Value.Value, + delivered.Source.ReminderId!.Value, ChannelType.Slack, Delivered: true, ObservedAtMs: TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds())); @@ -1024,7 +1024,7 @@ public async Task CurrentSession_discord_delivery_required_succeeds_when_deliver Assert.NotNull(delivered.Source.DeliveryObserver); delivered.Source.DeliveryObserver!.Tell(new ReminderDeliveryResult( - delivered.Source.ReminderId!.Value.Value, + delivered.Source.ReminderId!.Value, ChannelType.Discord, Delivered: true, ObservedAtMs: TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds())); diff --git a/src/Netclaw.Actors.Tests/Sessions/LlmSessionImageDeliveryTests.cs b/src/Netclaw.Actors.Tests/Sessions/LlmSessionImageDeliveryTests.cs index 47e3004ec..c669ce689 100644 --- a/src/Netclaw.Actors.Tests/Sessions/LlmSessionImageDeliveryTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/LlmSessionImageDeliveryTests.cs @@ -27,12 +27,13 @@ namespace Netclaw.Actors.Tests.Sessions; /// . /// /// Regression: the streaming completion path -/// (LlmSessionActor.ApplyToolCallRecorded) handed its mutable -/// _pendingModelInputMediaReferences accumulator to the media nudge and -/// then Clear()ed that same instance. Because the nudge aliased the list -/// instead of copying it, the image reference was wiped before the follow-up -/// LLM call hydrated it — the model was told "Image loaded" but never received -/// the bytes and hallucinated. This test drives a real streaming tool call +/// (LlmSessionActor.ApplyToolCallRecorded) drains the +/// ModelInputMediaBuffer accumulator into the media nudge and then reuses +/// that buffer for the next batch. If the nudge aliased the list instead of +/// copying it (see SessionState.BuildNudgeMessage), the image reference +/// would be wiped before the follow-up LLM call hydrated it — the model would be +/// told "Image loaded" but never receive the bytes and hallucinate. This test +/// drives a real streaming tool call /// through the actor and asserts the image bytes reach the chat client on the /// second call; it fails (no DataContent on call 2) without the defensive /// snapshot in SessionState. See GitHub #1264. diff --git a/src/Netclaw.Actors.Tests/Sessions/SessionStateTests.cs b/src/Netclaw.Actors.Tests/Sessions/SessionStateTests.cs index b7357a12a..02f282b82 100644 --- a/src/Netclaw.Actors.Tests/Sessions/SessionStateTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/SessionStateTests.cs @@ -224,12 +224,12 @@ public void AddSystemNudge_can_carry_media_without_becoming_last_user_message() [Fact] public void AddSystemNudge_snapshots_media_so_caller_clear_cannot_empty_it() { - // Regression: LlmSessionActor hands its mutable - // _pendingModelInputMediaReferences accumulator to AddSystemNudge and then - // Clear()s it. Without a defensive snapshot the nudge aliased that list, so - // the Clear() wiped the tool-loaded image before the next LLM call hydrated - // it — the model was told "Image loaded" but never saw the bytes and - // hallucinated. The nudge must retain its own copy. + // Regression: LlmSessionActor hands a caller-owned media accumulator + // (ModelInputMediaBuffer) to AddSystemNudge and then reuses/empties it. + // Without a defensive snapshot the nudge would alias that list, so the + // caller's reuse wiped the tool-loaded image before the next LLM call + // hydrated it — the model was told "Image loaded" but never saw the bytes + // and hallucinated. The nudge must retain its own copy. var media = new SerializableMediaReference { RelativePath = "image.png", diff --git a/src/Netclaw.Actors/Reminders/ReminderExecutionActor.cs b/src/Netclaw.Actors/Reminders/ReminderExecutionActor.cs index a9b3c6507..ded4e1170 100644 --- a/src/Netclaw.Actors/Reminders/ReminderExecutionActor.cs +++ b/src/Netclaw.Actors/Reminders/ReminderExecutionActor.cs @@ -52,7 +52,7 @@ internal sealed class ReminderExecutionActor : ReceiveActor private string? _sessionIdValue; private HistoryRecord? _pendingHistory; private bool _awaitingDeliveryResult; - private string? _expectedReminderDeliveryKey; + private ReminderId? _expectedReminderDeliveryKey; private ICancelable? _deliveryTimeoutCancelable; private bool RoutesBackToOriginSession => _definition.Delivery.Kind == DeliveryKind.CurrentSession; @@ -265,7 +265,7 @@ private async Task InitializeCurrentSessionAsync() // the ReminderDeliveryResult message it is waiting // for. Arm state + a backstop timer and return; the // result (or timeout) is handled as a normal message. - BeginAwaitingDeliveryResult(reminderDeliveryKey); + BeginAwaitingDeliveryResult(new ReminderId(reminderDeliveryKey)); break; } @@ -310,7 +310,7 @@ private async Task InitializeCurrentSessionAsync() /// processing and can handle the the /// binding actor tells it (or the ). /// - private void BeginAwaitingDeliveryResult(string reminderDeliveryKey) + private void BeginAwaitingDeliveryResult(ReminderId reminderDeliveryKey) { _awaitingDeliveryResult = true; _expectedReminderDeliveryKey = reminderDeliveryKey; @@ -323,7 +323,7 @@ private void BeginAwaitingDeliveryResult(string reminderDeliveryKey) private void HandleDeliveryResult(ReminderDeliveryResult result) { - if (!_awaitingDeliveryResult || _expectedReminderDeliveryKey is null) + if (!_awaitingDeliveryResult || _expectedReminderDeliveryKey is not { } expectedKey) return; // Correlate on the delivery key alone. The result was told point-to- @@ -332,7 +332,7 @@ private void HandleDeliveryResult(ReminderDeliveryResult result) // ChannelType too would only fail closed (e.g. a cold SignalR actor // reports its default Tui rather than the origin SignalR), silently // dropping a valid result and stalling on the backstop. - if (!string.Equals(result.ReminderDeliveryKey, _expectedReminderDeliveryKey, StringComparison.Ordinal)) + if (result.ReminderDeliveryKey != expectedKey) return; _awaitingDeliveryResult = false; @@ -372,10 +372,10 @@ private void HandleDeliveryResult(ReminderDeliveryResult result) private void HandleDeliveryBackstopTimeout(DeliveryBackstopTimeout msg) { - if (!_awaitingDeliveryResult || _expectedReminderDeliveryKey is null) + if (!_awaitingDeliveryResult || _expectedReminderDeliveryKey is not { } expectedKey) return; - if (!string.Equals(msg.ReminderDeliveryKey, _expectedReminderDeliveryKey, StringComparison.Ordinal)) + if (msg.ReminderDeliveryKey != expectedKey) return; _awaitingDeliveryResult = false; @@ -620,5 +620,5 @@ private sealed record ExecutionOutput(SessionOutput Output) : INoSerializationVe /// arrives within (e.g. the binding /// actor crashed mid-turn). /// - private sealed record DeliveryBackstopTimeout(string ReminderDeliveryKey) : INoSerializationVerificationNeeded; + private sealed record DeliveryBackstopTimeout(ReminderId ReminderDeliveryKey) : INoSerializationVerificationNeeded; } diff --git a/src/Netclaw.Actors/Reminders/ReminderProtocol.cs b/src/Netclaw.Actors/Reminders/ReminderProtocol.cs index 1ac3b7db6..5700a19f6 100644 --- a/src/Netclaw.Actors/Reminders/ReminderProtocol.cs +++ b/src/Netclaw.Actors/Reminders/ReminderProtocol.cs @@ -380,7 +380,7 @@ public sealed record GetReminderResponse(ReminderInfo? Reminder) : IReminderResp /// Optional timestamp when the outbound delivery outcome was observed. /// public sealed record ReminderDeliveryResult( - string ReminderDeliveryKey, + ReminderId ReminderDeliveryKey, Channels.ChannelType ChannelType, bool Delivered, string? FailureReason = null, diff --git a/src/Netclaw.Actors/Sessions/Handlers/InFlightTurnDedup.cs b/src/Netclaw.Actors/Sessions/Handlers/InFlightTurnDedup.cs index 3e36e8a15..23dceefd8 100644 --- a/src/Netclaw.Actors/Sessions/Handlers/InFlightTurnDedup.cs +++ b/src/Netclaw.Actors/Sessions/Handlers/InFlightTurnDedup.cs @@ -19,36 +19,37 @@ namespace Netclaw.Actors.Sessions.Handlers; /// internal sealed class InFlightTurnDedup { - private readonly HashSet _reminders = []; - private readonly HashSet _backgroundJobs = []; - - public bool IsReminderInFlight(ReminderId? reminderId) => - reminderId is { } id && _reminders.Contains(id); - - public void ReserveReminder(ReminderId? reminderId) + // Both ledgers share one guard implementation so the reserve/complete/lookup + // semantics (notably: reserve ignores empty ids, complete/lookup don't) can't + // drift between the reminder and background-job paths. + private readonly InFlightSet _reminders = new(id => id.Value); + private readonly InFlightSet _backgroundJobs = new(id => id.Value); + + public bool IsReminderInFlight(ReminderId? reminderId) => _reminders.Contains(reminderId); + public void ReserveReminder(ReminderId? reminderId) => _reminders.Reserve(reminderId); + public void CompleteReminder(ReminderId? reminderId) => _reminders.Remove(reminderId); + + public bool IsBackgroundJobInFlight(BackgroundJobId? backgroundJobId) => _backgroundJobs.Contains(backgroundJobId); + public void ReserveBackgroundJob(BackgroundJobId? backgroundJobId) => _backgroundJobs.Reserve(backgroundJobId); + public void CompleteBackgroundJob(BackgroundJobId? backgroundJobId) => _backgroundJobs.Remove(backgroundJobId); + + private sealed class InFlightSet(Func valueOf) + where T : struct { - if (reminderId is { } id && !string.IsNullOrEmpty(id.Value)) - _reminders.Add(id); - } + private readonly HashSet _ids = []; - public void CompleteReminder(ReminderId? reminderId) - { - if (reminderId is { } id) - _reminders.Remove(id); - } + public bool Contains(T? id) => id is { } value && _ids.Contains(value); - public bool IsBackgroundJobInFlight(BackgroundJobId? backgroundJobId) => - backgroundJobId is { } id && _backgroundJobs.Contains(id); + public void Reserve(T? id) + { + if (id is { } value && !string.IsNullOrEmpty(valueOf(value))) + _ids.Add(value); + } - public void ReserveBackgroundJob(BackgroundJobId? backgroundJobId) - { - if (backgroundJobId is { } id && !string.IsNullOrEmpty(id.Value)) - _backgroundJobs.Add(id); - } - - public void CompleteBackgroundJob(BackgroundJobId? backgroundJobId) - { - if (backgroundJobId is { } id) - _backgroundJobs.Remove(id); + public void Remove(T? id) + { + if (id is { } value) + _ids.Remove(value); + } } } diff --git a/src/Netclaw.Actors/Sessions/Handlers/ModelInputMediaBuffer.cs b/src/Netclaw.Actors/Sessions/Handlers/ModelInputMediaBuffer.cs index 36d6e5240..a2acd8525 100644 --- a/src/Netclaw.Actors/Sessions/Handlers/ModelInputMediaBuffer.cs +++ b/src/Netclaw.Actors/Sessions/Handlers/ModelInputMediaBuffer.cs @@ -17,19 +17,24 @@ namespace Netclaw.Actors.Sessions.Handlers; /// internal sealed class ModelInputMediaBuffer { - private readonly List _pending = []; + private List _pending = []; public void Add(IEnumerable references) => _pending.AddRange(references); - /// Returns the buffered references and clears the buffer. + /// + /// Hands off the buffered references and resets the buffer to empty. The + /// existing list is returned by reference and the buffer adopts a fresh one, + /// so no copy is made here — the consumer (AddSystemNudge / BuildNudgeMessage) + /// makes the single defensive copy the immutable persistence type needs. + /// public IReadOnlyList DrainSnapshot() { if (_pending.Count == 0) return []; - var snapshot = _pending.ToArray(); - _pending.Clear(); - return snapshot; + var drained = _pending; + _pending = []; + return drained; } public void Clear() => _pending.Clear(); diff --git a/src/Netclaw.Actors/Sessions/SessionState.cs b/src/Netclaw.Actors/Sessions/SessionState.cs index 69dff1336..8e5a0e983 100644 --- a/src/Netclaw.Actors/Sessions/SessionState.cs +++ b/src/Netclaw.Actors/Sessions/SessionState.cs @@ -337,12 +337,12 @@ private static SerializableChatMessage BuildNudgeMessage( Role = ChatRole.User, Content = $"{SystemNudgePrefix} {nudge}]", // Snapshot, never alias. The model-input media nudge is built from - // LlmSessionActor._pendingModelInputMediaReferences, a mutable - // accumulator the actor Clear()s immediately after handing it off. + // the caller's media accumulator (ModelInputMediaBuffer.DrainSnapshot), + // which reuses/empties its backing list across batches. // SerializableChatMessage is an immutable persistence type that must - // own its media list — without this copy the subsequent Clear() - // empties the nudge's attachments before the next LLM call hydrates - // them, so a tool-loaded image silently never reaches the model. + // own its media list — without this copy the caller's reuse could + // empty the nudge's attachments before the next LLM call hydrates + // them, so a tool-loaded image would silently never reach the model. MediaReferences = [.. mediaReferences] } : new() { Role = ChatRole.User, Content = $"{SystemNudgePrefix} {nudge}]" }; diff --git a/src/Netclaw.Channels.Discord/DiscordSessionBindingActor.cs b/src/Netclaw.Channels.Discord/DiscordSessionBindingActor.cs index 24fd8f321..918bd42c0 100644 --- a/src/Netclaw.Channels.Discord/DiscordSessionBindingActor.cs +++ b/src/Netclaw.Channels.Discord/DiscordSessionBindingActor.cs @@ -1200,7 +1200,7 @@ private async Task HandleOutputReceivedAsync(OutputReceived msg) && _reminderDeliveryObservers.Remove(sourceReminderKey, out var reminderObserver)) { reminderObserver.Tell(new ReminderDeliveryResult( - sourceReminderKey.Value, + sourceReminderKey, ChannelType.Discord, Delivered: _deliveredThisTurn, FailureReason: _deliveredThisTurn ? null : "Discord post did not succeed", @@ -1439,7 +1439,7 @@ private void FailPendingReminderDeliveries(string reason) return; foreach (var (key, observer) in _reminderDeliveryObservers) - observer.Tell(new ReminderDeliveryResult(key.Value, ChannelType.Discord, Delivered: false, FailureReason: reason)); + observer.Tell(new ReminderDeliveryResult(key, ChannelType.Discord, Delivered: false, FailureReason: reason)); _reminderDeliveryObservers.Clear(); } diff --git a/src/Netclaw.Channels.Mattermost/MattermostSessionBindingActor.cs b/src/Netclaw.Channels.Mattermost/MattermostSessionBindingActor.cs index 120cf2ae1..d372126e8 100644 --- a/src/Netclaw.Channels.Mattermost/MattermostSessionBindingActor.cs +++ b/src/Netclaw.Channels.Mattermost/MattermostSessionBindingActor.cs @@ -1171,7 +1171,7 @@ private async Task HandleOutputReceivedAsync(OutputReceived msg) && _reminderDeliveryObservers.Remove(sourceReminderKey, out var reminderObserver)) { reminderObserver.Tell(new ReminderDeliveryResult( - sourceReminderKey.Value, + sourceReminderKey, ChannelType.Mattermost, Delivered: _deliveredThisTurn, FailureReason: _deliveredThisTurn ? null : "Mattermost post did not succeed", @@ -1379,7 +1379,7 @@ private void FailPendingReminderDeliveries(string reason) return; foreach (var (key, observer) in _reminderDeliveryObservers) - observer.Tell(new ReminderDeliveryResult(key.Value, ChannelType.Mattermost, Delivered: false, FailureReason: reason)); + observer.Tell(new ReminderDeliveryResult(key, ChannelType.Mattermost, Delivered: false, FailureReason: reason)); _reminderDeliveryObservers.Clear(); } diff --git a/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs b/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs index e0e74b60e..c69384f07 100644 --- a/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs +++ b/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs @@ -1119,7 +1119,7 @@ private async Task HandleOutputAsync(ThreadOutput threadOutput) { var delivered = _postedThisTurn || _uploadedFileThisTurn; reminderObserver.Tell(new ReminderDeliveryResult( - sourceReminderKey.Value, + sourceReminderKey, ChannelType.Slack, Delivered: delivered, FailureReason: delivered ? null : "Slack post did not succeed", @@ -1471,7 +1471,7 @@ private void FailPendingReminderDeliveries(string reason) return; foreach (var (key, observer) in _reminderDeliveryObservers) - observer.Tell(new ReminderDeliveryResult(key.Value, ChannelType.Slack, Delivered: false, FailureReason: reason)); + observer.Tell(new ReminderDeliveryResult(key, ChannelType.Slack, Delivered: false, FailureReason: reason)); _reminderDeliveryObservers.Clear(); } diff --git a/src/Netclaw.Daemon/Gateway/SignalRSessionActor.cs b/src/Netclaw.Daemon/Gateway/SignalRSessionActor.cs index 2e6d235b1..2c9a635cc 100644 --- a/src/Netclaw.Daemon/Gateway/SignalRSessionActor.cs +++ b/src/Netclaw.Daemon/Gateway/SignalRSessionActor.cs @@ -321,7 +321,7 @@ private void ReportReminderDeliveryResult(TurnCompleted completed, bool delivere && _reminderDeliveryObservers.Remove(sourceReminderKey, out var observer)) { observer.Tell(new ReminderDeliveryResult( - sourceReminderKey.Value, + sourceReminderKey, _channelType, Delivered: delivered, FailureReason: delivered ? null : "SignalR client did not receive the reply", @@ -341,7 +341,7 @@ private void FailPendingReminderDeliveries(string reason) return; foreach (var (key, observer) in _reminderDeliveryObservers) - observer.Tell(new ReminderDeliveryResult(key.Value, _channelType, Delivered: false, FailureReason: reason)); + observer.Tell(new ReminderDeliveryResult(key, _channelType, Delivered: false, FailureReason: reason)); _reminderDeliveryObservers.Clear(); }