diff --git a/src/Netclaw.Actors.Tests/Sessions/CurrentTurnScopeTests.cs b/src/Netclaw.Actors.Tests/Sessions/CurrentTurnScopeTests.cs new file mode 100644 index 000000000..54c0febd1 --- /dev/null +++ b/src/Netclaw.Actors.Tests/Sessions/CurrentTurnScopeTests.cs @@ -0,0 +1,100 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using Netclaw.Actors.Channels; +using Netclaw.Actors.Protocol; +using Netclaw.Actors.Sessions.Handlers; +using Netclaw.Configuration; +using Xunit; + +namespace Netclaw.Actors.Tests.Sessions; + +/// +/// Characterization tests for the turn correlation-identity derivation that +/// took over from the actor's former +/// BindTurnTelemetry overloads. The fallback chain (source turn id → +/// source message id → generated id) is the only real logic in the scope; these +/// lock it before the actor's ~40 read sites are rewired onto the container. +/// +public sealed class CurrentTurnScopeTests +{ + private static MessageSource Source(string? messageId, TurnId? turnId, ChannelType channelType = ChannelType.Slack) + => new() + { + ChannelType = channelType, + SenderId = new SenderId("U123"), + MessageId = messageId, + TurnId = turnId, + Audience = TrustAudience.Team, + Boundary = TrustBoundary.Team, + Principal = PrincipalClassification.TrustedInternal, + Provenance = new SourceProvenance(TransportAuthenticity.Verified, PayloadTaint.Community) + }; + + [Fact] + public void Bind_from_source_uses_the_source_turn_id_when_present() + { + var scope = new CurrentTurnScope(); + + scope.Bind(Source(messageId: "msg-1", turnId: new TurnId("turn-1"), channelType: ChannelType.Discord)); + + Assert.Equal("turn-1", scope.TurnId?.Value); + Assert.Equal("msg-1", scope.MessageId); + Assert.Equal(ChannelType.Discord, scope.ChannelType); + } + + [Fact] + public void Bind_from_source_falls_back_to_message_id_when_no_turn_id() + { + var scope = new CurrentTurnScope(); + + scope.Bind(Source(messageId: "msg-2", turnId: null)); + + Assert.Equal("msg-2", scope.TurnId?.Value); + Assert.Equal("msg-2", scope.MessageId); + } + + [Fact] + public void Bind_from_source_generates_a_turn_id_when_neither_is_present() + { + var scope = new CurrentTurnScope(); + + scope.Bind(Source(messageId: null, turnId: null)); + + Assert.False(string.IsNullOrWhiteSpace(scope.TurnId?.Value)); + Assert.Null(scope.MessageId); + } + + [Fact] + public void Bind_from_null_source_still_yields_a_generated_turn_id() + { + var scope = new CurrentTurnScope(); + + scope.Bind((MessageSource?)null); + + Assert.False(string.IsNullOrWhiteSpace(scope.TurnId?.Value)); + Assert.Null(scope.MessageId); + Assert.Null(scope.ChannelType); + } + + [Fact] + public void Bind_from_turn_context_takes_id_and_channel_and_clears_message_id() + { + var scope = new CurrentTurnScope(); + // A prior source bind leaves a message id behind; the context re-bind must clear it. + scope.Bind(Source(messageId: "stale", turnId: new TurnId("old"))); + + var context = TurnContext.FromMessageSource( + new SessionId("C1/1"), + new TurnId("turn-ctx"), + Source(messageId: "ignored", turnId: new TurnId("ignored"), channelType: ChannelType.Mattermost)); + + scope.Bind(context); + + Assert.Equal("turn-ctx", scope.TurnId?.Value); + Assert.Equal(ChannelType.Mattermost, scope.ChannelType); + Assert.Null(scope.MessageId); + } +} diff --git a/src/Netclaw.Actors/Sessions/Handlers/CurrentTurnScope.cs b/src/Netclaw.Actors/Sessions/Handlers/CurrentTurnScope.cs new file mode 100644 index 000000000..a2b0536ce --- /dev/null +++ b/src/Netclaw.Actors/Sessions/Handlers/CurrentTurnScope.cs @@ -0,0 +1,69 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using Netclaw.Actors.Channels; +using Netclaw.Configuration; + +namespace Netclaw.Actors.Sessions.Handlers; + +/// +/// Owns the transient "what turn is active and where did it come from" state: +/// the inbound source, the derived turn/trust context, the turn's recalled +/// memories, and the diagnostic correlation identity (turn/message/channel). +/// Populated at turn start and re-bound on approval re-drive; the actor reads +/// these to build persisted event records, authorize tool exposure, and enrich +/// turn-scoped logs. +/// +/// The correlation identity (// +/// ) is mutated only through +/// and so the three stay in lockstep. The +/// remaining fields are overwritten each turn; only +/// is explicitly cleared at a turn boundary (the actor nulls it directly). +/// +internal sealed class CurrentTurnScope +{ + /// Provenance of the active turn (channel, sender, reminder/job id). + public MessageSource? Source { get; set; } + + /// Trust/boundary/audience context derived from . + public TurnContext? TurnContext { get; set; } + + /// Effective trust context used to authorize approvals and tool exposure. + public EffectiveTrustContext? TrustContext { get; set; } + + /// Memories recalled for this turn, reused across the tool loop. + public AutomaticRecallResult? Recall { get; set; } + + /// Correlation turn id for telemetry/logging (ephemeral). + public Protocol.TurnId? TurnId { get; private set; } + + /// Inbound message id for crash-context breadcrumbs (ephemeral). + public string? MessageId { get; private set; } + + /// Channel type of the active turn (ephemeral). + public Channels.ChannelType? ChannelType { get; private set; } + + /// + /// Establishes the diagnostic correlation identity for a turn from its + /// inbound source, generating a turn id when the source carries none. + /// + public void Bind(MessageSource? source) + { + MessageId = source?.MessageId; + TurnId = source?.TurnId ?? new Protocol.TurnId(MessageId ?? IdGen.ShortId()); + ChannelType = source?.ChannelType; + } + + /// + /// Re-binds correlation identity from a recovered or parked turn context. + /// No inbound message id is available on this path. + /// + public void Bind(TurnContext context) + { + MessageId = null; + TurnId = context.TurnId; + ChannelType = context.ChannelType; + } +} diff --git a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs index aa34da912..cced4daef 100644 --- a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs +++ b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs @@ -85,10 +85,10 @@ public sealed class LlmSessionActor : ReceivePersistentActor, IWithTimers // 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; + // "What turn is active and where did it come from": source, derived + // turn/trust context, recalled memories, and diagnostic correlation identity. + private readonly CurrentTurnScope _turn = new(); private bool _processingStateActive; - private ApprovalTurnState _approvalTurnState = ApprovalTurnState.None; private readonly ToolRegistry? _fullRegistry; private readonly ToolAccessPolicy? _toolAccessPolicy; private readonly TrustContextDeriver? _trustContextDeriver; @@ -155,13 +155,6 @@ public sealed class LlmSessionActor : ReceivePersistentActor, IWithTimers // budget after). private bool _anyContentStreamed; - // Per-turn diagnostic correlation (ephemeral) - private Protocol.TurnId? _activeTurnId; - private string? _activeMessageId; - private Channels.ChannelType? _activeChannelType; - private AutomaticRecallResult? _activeRecall; - private EffectiveTrustContext? _currentTrustContext; - // Startup context layers: injected on first LLM call, re-injected after compaction private bool _startupContextInjected; @@ -263,7 +256,7 @@ public LlmSessionActor( ApplyTurnRecorded(evt); _pendingToolInteractions.Clear(); _resolvedToolApprovals.Clear(); - ClearApprovalTurnState(); + ClearCurrentTurnContext(); ClearActiveToolBatchTracking(); }); Recover(evt => _state = _state.Apply(evt)); @@ -273,7 +266,7 @@ public LlmSessionActor( _state = _state.Apply(evt); _pendingToolInteractions.Clear(); _resolvedToolApprovals.Clear(); - ClearApprovalTurnState(); + ClearCurrentTurnContext(); ClearActiveToolBatchTracking(); }); Recover(ApplyToolBatchStarted); @@ -838,7 +831,7 @@ private void HandleToolExecutionCompleted(ToolExecutionCompleted msg) EnqueueCheckpointFireAndForget(new MemoryCheckpointRequest( SessionId: _sessionId, - TurnId: _activeTurnId, + TurnId: _turn.TurnId, TriggerType: Memory.CheckpointTriggerType.SubagentFindings, Priority: 80, Payload: SessionMemoryCheckpointFactory.ForSubAgentFinding( @@ -990,7 +983,6 @@ private void HandleToolExecutionCompleted(ToolExecutionCompleted msg) _resolvedToolApprovals.Clear(); TurnLog().Info("turn_tool_execution_complete iteration={Iteration} callCount={CallCount} max={Max} resultCount={ResultCount}", _turnState.ToolIterationCount, _turnState.ToolCallCount, _config.MaxToolIterationsPerTurn, msg.ToolResults.Count); - MarkApprovalRunningAfterRedrive(); FireLlmCall(); } @@ -1025,7 +1017,7 @@ private void HandleDistillationResult(SessionDistillationCompleted msg, bool sto if (msg.Proposals.Count > 0 && _curationActor is not null && CurrentTurnAudience() != TrustAudience.Public && _memoryConfig.Enabled) { - if (ShouldSkipMemoryCurationForThirdPartyAdoptedContext(_currentTurnContext, _currentTurnSource)) + if (ShouldSkipMemoryCurationForThirdPartyAdoptedContext(_turn.TurnContext, _turn.Source)) { TurnLog().Info("memory_curation_skipped third-party adopted-context present; waiting for explicit elevation"); if (stopAfterAcceptedProposalPersistence) @@ -1237,7 +1229,7 @@ private void HandleCompactionWorkCompleted(CompactionWorkCompleted msg) EnqueueCheckpointFireAndForget(new MemoryCheckpointRequest( SessionId: _sessionId, - TurnId: _activeTurnId, + TurnId: _turn.TurnId, TriggerType: Memory.CheckpointTriggerType.CompactionBoundary, Priority: 90, Payload: SessionMemoryCheckpointFactory.ForCompactionBoundary( @@ -1376,7 +1368,7 @@ private void DrainBufferOrReady() } else { - ClearApprovalTurnState(); + ClearCurrentTurnContext(); TransitionTo(SessionPhase.Ready); } @@ -1994,7 +1986,7 @@ await self.Ask( _activeToolExecutionCts = new CancellationTokenSource(); var toolExecutionCt = _activeToolExecutionCts.Token; - _ = SessionToolExecutionPipeline.ExecuteToolsAsync(executor, toolCalls, sessionId, _currentTurnSource, auditLogger, tp, sessionDir, maxInlineToolResultChars, toolExecutionTimeout, self, emitSubAgentOutput, spawnChildActor, + _ = SessionToolExecutionPipeline.ExecuteToolsAsync(executor, toolCalls, sessionId, _turn.Source, auditLogger, tp, sessionDir, maxInlineToolResultChars, toolExecutionTimeout, self, emitSubAgentOutput, spawnChildActor, approvalChannel: _approvalChannel, emitApprovalRequest: request => self.Tell(request), approvalTimeout: Timeout.InfiniteTimeSpan, @@ -2005,7 +1997,7 @@ await self.Ask( modelInputModalities: _model.InputModalities, oneTimeApprovalPreSeed: oneTimeApprovalPreSeed, decisionOverride: decisionOverride, - turnContext: _currentTurnContext, + turnContext: _turn.TurnContext, ct: toolExecutionCt); } @@ -2038,8 +2030,8 @@ private void HandleTextResponse( }, AssistantReply = reply, RecordedAtMs = NowMs(), - SourceReminderId = _currentTurnSource?.ReminderId, - SourceBackgroundJobId = _currentTurnSource?.BackgroundJobId + SourceReminderId = _turn.Source?.ReminderId, + SourceBackgroundJobId = _turn.Source?.BackgroundJobId }; Persist(turnEvent, evt => @@ -2063,11 +2055,11 @@ private void HandleTextResponse( EmitResponseOutputs(lastMessage, usage, includeText: true, includeThinking: true); MaybeSnapshot(); MaybeGenerateTitle(); - _activeRecall = recallResult; + _turn.Recall = recallResult; EnqueueCheckpointFireAndForget(new MemoryCheckpointRequest( SessionId: _sessionId, - TurnId: _activeTurnId, + TurnId: _turn.TurnId, TriggerType: Memory.CheckpointTriggerType.TurnComplete, Priority: 40, Payload: SessionMemoryCheckpointFactory.ForTurnComplete( @@ -2118,7 +2110,7 @@ private void DrainBufferedMessagesOrBecomeReady() return; } - ClearApprovalTurnState(); + ClearCurrentTurnContext(); TransitionTo(SessionPhase.Ready); } @@ -2206,8 +2198,6 @@ private void HandleIncomingUserMessage(SendUserMessage cmd) // API, which would otherwise wedge every subsequent turn. if (_pendingToolInteractions.Count > 0) { - if (_approvalTurnState is WaitingApprovalTurn waiting) - _approvalTurnState = new AbandoningApprovalTurn(waiting.Context, "superseded_by_new_message"); var abandoned = BuildToolBatchAbandonedEvent(); Persist(abandoned, evt => { @@ -2245,14 +2235,16 @@ private void HandleIncomingUserMessage(SendUserMessage cmd) private void ContinueIncomingUserMessage(SendUserMessage cmd) { _deliveryRetry.Clear(); - _currentTurnSource = cmd.Source; + _turn.Source = cmd.Source; BindTurnTelemetry(cmd.Source); - _currentTurnContext = TurnContext.FromMessageSource( + // _turn.TurnId is non-null after Bind (it generates one from the message id + // or IdGen when the source carries none), so reuse it rather than forking a + // second id-generation path here. + _turn.TurnContext = TurnContext.FromMessageSource( _sessionId, - _activeTurnId ?? new Protocol.TurnId(IdGen.ShortId()), + _turn.TurnId!.Value, cmd.Source); - _approvalTurnState = new RunningApprovalTurn(_currentTurnContext); - _currentTrustContext = _trustContextDeriver?.DeriveFromTurnContext(_currentTurnContext); + _turn.TrustContext = _trustContextDeriver?.DeriveFromTurnContext(_turn.TurnContext); PersistAdoptedContextIfNeeded(cmd.Source); // Sessions created from Slack/Discord start without transport-derived @@ -2639,10 +2631,10 @@ private void FireLlmCall(string? recallQuery = null, bool forceNoTools = false) recallQuery, _state, _sessionId, - _currentTurnSource, + _turn.Source, _memoryRecallCoordinator, _memoryConfig.Enabled, - turnContext: _currentTurnContext); + turnContext: _turn.TurnContext); recallSw.Stop(); resolved = _recallManager.ApplyProgressiveRecall(resolved, _log); @@ -2686,7 +2678,7 @@ private void FireLlmCall(string? recallQuery = null, bool forceNoTools = false) // (llama.cpp, vLLM, OpenAI, Ollama, ...) extend the cache prefix // through this content on every subsequent turn instead of // re-tokenizing it from scratch. - _activeRecall = _recallManager.TurnRecallCache; + _turn.Recall = _recallManager.TurnRecallCache; var volatileBlock = SessionMessageAssembler.BuildVolatileContextBlock(new ContextAssemblyInput( State: _state, ContextLayers: _contextLayers, @@ -2697,7 +2689,7 @@ private void FireLlmCall(string? recallQuery = null, bool forceNoTools = false) SessionId: _sessionId, SessionsBasePath: _sessionsBasePath, FileReadGranted: HasFileReadGranted(), - ActiveRecall: _activeRecall, + ActiveRecall: _turn.Recall, Audience: CurrentTurnAudience(), SkillHint: BuildSkillHint())); if (!string.IsNullOrEmpty(volatileBlock)) @@ -2706,7 +2698,7 @@ private void FireLlmCall(string? recallQuery = null, bool forceNoTools = false) } } - _activeRecall = _recallManager.TurnRecallCache; + _turn.Recall = _recallManager.TurnRecallCache; // Build the full message list via the cache-stable assembler. // Static content (persisted prompt, OnceAtStart layers, [session], @@ -2729,7 +2721,7 @@ private void FireLlmCall(string? recallQuery = null, bool forceNoTools = false) SessionId: _sessionId, SessionsBasePath: _sessionsBasePath, FileReadGranted: HasFileReadGranted(), - ActiveRecall: _activeRecall, + ActiveRecall: _turn.Recall, Audience: CurrentTurnAudience(), SkillHint: skillHint, // Canonical names live in history (post-PR follow-up); the @@ -2763,16 +2755,16 @@ private void FireLlmCall(string? recallQuery = null, bool forceNoTools = false) private TrustAudience CurrentTurnAudience() - => _currentTurnContext?.Audience - ?? _currentTurnSource?.Audience + => _turn.TurnContext?.Audience + ?? _turn.Source?.Audience ?? SecurityPolicyDefaults.ResolveAudienceFromSessionId(_sessionId.Value); private string CurrentMemoryAudience() - => (_currentTurnContext?.Audience ?? _currentTurnSource?.Audience ?? TrustAudience.Public).ToWireValue(); + => (_turn.TurnContext?.Audience ?? _turn.Source?.Audience ?? TrustAudience.Public).ToWireValue(); private string CurrentMemoryBoundary() - => _currentTurnContext?.Boundary.Value - ?? _currentTurnSource?.Boundary.Value + => _turn.TurnContext?.Boundary.Value + ?? _turn.Source?.Boundary.Value ?? SecurityPolicyDefaults.ResolveBoundaryFromSessionId(_sessionId.Value, CurrentTurnAudience()).Value; private IReadOnlyList ResolveExposedToolsForCurrentTurn() @@ -2781,7 +2773,7 @@ private IReadOnlyList ResolveExposedToolsForCurrentTurn() if (_toolAccessPolicy is null || _fullRegistry is null || availableTools.Count == 0) return availableTools; - return _toolAccessPolicy.FilterExposedTools(availableTools, _fullRegistry, _currentTrustContext); + return _toolAccessPolicy.FilterExposedTools(availableTools, _fullRegistry, _turn.TrustContext); } /// @@ -2798,7 +2790,7 @@ private bool IsSetWorkingDirectoryAvailable() var registration = _fullRegistry.GetRegistrationByToolName(SetWorkingDirectoryTool.ToolName); return registration is not null - && _toolAccessPolicy.IsToolExposed(registration, _currentTrustContext); + && _toolAccessPolicy.IsToolExposed(registration, _turn.TrustContext); } @@ -2843,7 +2835,7 @@ private bool TryHandleSlashCommand(string userContent, IReadOnlyList @@ -3128,7 +3120,7 @@ private void HandleRoutedSkillExecutionCompleted(RoutedSkillExecutionCompleted m SessionId = _sessionId, TurnNumber = new TurnNumber(_state.TurnCount), Outcome = TurnOutcome.Completed, - SourceReminderId = _currentTurnSource?.ReminderId + SourceReminderId = _turn.Source?.ReminderId }); MaybeSnapshot(); @@ -3186,7 +3178,7 @@ private bool TryActivateDiscoveredTool(string toolName) var registration = _fullRegistry.GetRegistrationByToolName(toolName); if (registration is null) return false; - if (_toolAccessPolicy is not null && !_toolAccessPolicy.IsToolExposed(registration, _currentTrustContext)) + if (_toolAccessPolicy is not null && !_toolAccessPolicy.IsToolExposed(registration, _turn.TrustContext)) return false; var tool = registration.Tool; @@ -3282,11 +3274,11 @@ private void HandleToolInteractionRequestDispatch(ToolInteractionRequestDispatch ToolName = msg.ToolName.Value, Patterns = msg.Patterns, CandidateVerbs = msg.CandidateVerbs, - Audience = _currentTurnContext?.Audience ?? CurrentTurnAudience(), - Boundary = _currentTurnContext?.Boundary ?? _currentTurnSource?.Boundary, - ChannelType = _currentTurnContext?.ChannelType?.ToWireValue() ?? _currentTurnSource?.ChannelType.ToWireValue(), - SupportsInteractiveApproval = _currentTurnContext?.SupportsInteractiveApproval - ?? _currentTurnSource?.ChannelType.SupportsInteractiveApproval(), + Audience = _turn.TurnContext?.Audience ?? CurrentTurnAudience(), + Boundary = _turn.TurnContext?.Boundary ?? _turn.Source?.Boundary, + ChannelType = _turn.TurnContext?.ChannelType?.ToWireValue() ?? _turn.Source?.ChannelType.ToWireValue(), + SupportsInteractiveApproval = _turn.TurnContext?.SupportsInteractiveApproval + ?? _turn.Source?.ChannelType.SupportsInteractiveApproval(), RequesterSenderId = msg.RequesterSenderId, RequesterPrincipal = msg.RequesterPrincipal, HasThirdPartyAdoptedContext = msg.HasThirdPartyAdoptedContext, @@ -3294,7 +3286,7 @@ private void HandleToolInteractionRequestDispatch(ToolInteractionRequestDispatch Cwd = msg.Cwd, OptionKeys = msg.Options.Select(o => o.Key.Value).ToArray(), Candidates = msg.Candidates, - TurnContext = _currentTurnContext?.ToRecord(), + TurnContext = _turn.TurnContext?.ToRecord(), RequestedAtMs = NowMs() }; @@ -3343,8 +3335,10 @@ private void ApplyToolApprovalRequested(ToolApprovalRequested evt, bool persistA evt.Candidates); _resolvedToolApprovals.Remove(evt.CallId); + // Restore the parked turn context so a later re-drive authorizes the + // approval at the audience/boundary the request was originally made under. if (persistApprovalState && turnContext is not null) - RecordWaitingApprovalState(turnContext, evt.CallId, recovered: _phase.Current == SessionPhase.Recovering); + _turn.TurnContext = turnContext; else if (persistApprovalState && restoreFailure is not null) _log.Warning( "Approval request {CallId} could not restore turn context: {Reason}", @@ -3352,37 +3346,7 @@ private void ApplyToolApprovalRequested(ToolApprovalRequested evt, bool persistA restoreFailure); } - private void RecordWaitingApprovalState(TurnContext context, string callId, bool recovered) - { - var pendingCallIds = _approvalTurnState is WaitingApprovalTurn waiting - ? new HashSet(waiting.PendingCallIds, StringComparer.Ordinal) - : new HashSet(StringComparer.Ordinal); - pendingCallIds.Add(callId); - - _currentTurnContext = context; - _approvalTurnState = new WaitingApprovalTurn(context, pendingCallIds, recovered); - } - - private void MarkApprovalRedrive(PendingToolInteraction pending, string callId) - { - if (pending.TurnContext is null) - return; - - _currentTurnContext = pending.TurnContext; - _approvalTurnState = new RedrivingApprovalTurn(pending.TurnContext, callId); - } - - private void MarkApprovalRunningAfterRedrive() - { - if (_approvalTurnState is RedrivingApprovalTurn redriving) - _approvalTurnState = new RunningApprovalTurn(redriving.Context); - } - - private void ClearApprovalTurnState() - { - _approvalTurnState = ApprovalTurnState.None; - _currentTurnContext = null; - } + private void ClearCurrentTurnContext() => _turn.TurnContext = null; private void ApplyToolApprovalResolved(ToolApprovalResolved evt) { @@ -3391,12 +3355,7 @@ private void ApplyToolApprovalResolved(ToolApprovalResolved evt) : ApprovalDecision.Denied; if (_pendingToolInteractions.Remove(evt.CallId, out var pending)) - { _resolvedToolApprovals[evt.CallId] = new ResolvedToolApproval(pending, decision); - - if (_pendingToolInteractions.Count == 0 && pending.TurnContext is not null) - _approvalTurnState = new RunningApprovalTurn(pending.TurnContext); - } } private void ApplyToolBatchAbandoned(ToolBatchAbandoned evt) @@ -3410,7 +3369,7 @@ private void ApplyToolBatchAbandoned(ToolBatchAbandoned evt) }); _pendingToolInteractions.Clear(); _resolvedToolApprovals.Clear(); - ClearApprovalTurnState(); + ClearCurrentTurnContext(); ClearActiveToolBatchTracking(); } @@ -3502,7 +3461,7 @@ private void EmitResponseOutputs( { SessionId = _sessionId, TurnNumber = new TurnNumber(_state.TurnCount), - SourceReminderId = _currentTurnSource?.ReminderId + SourceReminderId = _turn.Source?.ReminderId }); } @@ -4101,7 +4060,7 @@ private ApprovalRedrivePlan BuildApprovalRedrivePlan(SerializableChatMessage ass /// assistant message in history whose tool calls have no later tool result, /// transitions to , and dispatches it /// under the parked turn's persisted trust context. After cold recovery - /// is null, so the persisted trust fields + /// is null, so the persisted trust fields /// are what keep the re-driven call faithful to the original turn. /// private bool RedriveToolBatchForApproval( @@ -4153,10 +4112,9 @@ private bool RedriveToolBatchForApproval( return false; } - _currentTurnContext = turnContext; - _currentTrustContext = _trustContextDeriver?.DeriveFromTurnContext(turnContext); + _turn.TurnContext = turnContext; + _turn.TrustContext = _trustContextDeriver?.DeriveFromTurnContext(turnContext); BindTurnTelemetry(turnContext); - MarkApprovalRedrive(pending, callId); TransitionTo(SessionPhase.Processing); DispatchToolBatch( @@ -4202,13 +4160,13 @@ private ToolBatchAbandoned BuildToolBatchAbandonedEvent(string resultContent) private void FailCurrentTurn(string errorMessage, Exception cause, ErrorCategory category = ErrorCategory.Unknown) { - _inFlightDedup.CompleteReminder(_currentTurnSource?.ReminderId); - _inFlightDedup.CompleteBackgroundJob(_currentTurnSource?.BackgroundJobId); + _inFlightDedup.CompleteReminder(_turn.Source?.ReminderId); + _inFlightDedup.CompleteBackgroundJob(_turn.Source?.BackgroundJobId); CancelAndDisposeToolExecutionCts(); _deliveryRetry.Clear(); _pendingToolInteractions.Clear(); _resolvedToolApprovals.Clear(); - ClearApprovalTurnState(); + ClearCurrentTurnContext(); _state = _state.AddErrorReply(errorMessage); var correlationId = Guid.NewGuid(); @@ -4232,7 +4190,7 @@ private void FailCurrentTurn(string errorMessage, Exception cause, ErrorCategory SessionId = _sessionId, TurnNumber = new TurnNumber(_state.TurnCount), Outcome = TurnOutcome.Failed, - SourceReminderId = _currentTurnSource?.ReminderId + SourceReminderId = _turn.Source?.ReminderId }); DrainBufferedMessagesOrBecomeReady(); @@ -4366,7 +4324,7 @@ private void ProcessToolCallResult(Pipelines.ToolCallResult result) EnqueueCheckpointFireAndForget(new MemoryCheckpointRequest( SessionId: _sessionId, - TurnId: _activeTurnId, + TurnId: _turn.TurnId, TriggerType: Memory.CheckpointTriggerType.SubagentFindings, Priority: 80, Payload: SessionMemoryCheckpointFactory.ForSubAgentFinding( @@ -4531,7 +4489,6 @@ private void CompleteToolBatch(int resultCount) ClearActiveToolBatchTracking(); TurnLog().Info("turn_tool_execution_complete iteration={Iteration} callCount={CallCount} max={Max} resultCount={ResultCount}", _turnState.ToolIterationCount, _turnState.ToolCallCount, _config.MaxToolIterationsPerTurn, resultCount); - MarkApprovalRunningAfterRedrive(); FireLlmCall(); } @@ -4607,45 +4564,38 @@ private enum ApprovalRedriveOutcome private void BindTurnTelemetry(MessageSource? source) { - var sourceMessageId = source?.MessageId; - _activeMessageId = sourceMessageId; - _activeTurnId = source?.TurnId - ?? new Protocol.TurnId(sourceMessageId ?? IdGen.ShortId()); - _activeChannelType = source?.ChannelType; - - CrashContextSnapshot.Update( - _sessionId.Value, - _activeTurnId?.Value, - _activeMessageId, - _activeChannelType?.ToWireValue(), - _timeProvider.GetUtcNow()); + _turn.Bind(source); + PublishCrashContext(); } private void BindTurnTelemetry(TurnContext context) { - _activeMessageId = null; - _activeTurnId = context.TurnId; - _activeChannelType = context.ChannelType; + _turn.Bind(context); + PublishCrashContext(); + } - CrashContextSnapshot.Update( + // Process-wide best-effort breadcrumb so crash handlers can name the + // in-flight turn. Uses actor-owned session id and clock, so it stays here + // rather than on the scope. + private void PublishCrashContext() + => CrashContextSnapshot.Update( _sessionId.Value, - _activeTurnId?.Value, - _activeMessageId, - _activeChannelType?.ToWireValue(), + _turn.TurnId?.Value, + _turn.MessageId, + _turn.ChannelType?.ToWireValue(), _timeProvider.GetUtcNow()); - } private ILoggingAdapter TurnLog() { var log = _log; - if (_activeTurnId is { Value: { Length: > 0 } turnIdValue }) + if (_turn.TurnId is { Value: { Length: > 0 } turnIdValue }) log = log.WithContext("TurnId", turnIdValue); - if (!string.IsNullOrWhiteSpace(_activeMessageId)) - log = log.WithContext("MessageId", _activeMessageId); + if (!string.IsNullOrWhiteSpace(_turn.MessageId)) + log = log.WithContext("MessageId", _turn.MessageId); - if (_activeChannelType is { } act) + if (_turn.ChannelType is { } act) log = log.WithContext("ChannelType", act.ToWireValue()); return log; diff --git a/src/Netclaw.Actors/Sessions/ToolApprovalState.cs b/src/Netclaw.Actors/Sessions/ToolApprovalState.cs index 282680f29..83620b2e4 100644 --- a/src/Netclaw.Actors/Sessions/ToolApprovalState.cs +++ b/src/Netclaw.Actors/Sessions/ToolApprovalState.cs @@ -48,24 +48,6 @@ internal sealed record ToolInteractionRequestDispatch( SessionProtocol.ToolInteractionRequest Request, bool PersistApprovalState) : INoSerializationVerificationNeeded; -internal abstract record ApprovalTurnState : INoSerializationVerificationNeeded -{ - public static ApprovalTurnState None { get; } = new NoActiveApprovalTurn(); -} - -internal sealed record NoActiveApprovalTurn : ApprovalTurnState; - -internal sealed record RunningApprovalTurn(TurnContext Context) : ApprovalTurnState; - -internal sealed record WaitingApprovalTurn( - TurnContext Context, - ISet PendingCallIds, - bool Recovered) : ApprovalTurnState; - -internal sealed record RedrivingApprovalTurn(TurnContext Context, string CallId) : ApprovalTurnState; - -internal sealed record AbandoningApprovalTurn(TurnContext Context, string Reason) : ApprovalTurnState; - internal sealed record ApprovalRedrivePlan( IReadOnlyDictionary>? OneTimeApprovalPreSeed, IReadOnlyDictionary? DecisionOverride);