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..5698d0657 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(); @@ -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); } @@ -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")); @@ -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); @@ -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(); @@ -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 @@ -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..b10cc9d86 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, 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, 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, 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/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/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..02f282b82 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; @@ -223,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", @@ -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 b15d61ddc..196ec93bc 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; @@ -746,7 +747,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.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/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..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; @@ -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 @@ -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/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/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/Handlers/InFlightTurnDedup.cs b/src/Netclaw.Actors/Sessions/Handlers/InFlightTurnDedup.cs new file mode 100644 index 000000000..23dceefd8 --- /dev/null +++ b/src/Netclaw.Actors/Sessions/Handlers/InFlightTurnDedup.cs @@ -0,0 +1,55 @@ +// ----------------------------------------------------------------------- +// +// 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 +{ + // 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 + { + private readonly HashSet _ids = []; + + public bool Contains(T? id) => id is { } value && _ids.Contains(value); + + public void Reserve(T? id) + { + if (id is { } value && !string.IsNullOrEmpty(valueOf(value))) + _ids.Add(value); + } + + 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 new file mode 100644 index 000000000..a2acd8525 --- /dev/null +++ b/src/Netclaw.Actors/Sessions/Handlers/ModelInputMediaBuffer.cs @@ -0,0 +1,41 @@ +// ----------------------------------------------------------------------- +// +// 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 List _pending = []; + + public void Add(IEnumerable references) => _pending.AddRange(references); + + /// + /// 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 drained = _pending; + _pending = []; + return drained; + } + + public void Clear() => _pending.Clear(); +} 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 b302d8f2f..aa34da912 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,16 +74,17 @@ 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); private readonly Dictionary _resolvedToolApprovals = new(StringComparer.Ordinal); // 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; @@ -89,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) @@ -197,7 +200,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!; @@ -251,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 => @@ -338,12 +340,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); @@ -369,9 +369,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; @@ -556,7 +553,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."; @@ -639,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); @@ -1236,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, @@ -2047,13 +2044,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 +2193,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 @@ -2283,7 +2280,6 @@ private void ContinueIncomingUserMessage(SendUserMessage cmd) _turnState.ResetForNewTurn(); _discoveredToolCache.PrepareForNewTurn( - _availableTools, _baseToolCount, _config.Tuning.DiscoveredToolRetentionTurns, _config.Tuning.DiscoveredToolMaxCount, _fullRegistry); @@ -2326,66 +2322,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() @@ -2499,7 +2467,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(); @@ -2809,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); } /// @@ -3133,13 +3102,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 @@ -3174,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; @@ -3229,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 @@ -3303,13 +3263,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()); } } } @@ -3386,7 +3344,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}", @@ -3459,7 +3417,7 @@ private void ApplyToolBatchAbandoned(ToolBatchAbandoned evt) private void ClearActiveToolBatchTracking() { _activeToolBatch.Clear(); - _pendingModelInputMediaReferences.Clear(); + _mediaBuffer.Clear(); } private void MaybeSnapshot() @@ -4244,8 +4202,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(); @@ -4520,8 +4478,7 @@ private void TryCompleteStreamedToolBatch() private void CompleteToolBatch(int resultCount) { - AddModelInputMediaNudge(_pendingModelInputMediaReferences); - _pendingModelInputMediaReferences.Clear(); + AddModelInputMediaNudge(_mediaBuffer.DrainSnapshot()); var budgetStatus = _turnState.RecordToolCompletion(resultCount, _config.MaxToolIterationsPerTurn); @@ -4699,7 +4656,7 @@ private void RequestRestartDrain() _restartDrainRequested = true; _restartDrainReplyTo = Sender; - if (_currentPhase == SessionPhase.Ready) + if (_phase.Current == SessionPhase.Ready) TransitionTo(SessionPhase.Passivating); } 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..8e5a0e983 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); @@ -336,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 7b04a87e9..918bd42c0 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, ChannelType.Discord, Delivered: _deliveredThisTurn, FailureReason: _deliveredThisTurn ? null : "Discord post did not succeed", diff --git a/src/Netclaw.Channels.Mattermost/MattermostSessionBindingActor.cs b/src/Netclaw.Channels.Mattermost/MattermostSessionBindingActor.cs index 4e91a3348..d372126e8 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, ChannelType.Mattermost, Delivered: _deliveredThisTurn, FailureReason: _deliveredThisTurn ? null : "Mattermost post did not succeed", diff --git a/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs b/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs index 8e5943be4..c69384f07 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, ChannelType.Slack, Delivered: delivered, FailureReason: delivered ? null : "Slack post did not succeed", diff --git a/src/Netclaw.Daemon/Gateway/SignalRSessionActor.cs b/src/Netclaw.Daemon/Gateway/SignalRSessionActor.cs index a5bfaf4d8..2c9a635cc 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, _channelType, Delivered: delivered, FailureReason: delivered ? null : "SignalR client did not receive the reply",