diff --git a/feeds/skills/.system/files/netclaw-operations/SKILL.md b/feeds/skills/.system/files/netclaw-operations/SKILL.md index 352cbe69f..f5a57c96b 100644 --- a/feeds/skills/.system/files/netclaw-operations/SKILL.md +++ b/feeds/skills/.system/files/netclaw-operations/SKILL.md @@ -3,7 +3,7 @@ name: netclaw-operations description: "REQUIRED when the user asks about scheduling, reminders, cron jobs, timers, background jobs, diagnostics, troubleshooting, MCP tools, daemon health, identity updates, or Netclaw capabilities and self-maintenance." metadata: author: netclaw - version: "2.21.0" + version: "2.22.0" --- # Netclaw Operations diff --git a/feeds/skills/.system/files/netclaw-operations/references/diagnostics.md b/feeds/skills/.system/files/netclaw-operations/references/diagnostics.md index d6ff9d05a..9343dedb6 100644 --- a/feeds/skills/.system/files/netclaw-operations/references/diagnostics.md +++ b/feeds/skills/.system/files/netclaw-operations/references/diagnostics.md @@ -14,36 +14,55 @@ When something seems wrong with Netclaw itself: 3. Check daemon logs at `~/.netclaw/logs/daemon-{yyyy-MM-dd}.log` 4. Check session logs at `~/.netclaw/logs/sessions/{sanitized-session-id}/session.log` -Log split: - -- Daemon-global diagnostics stay in the daemon log (rolled daily, capped - at 10 MB per file). -- Session-owned diagnostics and session output audit trails append to - `~/.netclaw/logs/sessions/{sanitized-session-id}/session.log` — - one file per session, no rotation today (see netclaw-dev/netclaw#919). -- Session log directories use the sanitized session ID (`/`, `.`, spaces, - etc. replaced with `_`). Sub-agent diagnostics roll up into the parent - session's `session.log`; you will not find a separate file for a - sub-agent run. +Log split — one stream, partitioned locally by session: + +- A log line that carries a session id (an actor's `WithContext("SessionId", …)`, + a `{SessionId}` message field, or a `SessionId` logging scope) is written to that + session's `session.log` and **not** to `daemon.log`. The partition is by session + id — nothing is duplicated locally. +- `daemon.log` holds only sessionless, daemon-wide lines: startup/config, session + start/stop, and operational **alerts** (e.g. the `provider.unreachable` / + `provider.failover` alert raised when an inference provider goes down — surfaced + here, and to webhooks, by the notification sink). Note the *per-call* failover/retry + log lines emitted while serving a specific session carry that session's id, so they + partition into its `session.log`; the daemon-wide outage signal is the alert in + `daemon.log`. Rolled daily, capped at 10 MB per file. +- The **full** stream (daemon and session lines alike) is also exported to OTEL/Seq + with the session id as an attribute; do the global slicing/distilling on the OTEL + receiver side. +- Session log directories use the sanitized session ID (`/`, `.`, spaces, etc. + replaced with `_`). Each **sub-agent run** writes to its **own** `session.log`, + keyed by its sub-session id (`{parentId}/subagent/{name}/{runId}`, sanitized) — + so a sub-agent's detail stays out of the parent's log and you review it in that + run's own file. The parent's `session.log` keeps the spawn breadcrumbs (requested + → spawned → completed/failed) as the pointer to each run. In OTEL the sub-agent + lines still carry the parent `SessionId` (so they group under the parent) plus + `SubSessionId` (so they slice by run). No rotation today + (see netclaw-dev/netclaw#919). What to expect inside `session.log`: -- A single chronological timeline. One actor (`SessionLogActor`) is the - only writer per file, so audit lines and diagnostic lines interleave in - wall-clock order — useful for reading "what happened, in order" without - cross-referencing two files. -- Two line shapes: session output audit lines (`User:`, `Assistant:`, - `Thinking:`, `Tool call:`, `Tool result:`, `Usage:`, `Turn N completed`, - etc.) and `Diagnostic:` lines from MEL providers (LLM client, HTTP, - retry middleware) emitted under a session diagnostics scope. -- Best-effort observability. Individual lines may be dropped on transient - IO errors and logged at Debug level in the daemon log; this is not a - transactional audit trail. Cross-check the daemon log for warnings - if a critical line appears missing. -- Sidecar paths (compaction, title generation, sub-agents, memory - distillation) currently bypass the session diagnostics scope, so their - internal diagnostics may not appear in `session.log` even though their - output audit lines do. Tracked in netclaw-dev/netclaw#920. +- The session's **full local slice** of the log stream, in wall-clock order: the + conversation audit (`User:`, `Assistant:`, `Thinking:`, `Tool call:`, + `Tool result:`, `Usage:`, `Turn N completed`) interleaved with every operational + line scoped to that session — the LLM pipeline, retries, provider failover, tool + and sub-agent activity (spawn requested → child spawned → completed/failed, plus + guard rejections), memory, etc. One actor (`SessionLogActor`) is the only writer + per file. +- Because the partition is by session id, you usually do **not** need to grep — open + the one file for the session and read top to bottom. To correlate across sessions + or globally, use Seq/OTLP (every line is there with `SessionId` as a field). +- The session-log writer's own failure lines are the one exception: they go to + `daemon.log`, never routed back into the file that just failed. +- Writes are split by kind. The **conversation audit** (User/Assistant/Tool/Usage + lines) is flushed **immediately**, so a hard process death cannot drop the audit + tail. The higher-volume **diagnostics** are **batched** (flushed on a ~1s cadence), + so a recent diagnostic line may lag by up to a second. Individual lines may be + dropped on transient IO errors (a warning lands in `daemon.log`). + +What stays in `daemon.log`: only sessionless lines — daemon startup/config, session +lifecycle, and global errors. Debugging one session → read its `session.log`; +debugging a daemon-wide problem → read `daemon.log`. | Symptom | Check | |---------|-------| diff --git a/src/Netclaw.Actors.Tests/Sessions/LlmSessionIntegrationTests.cs b/src/Netclaw.Actors.Tests/Sessions/LlmSessionIntegrationTests.cs index 9da7e4c09..45607f873 100644 --- a/src/Netclaw.Actors.Tests/Sessions/LlmSessionIntegrationTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/LlmSessionIntegrationTests.cs @@ -1678,6 +1678,11 @@ internal sealed class FakeChatClient : IChatClient public List> ReceivedMessages { get; } = []; public List> ReceivedToolNames { get; } = []; + /// The object passed on each call, so tests can + /// assert the session actor threads a SessionScopedChatOptions carrier through + /// for per-session log correlation. + public List ReceivedOptions { get; } = []; + public TimeSpan Delay { get; set; } = TimeSpan.Zero; /// @@ -1778,6 +1783,7 @@ public async Task GetResponseAsync( var messageList = messages.ToList(); ReceivedMessages.Add(messageList); + ReceivedOptions.Add(options); ReceivedToolNames.Add(options?.Tools? .Select(t => t is AIFunction f ? f.Name : t.GetType().Name) .ToList() diff --git a/src/Netclaw.Actors.Tests/Sessions/SidecarDiagnosticsContextTests.cs b/src/Netclaw.Actors.Tests/Sessions/SidecarDiagnosticsContextTests.cs deleted file mode 100644 index 02101e129..000000000 --- a/src/Netclaw.Actors.Tests/Sessions/SidecarDiagnosticsContextTests.cs +++ /dev/null @@ -1,242 +0,0 @@ -// ----------------------------------------------------------------------- -// -// Copyright (C) 2026 - 2026 Petabridge, LLC -// -// ----------------------------------------------------------------------- -using System.Runtime.CompilerServices; -using Akka.Event; -using Akka.Hosting; -using Akka.Hosting.TestKit; -using Microsoft.Extensions.AI; -using Netclaw.Actors.Memory; -using Netclaw.Actors.Protocol; -using Netclaw.Actors.Sessions; -using Netclaw.Actors.Sessions.Pipelines; -using Netclaw.Actors.SubAgents; -using Netclaw.Configuration; -using Xunit; -using AiChatMessage = Microsoft.Extensions.AI.ChatMessage; -using AiChatRole = Microsoft.Extensions.AI.ChatRole; - -namespace Netclaw.Actors.Tests.Sessions; - -/// -/// Verifies that session-owned sidecar paths populate -/// around their IChatClient -/// calls so MEL provider diagnostics emitted during the call route into -/// the per-session log. One test per major sidecar covered by -/// netclaw-dev/netclaw#920. Test contract: a fake IChatClient -/// captures SessionDiagnosticsContext.SessionId at the moment the -/// chat client method is invoked. That is exactly the AsyncLocal value -/// any real provider plugin would see when emitting MEL log lines. -/// -public sealed class SidecarDiagnosticsContextTests : TestKit -{ - protected override void ConfigureAkka(AkkaConfigurationBuilder builder, IServiceProvider provider) - { - } - - [Fact] - public async Task TitleGenerator_populates_session_diagnostics_scope() - { - var sessionId = new SessionId("ch/title-thread"); - var captor = new SessionContextCapturingChatClient(); - var probe = CreateTestProbe(); - - await SessionTitleGenerator.GenerateAsync( - captor, - sessionId, - history: [], - self: probe.Ref, - log: NoLogger.Instance, - timeout: TimeSpan.FromSeconds(5)); - - Assert.Equal(sessionId.Value, captor.CapturedSessionId); - Assert.Null(SessionDiagnosticsContext.SessionId); - } - - [Fact] - public async Task CompactionPipeline_populates_session_diagnostics_scope() - { - var sessionId = new SessionId("ch/compaction-thread"); - var captor = new SessionContextCapturingChatClient(); - var history = new List - { - new() { Role = Netclaw.Actors.Protocol.ChatRole.User, Content = "hello" }, - new() { Role = Netclaw.Actors.Protocol.ChatRole.Assistant, Content = "hi" } - }; - - var observation = await SessionCompactionPipeline.GenerateObservationsAsync( - client: captor, - sessionId: sessionId, - history: history, - systemOffset: 0, - keepStartIndex: 1, - sidecarTimeout: TimeSpan.FromSeconds(5), - log: NoLogger.Instance, - cancellationToken: TestContext.Current.CancellationToken); - - Assert.Equal(sessionId.Value, captor.CapturedSessionId); - Assert.Null(SessionDiagnosticsContext.SessionId); - Assert.NotNull(observation); - } - - [Fact] - public async Task MemoryExtraction_populates_session_diagnostics_scope() - { - var sessionId = new SessionId("ch/memory-extraction-thread"); - var captor = new SessionContextCapturingChatClient(); - var probe = CreateTestProbe(); - - await LlmSessionActor.InvokeMemoryExtractionCoreAsync( - captor, - sessionId, - history: [], - self: probe.Ref, - timeout: TimeSpan.FromSeconds(5)); - - Assert.Equal(sessionId.Value, captor.CapturedSessionId); - Assert.Null(SessionDiagnosticsContext.SessionId); - } - - [Fact] - public async Task SubAgent_InvokeLlm_populates_session_diagnostics_scope() - { - var sessionId = new SessionId("ch/subagent-thread"); - var captor = new SessionContextCapturingChatClient(); - var probe = CreateTestProbe(); - - await SubAgentActor.InvokeLlmAsync( - captor, - messages: [], - options: null, - sessionId: sessionId, - self: probe.Ref, - ct: TestContext.Current.CancellationToken); - - Assert.Equal(sessionId.Value, captor.CapturedSessionId); - Assert.Null(SessionDiagnosticsContext.SessionId); - } - - [Fact] - public async Task SubAgent_InvokeLlm_with_null_sessionId_pushes_null_scope() - { - // Sub-agents that run outside any session legitimately have no - // session id. The Push contract accepts null and the captor should - // see null inside the call. - var captor = new SessionContextCapturingChatClient(); - var probe = CreateTestProbe(); - - await SubAgentActor.InvokeLlmAsync( - captor, - messages: [], - options: null, - sessionId: null, - self: probe.Ref, - ct: TestContext.Current.CancellationToken); - - Assert.Null(captor.CapturedSessionId); - Assert.Null(SessionDiagnosticsContext.SessionId); - } - - [Fact] - public async Task MemoryObserver_RunDistillation_populates_session_diagnostics_scope() - { - var sessionId = new SessionId("ch/distillation-thread"); - var captor = new SessionContextCapturingChatClient(); - var probe = CreateTestProbe(); - - await SessionMemoryObserverActor.RunDistillationAsync( - client: captor, - sessionId: sessionId, - turnCount: 5, - transcript: "user: hello\nassistant: hi", - existingProposals: [], - timeout: TimeSpan.FromSeconds(5), - self: probe.Ref, - runId: 1, - contentVersion: 1, - log: NoLogger.Instance); - - Assert.Equal(sessionId.Value, captor.CapturedSessionId); - Assert.Null(SessionDiagnosticsContext.SessionId); - } - - [Fact] - public async Task MemoryCuration_TryLlmEvaluation_populates_session_diagnostics_scope() - { - var sessionId = new SessionId("ch/curation-thread"); - var captor = new SessionContextCapturingChatClient(); - var operation = new SQLiteMemoryCurationOperation( - Kind: "document", - MemoryClass: "durable_fact", - MemoryId: null, - AnchorCanonicalName: "test-anchor", - AnchorType: "preference", - Title: "Test", - Content: "Test content for diagnostics scope verification.", - AliasesJson: "[]", - FacetsJson: "[]", - SlotsJson: null, - Relations: null, - UpdateSemantics: "merge-document", - Boundary: TrustBoundary.TrustedInstanceValue, - Audience: TrustAudience.Public, - Sensitivity: "normal", - RecallMode: "auto", - Confidence: 0.9, - FreshnessAtMs: TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds(), - ExpiresAtMs: null); - - await MemoryCurationActor.TryLlmEvaluationAsync( - captor, - sessionId, - operation, - candidates: [], - log: NoLogger.Instance); - - Assert.Equal(sessionId.Value, captor.CapturedSessionId); - Assert.Null(SessionDiagnosticsContext.SessionId); - } - - /// - /// IChatClient stub that captures - /// at the moment its methods are invoked. The captured value is the - /// AsyncLocal seen by the chat client — the same value any MEL provider - /// plugin emitting diagnostics inside the call would see. - /// - private sealed class SessionContextCapturingChatClient : IChatClient - { - public string? CapturedSessionId { get; private set; } - - public Task GetResponseAsync( - IEnumerable messages, - ChatOptions? options = null, - CancellationToken cancellationToken = default) - { - CapturedSessionId = SessionDiagnosticsContext.SessionId; - var response = new ChatResponse(new AiChatMessage( - AiChatRole.Assistant, - (IList)[new TextContent("captured")])); - return Task.FromResult(response); - } - - public IAsyncEnumerable GetStreamingResponseAsync( - IEnumerable messages, - ChatOptions? options = null, - CancellationToken cancellationToken = default) - => StreamAsync(cancellationToken); - - private async IAsyncEnumerable StreamAsync( - [EnumeratorCancellation] CancellationToken cancellationToken) - { - CapturedSessionId = SessionDiagnosticsContext.SessionId; - yield return new ChatResponseUpdate(AiChatRole.Assistant, "captured"); - await Task.CompletedTask; - } - - public object? GetService(Type serviceType, object? serviceKey = null) => null; - - public void Dispose() { } - } -} diff --git a/src/Netclaw.Actors.Tests/Sessions/SidecarSessionCorrelationTests.cs b/src/Netclaw.Actors.Tests/Sessions/SidecarSessionCorrelationTests.cs new file mode 100644 index 000000000..4425ed83c --- /dev/null +++ b/src/Netclaw.Actors.Tests/Sessions/SidecarSessionCorrelationTests.cs @@ -0,0 +1,173 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using System.Runtime.CompilerServices; +using Akka.Event; +using Akka.Hosting.TestKit; +using Microsoft.Extensions.AI; +using Netclaw.Actors.Memory; +using Netclaw.Actors.Protocol; +using Netclaw.Actors.Sessions; +using Netclaw.Actors.Sessions.Pipelines; +using Netclaw.Configuration; +using Xunit; +using AiChatMessage = Microsoft.Extensions.AI.ChatMessage; +using AiChatRole = Microsoft.Extensions.AI.ChatRole; + +namespace Netclaw.Actors.Tests.Sessions; + +/// +/// Regression guard (replaces the deleted SidecarDiagnosticsContextTests, netclaw-dev/netclaw#920) +/// for the session-owned sidecar LLM paths — title generation, compaction observation, memory +/// extraction/distillation, and memory curation. After the SessionDiagnosticsContext AsyncLocal +/// was removed, each must carry its owning session id explicitly via +/// on the chat-call's , so the +/// file-logger routes the call's chat-client diagnostics into that session's session.log (and +/// Seq/OTLP correlate them). Test contract: a fake IChatClient captures the ChatOptions it is +/// invoked with; the assertion is that it is a SessionScopedChatOptions naming the session. +/// Without the carrier these calls silently regress to daemon.log, uncorrelated. +/// +public sealed class SidecarSessionCorrelationTests : TestKit +{ + protected override void ConfigureAkka(Akka.Hosting.AkkaConfigurationBuilder builder, IServiceProvider provider) + { + } + + [Fact] + public async Task TitleGenerator_carries_session_scoped_options() + { + var sessionId = new SessionId("ch/title-thread"); + var captor = new OptionsCapturingChatClient(); + + await SessionTitleGenerator.GenerateAsync( + captor, sessionId, history: [], self: CreateTestProbe().Ref, + log: NoLogger.Instance, timeout: TimeSpan.FromSeconds(5)); + + AssertScopedTo(sessionId, captor); + } + + [Fact] + public async Task CompactionObserver_carries_session_scoped_options() + { + var sessionId = new SessionId("ch/compaction-thread"); + var captor = new OptionsCapturingChatClient(); + var history = new List + { + new() { Role = Netclaw.Actors.Protocol.ChatRole.User, Content = "hello" }, + new() { Role = Netclaw.Actors.Protocol.ChatRole.Assistant, Content = "hi" } + }; + + await SessionCompactionPipeline.GenerateObservationsAsync( + client: captor, sessionId: sessionId, history: history, systemOffset: 0, + keepStartIndex: 1, sidecarTimeout: TimeSpan.FromSeconds(5), log: NoLogger.Instance, + cancellationToken: TestContext.Current.CancellationToken); + + AssertScopedTo(sessionId, captor); + } + + [Fact] + public async Task MemoryExtraction_carries_session_scoped_options() + { + var sessionId = new SessionId("ch/memory-extraction-thread"); + var captor = new OptionsCapturingChatClient(); + + await LlmSessionActor.InvokeMemoryExtractionCoreAsync( + captor, sessionId, history: [], self: CreateTestProbe().Ref, timeout: TimeSpan.FromSeconds(5)); + + AssertScopedTo(sessionId, captor); + } + + [Fact] + public async Task MemoryDistillation_carries_session_scoped_options() + { + var sessionId = new SessionId("ch/distillation-thread"); + var captor = new OptionsCapturingChatClient(); + + await SessionMemoryObserverActor.RunDistillationAsync( + client: captor, sessionId: sessionId, turnCount: 5, + transcript: "user: hello\nassistant: hi", existingProposals: [], + timeout: TimeSpan.FromSeconds(5), self: CreateTestProbe().Ref, runId: 1, + contentVersion: 1, log: NoLogger.Instance); + + AssertScopedTo(sessionId, captor); + } + + [Fact] + public async Task MemoryCuration_carries_session_scoped_options() + { + var sessionId = new SessionId("ch/curation-thread"); + var captor = new OptionsCapturingChatClient(); + var operation = new SQLiteMemoryCurationOperation( + Kind: "document", + MemoryClass: "durable_fact", + MemoryId: null, + AnchorCanonicalName: "test-anchor", + AnchorType: "preference", + Title: "Test", + Content: "Test content for session correlation verification.", + AliasesJson: "[]", + FacetsJson: "[]", + SlotsJson: null, + Relations: null, + UpdateSemantics: "merge-document", + Boundary: TrustBoundary.TrustedInstanceValue, + Audience: TrustAudience.Public, + Sensitivity: "normal", + RecallMode: "auto", + Confidence: 0.9, + FreshnessAtMs: TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds(), + ExpiresAtMs: null); + + await MemoryCurationActor.TryLlmEvaluationAsync( + captor, sessionId, operation, candidates: [], log: NoLogger.Instance); + + AssertScopedTo(sessionId, captor); + } + + private static void AssertScopedTo(SessionId sessionId, OptionsCapturingChatClient captor) + { + var scoped = Assert.IsType(captor.CapturedOptions); + Assert.Equal(sessionId.Value, scoped.SessionId); + } + + /// + /// IChatClient stub that records the it is invoked with — the + /// object the chat-client decorators read the session id from to open the routing scope. + /// + private sealed class OptionsCapturingChatClient : IChatClient + { + public ChatOptions? CapturedOptions { get; private set; } + + public Task GetResponseAsync( + IEnumerable messages, + ChatOptions? options = null, + CancellationToken cancellationToken = default) + { + CapturedOptions = options; + return Task.FromResult(new ChatResponse(new AiChatMessage( + AiChatRole.Assistant, (IList)[new TextContent("captured")]))); + } + + public IAsyncEnumerable GetStreamingResponseAsync( + IEnumerable messages, + ChatOptions? options = null, + CancellationToken cancellationToken = default) + { + CapturedOptions = options; + return StreamAsync(cancellationToken); + } + + private static async IAsyncEnumerable StreamAsync( + [EnumeratorCancellation] CancellationToken cancellationToken) + { + yield return new ChatResponseUpdate(AiChatRole.Assistant, "captured"); + await Task.CompletedTask; + } + + public object? GetService(Type serviceType, object? serviceKey = null) => null; + + public void Dispose() { } + } +} diff --git a/src/Netclaw.Actors.Tests/Sessions/SubAgentSpawnIntegrationTests.cs b/src/Netclaw.Actors.Tests/Sessions/SubAgentSpawnIntegrationTests.cs index 196ec93bc..26e8aaab7 100644 --- a/src/Netclaw.Actors.Tests/Sessions/SubAgentSpawnIntegrationTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/SubAgentSpawnIntegrationTests.cs @@ -265,6 +265,15 @@ await sessionManager.Ask(new SendUserMessage m.Role == Microsoft.Extensions.AI.ChatRole.System && (m.Text?.Contains("test assistant with subagent support", StringComparison.Ordinal) ?? false)); Assert.DoesNotContain(subagentCall, m => m.Role == Microsoft.Extensions.AI.ChatRole.User && (m.Text?.Contains("Use a subagent to summarize the file", StringComparison.Ordinal) ?? false)); + + // The session actor must thread a SessionScopedChatOptions carrier so the + // chat-client decorators can correlate LLM diagnostics to the session. The + // sub-agent call carries the *parent* session id (collapsing the scope suffix), + // so both the main turn and the sub-agent's LLM calls correlate to one session. + var mainOptions = Assert.IsType(_clientProvider.Main.ReceivedOptions[^1]); + Assert.Equal(sessionId.Value, mainOptions.SessionId); + var subagentOptions = Assert.IsType(_clientProvider.Compaction.ReceivedOptions[^1]); + Assert.Equal(sessionId.Value, subagentOptions.SessionId); } [Fact] @@ -539,6 +548,11 @@ await sessionManager.Ask(new SendUserMessage && (m.Text?.Contains("Context:", StringComparison.Ordinal) ?? false)); } + // NOTE: routing the spawn lifecycle to session.log is no longer per-path-wired — the + // breadcrumbs log under a SessionId scope and the file-logger partitions them regardless of + // which path (tool-execution or routed-skill) drove the spawn. The producer side is covered + // by SubAgentSpawnObservabilityTests; the routing by RollingFileLoggerPartitionTests. + [Fact] public async Task Reminder_sourced_slash_command_routes_like_normal_slash_dispatch() { diff --git a/src/Netclaw.Configuration.Tests/SessionDiagnosticsContextTests.cs b/src/Netclaw.Actors.Tests/SubAgents/SubAgentSessionScopeTests.cs similarity index 66% rename from src/Netclaw.Configuration.Tests/SessionDiagnosticsContextTests.cs rename to src/Netclaw.Actors.Tests/SubAgents/SubAgentSessionScopeTests.cs index 6d023b24d..ed5afab08 100644 --- a/src/Netclaw.Configuration.Tests/SessionDiagnosticsContextTests.cs +++ b/src/Netclaw.Actors.Tests/SubAgents/SubAgentSessionScopeTests.cs @@ -1,14 +1,14 @@ // ----------------------------------------------------------------------- -// +// // Copyright (C) 2026 - 2026 Petabridge, LLC // // ----------------------------------------------------------------------- -using Netclaw.Configuration; +using Netclaw.Actors.SubAgents; using Xunit; -namespace Netclaw.Configuration.Tests; +namespace Netclaw.Actors.Tests.SubAgents; -public sealed class SessionDiagnosticsContextTests +public sealed class SubAgentSessionScopeTests { [Theory] [InlineData("ch/thr/subagent/skill", "ch/thr")] @@ -16,7 +16,7 @@ public sealed class SessionDiagnosticsContextTests [InlineData("ch/thr/subagent/skill/inner", "ch/thr")] public void NormalizeSessionId_collapses_subagent_marker_to_parent_id(string input, string expected) { - Assert.Equal(expected, SessionDiagnosticsContext.NormalizeSessionId(input)); + Assert.Equal(expected, SubAgentSessionScope.NormalizeSessionId(input)); } [Theory] @@ -25,6 +25,6 @@ public void NormalizeSessionId_collapses_subagent_marker_to_parent_id(string inp [InlineData(" ")] public void NormalizeSessionId_returns_null_for_blank(string? input) { - Assert.Null(SessionDiagnosticsContext.NormalizeSessionId(input)); + Assert.Null(SubAgentSessionScope.NormalizeSessionId(input)); } } diff --git a/src/Netclaw.Actors.Tests/SubAgents/SubAgentSpawnObservabilityTests.cs b/src/Netclaw.Actors.Tests/SubAgents/SubAgentSpawnObservabilityTests.cs index 2bedfcb23..afea66548 100644 --- a/src/Netclaw.Actors.Tests/SubAgents/SubAgentSpawnObservabilityTests.cs +++ b/src/Netclaw.Actors.Tests/SubAgents/SubAgentSpawnObservabilityTests.cs @@ -4,6 +4,7 @@ // // ----------------------------------------------------------------------- using Microsoft.Extensions.Logging; +using Netclaw.Actors.Protocol; using Netclaw.Actors.SubAgents; using Netclaw.Actors.Tools; using Netclaw.Configuration; @@ -14,14 +15,11 @@ namespace Netclaw.Actors.Tests.SubAgents; /// -/// Regression coverage for sub-agent spawn observability. A sub-agent's own actor -/// logs go through Akka's async logger bridge, where the diagnostics AsyncLocal is -/// gone, so they never reach the per-session session.log. The spawn lifecycle -/// is instead recorded by parent-side breadcrumbs that must run while the parent -/// session scope is active — otherwise a refused or failed spawn is invisible in the -/// session transcript. These tests assert the scope is active at log time, which is -/// exactly the condition RollingFileLoggerProvider uses to route a line to -/// session.log. +/// Regression coverage for sub-agent spawn observability. Each lifecycle/rejection breadcrumb is +/// an ordinary structured log call wrapped in a SessionId scope; the file-logger +/// partitions scoped lines into the spawning session's session.log (routing itself is +/// covered by RollingFileLoggerPartitionTests). These tests assert the producer side: a +/// refused or failed spawn still logs its real reason under the session's id, so it is not lost. /// public sealed class SubAgentSpawnObservabilityTests : IDisposable { @@ -39,9 +37,9 @@ public SubAgentSpawnObservabilityTests() public void Dispose() => _dir.Dispose(); [Fact] - public async Task Spawner_missing_session_context_records_breadcrumbs_under_session_scope() + public async Task Spawner_missing_session_context_logs_lifecycle_under_session_scope() { - var logger = new CapturingLogger(); + var logger = new RecordingLogger(); // Only the parent-side breadcrumb path runs before the early return, so the // unused collaborators are never dereferenced. var spawner = new SubAgentSpawner( @@ -60,16 +58,19 @@ public async Task Spawner_missing_session_context_records_breadcrumbs_under_sess Profile("summarizer"), "do the work", null, context, TestContext.Current.CancellationToken); Assert.False(result.Success); - // The spawn attempt and its failure are both visible in the session transcript. - Assert.Contains(logger.Entries, e => e.SessionScope == SessionId && e.Message.Contains("spawn requested")); - Assert.Contains(logger.Entries, e => e.SessionScope == SessionId && e.Message.Contains("no session context available")); + // The spawn attempt and its failure are both logged under the session's id, so the + // file-logger routes them to that session's session.log. + var requested = Assert.Single(logger.Entries, e => e.Message.Contains("spawn requested", StringComparison.Ordinal)); + Assert.Equal(SessionId, requested.SessionId); + var noContext = Assert.Single(logger.Entries, e => e.Message.Contains("no session context available", StringComparison.Ordinal)); + Assert.Equal(SessionId, noContext.SessionId); } [Fact] - public async Task Tool_refusal_records_real_reason_under_session_scope() + public async Task Tool_refusal_logs_real_reason_under_session_scope() { - var logger = new CapturingLogger(); var registry = new SubAgentDefinitionRegistry(); + var logger = new RecordingLogger(); var tool = new SpawnAgentTool(registry, spawner: null!, _paths, logger: logger); // Public audience is refused with a deliberately opaque model-facing string; @@ -82,12 +83,10 @@ public async Task Tool_refusal_records_real_reason_under_session_scope() TestContext.Current.CancellationToken); Assert.Equal("Error: This tool is not available.", result); - Assert.Contains( + var refused = Assert.Single( logger.Entries, - e => e.Level == LogLevel.Warning - && e.SessionScope == SessionId - && e.Message.Contains("refused") - && e.Message.Contains("Public")); + e => e.Message.Contains("refused", StringComparison.Ordinal) && e.Message.Contains("Public", StringComparison.Ordinal)); + Assert.Equal(SessionId, refused.SessionId); } private static SubAgentProfile Profile(string name) => new() @@ -99,20 +98,41 @@ public async Task Tool_refusal_records_real_reason_under_session_scope() Visibility = SubAgentVisibility.UserFacing }; - private sealed class CapturingLogger : ILogger + // Records each log line with the SessionId on the scope active at emit time, so a test can + // assert a specific breadcrumb was logged under the right session. + private sealed class RecordingLogger : ILogger { - public readonly List<(LogLevel Level, string Message, string? SessionScope)> Entries = new(); + private readonly List _scopes = []; + public List<(string Message, string? SessionId)> Entries { get; } = []; - public IDisposable? BeginScope(TState state) where TState : notnull => null; + public IDisposable BeginScope(TState state) where TState : notnull + { + _scopes.Add(state); + return new Pop(_scopes); + } public bool IsEnabled(LogLevel logLevel) => true; - public void Log( - LogLevel logLevel, - EventId eventId, - TState state, - Exception? exception, - Func formatter) - => Entries.Add((logLevel, formatter(state, exception), SessionDiagnosticsContext.SessionId)); + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) + => Entries.Add((formatter(state, exception), ActiveSessionId())); + + private string? ActiveSessionId() + { + for (var i = _scopes.Count - 1; i >= 0; i--) + if (_scopes[i] is IEnumerable> kvps) + foreach (var kv in kvps) + if (kv.Key == NetclawLogProperties.SessionId && kv.Value is string s) + return s; + return null; + } + + private sealed class Pop(List scopes) : IDisposable + { + public void Dispose() + { + if (scopes.Count > 0) + scopes.RemoveAt(scopes.Count - 1); + } + } } } diff --git a/src/Netclaw.Actors/Memory/MemoryCurationActor.cs b/src/Netclaw.Actors/Memory/MemoryCurationActor.cs index ae7b9fda9..1983196b1 100644 --- a/src/Netclaw.Actors/Memory/MemoryCurationActor.cs +++ b/src/Netclaw.Actors/Memory/MemoryCurationActor.cs @@ -6,6 +6,7 @@ using Akka.Actor; using Akka.Event; using Microsoft.Extensions.AI; +using Netclaw.Actors.Sessions; using Netclaw.Actors.Sessions.Pipelines; using Netclaw.Configuration; using SessionId = Netclaw.Actors.Protocol.SessionId; @@ -298,7 +299,6 @@ private async Task EvaluateSingleAsync( try { using var cts = new CancellationTokenSource(LlmTimeout); - using var diagnosticsScope = SessionDiagnosticsContext.Push(sessionId.Value); var messages = new List { @@ -306,8 +306,12 @@ private async Task EvaluateSingleAsync( new(ChatRole.User, CurationPromptBuilder.BuildUserMessage(operation, candidates)) }; - var options = new ChatOptions + // SessionScopedChatOptions carries the session id so this sidecar's chat-client + // diagnostics route to the session's session.log and correlate in Seq/OTLP (replaces + // the deleted SessionDiagnosticsContext AsyncLocal). + var options = new SessionScopedChatOptions { + SessionId = sessionId.Value, // Headroom for reasoning models: with a tight cap a thinking model // spends the whole budget on hidden reasoning and returns empty // content, silently disabling this LLM tier. Non-reasoning models diff --git a/src/Netclaw.Actors/Protocol/NetclawLogProperties.cs b/src/Netclaw.Actors/Protocol/NetclawLogProperties.cs new file mode 100644 index 000000000..682708f15 --- /dev/null +++ b/src/Netclaw.Actors/Protocol/NetclawLogProperties.cs @@ -0,0 +1,32 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- + +namespace Netclaw.Actors.Protocol; + +/// +/// Shared structured-logging attribute keys used across the actor system, +/// channel adapters, and chat-client decorators. Centralising these constants +/// ensures that every log producer emits the same filterable field name so log +/// aggregators (Seq, OTLP) can correlate entries across subsystem boundaries. +/// +public static class NetclawLogProperties +{ + /// + /// Structured-logging attribute key correlating a log line to a session + /// ({channelId}/{threadTs}). Used as the WithContext / BeginScope key so + /// actor logs and chat-client decorator logs share one filterable field in + /// Seq/OTLP. + /// + public const string SessionId = "SessionId"; + + /// + /// Structured-logging attribute key correlating a log line to a specific + /// sub-agent run within its parent session. Used alongside + /// so sub-agent and parent logs share a parent + /// filter while remaining independently queryable by run. + /// + public const string SubSessionId = "SubSessionId"; +} diff --git a/src/Netclaw.Actors/Protocol/SessionLogDiagnostic.cs b/src/Netclaw.Actors/Protocol/SessionLogDiagnostic.cs index 5b6a551d4..76e407edc 100644 --- a/src/Netclaw.Actors/Protocol/SessionLogDiagnostic.cs +++ b/src/Netclaw.Actors/Protocol/SessionLogDiagnostic.cs @@ -4,16 +4,15 @@ // // ----------------------------------------------------------------------- using Akka.Actor; -using Netclaw.Configuration; namespace Netclaw.Actors.Protocol; /// -/// Pre-formatted diagnostic line carried from the MEL logger provider -/// to the SessionLogDispatcher. The provider snapshots -/// SessionDiagnosticsContext.SessionId at log-emit time and embeds -/// it here so the dispatcher routes the line by message field, not by -/// any ambient context that would not flow across actor mailboxes. +/// Pre-formatted diagnostic line published explicitly into a session's +/// session.log via the SessionLogDispatcher. The producer names the +/// owning on the message itself, so the dispatcher routes +/// the line by message field rather than by ambient context (which would not flow +/// across actor mailboxes) or by inferring intent from log metadata at the sink. /// public sealed record SessionLogDiagnostic(SessionId SessionId, string Line) : IWithSessionId, INoSerializationVerificationNeeded; diff --git a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs index 0c9a76a57..9673100a0 100644 --- a/src/Netclaw.Actors/Sessions/LlmSessionActor.cs +++ b/src/Netclaw.Actors/Sessions/LlmSessionActor.cs @@ -246,7 +246,7 @@ public LlmSessionActor( PersistenceId = $"session-{entityId}"; // Enrich logger with session context — all log messages automatically include SessionId - _log = Context.GetLogger().WithContext("SessionId", _sessionId.Value); + _log = Context.GetLogger().WithContext(NetclawLogProperties.SessionId, _sessionId.Value); // Load all non-MCP tools for initial LLM calls. // MCP tools are loaded dynamically via search_tools and can be retained for a @@ -1738,7 +1738,6 @@ internal static async Task InvokeMemoryExtractionCoreAsync( try { using var cts = new CancellationTokenSource(timeout); - using var diagnosticsScope = SessionDiagnosticsContext.Push(sessionId.Value); var extractionMessages = new List { new(Microsoft.Extensions.AI.ChatRole.System, @@ -1746,8 +1745,11 @@ internal static async Task InvokeMemoryExtractionCoreAsync( new(Microsoft.Extensions.AI.ChatRole.User, CompactionPromptBuilder.BuildMemoryExtractionUserPrompt(history)) }; + // Carry the session id so memory-extraction chat-client diagnostics route to the + // session's session.log and correlate in Seq/OTLP (replaces the deleted AsyncLocal). + var options = new SessionScopedChatOptions { SessionId = sessionId.Value }; var extractionResult = await StreamingResponseReader.ReadAsync( - client, extractionMessages, options: null, cts.Token); + client, extractionMessages, options, cts.Token); var extractedText = extractionResult.Response.Text ?? string.Empty; self.Tell(new MemoryExtractionCompleted { ExtractedMemories = extractedText }); } @@ -2745,20 +2747,19 @@ private void FireLlmCall(string? recallQuery = null, bool forceNoTools = false) var client = _chatClient; var exposedTools = ResolveExposedToolsForCurrentTurn(); - ChatOptions? options = null; + // Always carry the session id so the session-agnostic chat-client decorators + // (logging/retry/routing) can correlate LLM diagnostics — including provider + // failover/outage — back to this session in Seq. Tools are attached only when + // the turn exposes them; an empty Tools list is wire-equivalent to no options. + var options = new SessionScopedChatOptions { SessionId = _sessionId.Value }; if (!forceNoTools && exposedTools.Count > 0) - { - options = new ChatOptions - { - Tools = [.. exposedTools] - }; - } + options.Tools = [.. exposedTools]; _watchdog.Start(ProcessingWatchdog.LlmCall, _config.PrefillTimeout, Timers, _config.NoProgressTimeout); TurnLog().Info("turn_llm_call_start messages={MessageCount} toolsEnabled={ToolsEnabled} forceNoTools={ForceNoTools} callId={CallId}", messages.Count, - options?.Tools?.Count > 0, + options.Tools?.Count > 0, forceNoTools, _activeCallId); diff --git a/src/Netclaw.Actors/Sessions/Pipelines/SessionCompactionPipeline.cs b/src/Netclaw.Actors/Sessions/Pipelines/SessionCompactionPipeline.cs index 660bd8cc7..b0835e73b 100644 --- a/src/Netclaw.Actors/Sessions/Pipelines/SessionCompactionPipeline.cs +++ b/src/Netclaw.Actors/Sessions/Pipelines/SessionCompactionPipeline.cs @@ -173,7 +173,6 @@ public static async Task ExecuteAsync( { using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); cts.CancelAfter(sidecarTimeout); - using var diagnosticsScope = SessionDiagnosticsContext.Push(sessionId.Value); var observerMessages = new List { new(Microsoft.Extensions.AI.ChatRole.System, @@ -182,8 +181,11 @@ public static async Task ExecuteAsync( ObservationPromptBuilder.BuildObservationUserPrompt(remainingDiscarded)) }; + // Carry the session id so the observer's chat-client diagnostics route to the + // session's session.log and correlate in Seq/OTLP (replaces the deleted AsyncLocal). + var observerOptions = new SessionScopedChatOptions { SessionId = sessionId.Value }; var result = await StreamingResponseReader.ReadAsync( - client, observerMessages, options: null, cts.Token); + client, observerMessages, observerOptions, cts.Token); var text = result.Response.Text; if (string.IsNullOrWhiteSpace(text)) diff --git a/src/Netclaw.Actors/Sessions/Pipelines/SessionLlmInvoker.cs b/src/Netclaw.Actors/Sessions/Pipelines/SessionLlmInvoker.cs index a63dad637..76ce4e2ba 100644 --- a/src/Netclaw.Actors/Sessions/Pipelines/SessionLlmInvoker.cs +++ b/src/Netclaw.Actors/Sessions/Pipelines/SessionLlmInvoker.cs @@ -33,7 +33,6 @@ public static async Task InvokeAsync( // so sidecar calls (compaction, title gen) that bypass this invoker // naturally omit the header and round-robin across backends. SessionAffinityContext.SessionId = sessionId.Value; - using var diagnosticsScope = SessionDiagnosticsContext.Push(sessionId.Value); try { var response = await StreamAsync(client, messages, options, self, callId, cancellationToken); diff --git a/src/Netclaw.Actors/Sessions/Pipelines/SessionTitleGenerator.cs b/src/Netclaw.Actors/Sessions/Pipelines/SessionTitleGenerator.cs index 022e6e24f..c7f650b04 100644 --- a/src/Netclaw.Actors/Sessions/Pipelines/SessionTitleGenerator.cs +++ b/src/Netclaw.Actors/Sessions/Pipelines/SessionTitleGenerator.cs @@ -43,13 +43,16 @@ public static async Task GenerateAsync( try { using var cts = new CancellationTokenSource(timeout); - using var diagnosticsScope = SessionDiagnosticsContext.Push(sessionId.Value); var messages = new List { new(Microsoft.Extensions.AI.ChatRole.User, CompactionPromptBuilder.BuildTitleGenerationPrompt(history)) }; - var result = await StreamingResponseReader.ReadAsync(client, messages, options: null, cts.Token); + // Carry the session id so this sidecar's chat-client diagnostics (timing, retries, + // provider failover) route to the session's session.log and correlate in Seq/OTLP — + // the explicit carrier that replaced the deleted SessionDiagnosticsContext AsyncLocal. + var options = new SessionScopedChatOptions { SessionId = sessionId.Value }; + var result = await StreamingResponseReader.ReadAsync(client, messages, options, cts.Token); var title = result.Response.Text ?? string.Empty; if (string.IsNullOrWhiteSpace(title)) diff --git a/src/Netclaw.Actors/Sessions/SessionLogActor.cs b/src/Netclaw.Actors/Sessions/SessionLogActor.cs index 364976910..cb6105d30 100644 --- a/src/Netclaw.Actors/Sessions/SessionLogActor.cs +++ b/src/Netclaw.Actors/Sessions/SessionLogActor.cs @@ -27,18 +27,32 @@ namespace Netclaw.Actors.Sessions; /// /// File handle lifecycle: /// - Open once in with append mode + read-share. -/// - AutoFlush enabled — each WriteLine flushes immediately. -/// - Close/dispose in — no retry loop needed since the -/// handle is kept open and single-writer is enforced by the actor mailbox. +/// - The high-volume diagnostic lines are flushed in batches (after a write burst or on a +/// periodic tick), not per line: now that the whole per-session log stream lands here, an +/// fsync per line would dominate. The audit transcript (user/assistant/tool/usage) is +/// flushed immediately instead, so a hard process death cannot drop the audit record's tail. +/// is INotInfluenceReceiveTimeout so the flush cadence does +/// not keep an idle session alive. +/// - Close/dispose in (which flushes) — no retry loop needed since +/// the handle is kept open and single-writer is enforced by the actor mailbox. /// -public sealed class SessionLogActor : ReceiveActor +public sealed class SessionLogActor : ReceiveActor, IWithTimers { + // Flush after this many buffered writes (bounds the buffer under a burst) or on the + // periodic tick (bounds tail latency for a trickle), whichever comes first. + private const int FlushAfterWrites = 256; + private static readonly TimeSpan FlushInterval = TimeSpan.FromSeconds(1); + private readonly SessionId _sessionId; private readonly string _sessionLogsBasePath; private readonly TimeProvider _timeProvider; private readonly TimeSpan _idleTimeout; private readonly ILoggingAdapter _log = Context.GetLogger(); private StreamWriter? _writer; + private int _unflushedWrites; + private bool _flushFailing; + + public ITimerScheduler Timers { get; set; } = null!; public static Props CreateProps(SessionId sessionId, string sessionLogsBasePath, TimeProvider timeProvider, TimeSpan? idleTimeout = null) => Props.Create(() => new SessionLogActor(sessionId, sessionLogsBasePath, timeProvider, idleTimeout ?? TimeSpan.FromMinutes(10))); @@ -53,9 +67,18 @@ public SessionLogActor(SessionId sessionId, string sessionLogsBasePath, TimeProv Receive(OnUserMessage); Receive(OnOutput); Receive(OnDiagnostic); + Receive(_ => Flush()); Receive(_ => Context.Stop(Self)); } + // Periodic flush signal. INotInfluenceReceiveTimeout so the 1s cadence never resets the + // session's idle-stop timer. + private sealed class FlushTick : INotInfluenceReceiveTimeout + { + public static readonly FlushTick Instance = new(); + private FlushTick() { } + } + protected override void PreStart() { base.PreStart(); @@ -75,15 +98,78 @@ protected override void PreStart() bufferSize: 4096, useAsync: false); - _writer = new StreamWriter(stream) { AutoFlush = true }; + _writer = new StreamWriter(stream) { AutoFlush = false }; + Timers.StartPeriodicTimer("session-log-flush", FlushTick.Instance, FlushInterval); } protected override void PostStop() { - _writer?.Dispose(); + // Flush explicitly first so a write failure is reported via Flush's rate-limited warning. + // Dispose flushes again; guard it so a still-failing disk (the _flushFailing state) cannot + // throw IOException out of PostStop and bury the real cause under Akka's PostStop noise. + Flush(); + try + { + _writer?.Dispose(); + } + catch (Exception ex) + { + _log.Warning(ex, "Failed to close session.log for {Session}", _sessionId.Value); + } base.PostStop(); } + // Buffered write: lines accumulate in the StreamWriter buffer and are flushed in batches. + // Used for the high-volume diagnostic lines, where an fsync per line would dominate. + private void Write(string line) + { + _writer?.WriteLine(line); + if (++_unflushedWrites >= FlushAfterWrites) + Flush(); + } + + // Durable write for the audit transcript (user/assistant/tool/usage lines): flush immediately + // so a hard process death (SIGKILL/OOM) cannot drop the security/audit record's tail. The + // flush also drains any diagnostics buffered before this line, so audit lines are natural + // flush points. Diagnostics themselves stay on the batched Write() path above. + private void WriteDurable(string line) + { + Write(line); + Flush(); + } + + private void Flush() + { + // Nothing buffered AND not in a failing state → skip. While failing, fall through so the + // periodic 1s tick keeps retrying the flush (the bytes are still in the StreamWriter + // buffer) and becomes durable as soon as the disk recovers — even with no new writes. + if (_unflushedWrites == 0 && !_flushFailing) + return; + + try + { + _writer?.Flush(); + _unflushedWrites = 0; + if (_flushFailing) + { + _flushFailing = false; + _log.Info("session.log flushing recovered for {Session}", _sessionId.Value); + } + } + catch (Exception ex) + { + // Reset the write-batch counter so writes don't trigger a per-line flush storm during a + // persistent failure (full disk, locked file); _flushFailing stays set, so the 1s tick + // keeps retrying until recovery. Warn once on onset, once on recovery. + _unflushedWrites = 0; + if (!_flushFailing) + { + _flushFailing = true; + _log.Warning(ex, "Failed to flush session.log for {Session}; further flush failures suppressed until recovery", _sessionId.Value); + } + } + } + private void OnUserMessage(SendUserMessage msg) { try @@ -93,11 +179,11 @@ private void OnUserMessage(SendUserMessage msg) : string.Empty; var line = $"[{_timeProvider.GetUtcNow():o}] User: {TextTruncation.EllipsisAppend(msg.Content, 1000)}{mediaNote}"; - _writer?.WriteLine(line); + WriteDurable(line); } catch (Exception ex) { - _log.Warning(ex, "Dropped user message audit line for {SessionId}", _sessionId.Value); + _log.Warning(ex, "Dropped user message audit line for {Session}", _sessionId.Value); } } @@ -129,12 +215,12 @@ private void OnOutput(SessionOutput output) if (line is not null) { - _writer?.WriteLine($"[{_timeProvider.GetUtcNow():o}] {line}"); + WriteDurable($"[{_timeProvider.GetUtcNow():o}] {line}"); } } catch (Exception ex) { - _log.Warning(ex, "Dropped session log audit line for {SessionId}", _sessionId.Value); + _log.Warning(ex, "Dropped session log audit line for {Session}", _sessionId.Value); } } @@ -142,11 +228,11 @@ private void OnDiagnostic(SessionLogDiagnostic diagnostic) { try { - _writer?.WriteLine(diagnostic.Line); + Write(diagnostic.Line); } catch (Exception ex) { - _log.Warning(ex, "Dropped diagnostic audit line for {SessionId}", _sessionId.Value); + _log.Warning(ex, "Dropped diagnostic audit line for {Session}", _sessionId.Value); } } diff --git a/src/Netclaw.Actors/Sessions/SessionMemoryObserverActor.cs b/src/Netclaw.Actors/Sessions/SessionMemoryObserverActor.cs index 6f34a66b3..c4ff443fc 100644 --- a/src/Netclaw.Actors/Sessions/SessionMemoryObserverActor.cs +++ b/src/Netclaw.Actors/Sessions/SessionMemoryObserverActor.cs @@ -348,7 +348,6 @@ internal static async Task RunDistillationAsync( try { using var cts = new CancellationTokenSource(timeout); - using var diagnosticsScope = SessionDiagnosticsContext.Push(sessionId.Value); var messages = new List { new(Microsoft.Extensions.AI.ChatRole.System, DistillationSystemPrompt), @@ -356,7 +355,10 @@ internal static async Task RunDistillationAsync( sessionId, turnCount, transcript, existingProposals)) }; - var response = (await StreamingResponseReader.ReadAsync(client, messages, options: null, cts.Token)).Response; + // Carry the session id so memory-distillation chat-client diagnostics route to the + // session's session.log and correlate in Seq/OTLP (replaces the deleted AsyncLocal). + var options = new SessionScopedChatOptions { SessionId = sessionId.Value }; + var response = (await StreamingResponseReader.ReadAsync(client, messages, options, cts.Token)).Response; var text = response.Text ?? string.Empty; inputTokens = response.Usage?.InputTokenCount; diff --git a/src/Netclaw.Actors/Sessions/SessionScopedChatOptions.cs b/src/Netclaw.Actors/Sessions/SessionScopedChatOptions.cs new file mode 100644 index 000000000..ecf8f6610 --- /dev/null +++ b/src/Netclaw.Actors/Sessions/SessionScopedChatOptions.cs @@ -0,0 +1,40 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using Microsoft.Extensions.AI; + +namespace Netclaw.Actors.Sessions; + +/// +/// that also names the owning session, so the +/// session-agnostic chat-client decorators (logging / retry / routing, which live in +/// Netclaw.Daemon and are shared singletons across every session) can tag their +/// diagnostics with SessionId for per-session correlation in Seq/OTLP. +/// +/// This restores the correlation that the deleted SessionDiagnosticsContext +/// AsyncLocal used to provide, but through an explicit value carried on the call seam +/// rather than ambient context — ambient state does not flow across the actor mailbox +/// boundary between the session actor and the chat-client pipeline. +/// +/// The id rides as a typed CLR property, deliberately NOT in +/// : provider clients forward +/// AdditionalProperties verbatim onto the wire (see +/// OpenAiCompatibleChatClient), so a dictionary entry would leak the session id +/// into the LLM request body. A subclass property is invisible to the provider +/// serializers, which only read known fields plus AdditionalProperties. +/// +public sealed class SessionScopedChatOptions : ChatOptions +{ + /// The session whose turn this LLM call belongs to. For sub-agent calls this is + /// the parent session id, so OTEL/Seq can still group the call under the spawning session. + public required string SessionId { get; init; } + + /// For a sub-agent's LLM call, the sub-session id (its run scope + /// {parent}/subagent/{name}/{runId}). When set, the file-logger partitions these + /// lines into the sub-agent's own session.log (the local file is keyed by the + /// sub-session, while keeps OTEL grouping under the parent). Null + /// for top-level session calls. + public string? SubSessionId { get; init; } +} diff --git a/src/Netclaw.Actors/SubAgents/SpawnAgentTool.cs b/src/Netclaw.Actors/SubAgents/SpawnAgentTool.cs index c782b7113..92c2aee7d 100644 --- a/src/Netclaw.Actors/SubAgents/SpawnAgentTool.cs +++ b/src/Netclaw.Actors/SubAgents/SpawnAgentTool.cs @@ -141,19 +141,16 @@ private static string FormatResult(string agent, SubAgentResult result) private (string? Error, SubAgentProfile? Profile) Resolve(Params args, ToolExecutionContext context) { // Rejections here return a (sometimes deliberately opaque) error string to - // the model. Mirror the real reason to the session transcript under the - // parent session scope so an operator can tell *why* a spawn was refused — - // the "This tool is not available." string hides the audience-vs-disabled - // distinction from the model on purpose. - using var diagnosticsScope = SessionDiagnosticsContext.Push(context.SessionId); + // the model. The breadcrumb records the real reason under a SessionId scope (so it + // lands in the session.log) so an operator can tell *why* a spawn was refused; the + // "This tool is not available." string hides the audience-vs-disabled distinction on + // purpose. // Defense-in-depth: block subagent spawning for Public audience or when // the subagent subsystem is disabled. if (context.Audience == TrustAudience.Public || !_subAgentConfig.Enabled) { - _logger?.LogWarning( - "spawn_agent refused (agent={Agent}, audience={Audience}, subsystemEnabled={Enabled})", - args.Agent, context.Audience, _subAgentConfig.Enabled); + SubAgentSpawnBreadcrumbs.SpawnRefused(_logger, context, args.Agent, context.Audience, _subAgentConfig.Enabled); return ("Error: This tool is not available.", null); } @@ -169,9 +166,7 @@ private static string FormatResult(string agent, SubAgentResult result) if (profile is null || profile.Visibility != SubAgentVisibility.UserFacing) { var available = _registry.GetUserFacing(); - _logger?.LogWarning( - "spawn_agent refused: agent '{Agent}' not found or not user-facing (availableCount={Count})", - args.Agent, available.Count); + SubAgentSpawnBreadcrumbs.UnknownAgentRefused(_logger, context, args.Agent, available.Count); if (available.Count == 0) return ($"Error: No subagents are available. Agent '{args.Agent}' not found. Author one at {_paths.AgentsDirectory}/*.md or define a skill with metadata.subagent once #661 lands.", null); diff --git a/src/Netclaw.Actors/SubAgents/SubAgentActor.cs b/src/Netclaw.Actors/SubAgents/SubAgentActor.cs index fa2024a69..d28f62512 100644 --- a/src/Netclaw.Actors/SubAgents/SubAgentActor.cs +++ b/src/Netclaw.Actors/SubAgents/SubAgentActor.cs @@ -98,6 +98,13 @@ [Subagent Execution Contract] private IParentApprovalBridge? _approvalBridge; private ChannelWriter? _activitySink; + // Parent session id (scopeId with the "/subagent/..." suffix stripped) and the full + // sub-session scope id. Carried on the sub-agent's ChatOptions so its LLM-pipeline lines + // group under the parent in OTEL (SessionId) while the file-logger partitions them into the + // sub-agent's own session.log (SubSessionId) — matching the enriched logger's own context. + private string? _parentSessionId; + private string? _subSessionId; + // Default wait-for-first-delta budget when the spawn message carries none // (direct/test callers). Mirrors SessionConfig.PrefillTimeout so an unset // prefill never collapses to the tighter inter-delta budget — that collapse @@ -263,15 +270,18 @@ private void Idle() // (SubSessionId), and is plainly attributable to the sub-agent. scopeId is // "{parentSessionId}/subagent/{name}/{runId}"; NormalizeSessionId strips the // "/subagent/..." suffix to recover the parent. SessionId matches the key the - // session/channel actors already use (see SessionLoggingScope), so sub-agent - // and parent logs share one filterable attribute; SubSessionId isolates a - // single run within that session. - var parentSessionId = SessionDiagnosticsContext.NormalizeSessionId(scopeId); + // session/channel actors already tag their loggers with, so sub-agent and + // parent logs share one filterable attribute (so OTEL groups them under the parent); + // SubSessionId isolates a single run and is what the file-logger partitions on, so the + // sub-agent's lines land in its OWN session.log rather than the parent's. + var parentSessionId = SubAgentSessionScope.NormalizeSessionId(scopeId); + _parentSessionId = parentSessionId; + _subSessionId = scopeId; var enrichedLog = Context.GetLogger(); if (!string.IsNullOrWhiteSpace(parentSessionId)) - enrichedLog = enrichedLog.WithContext("SessionId", parentSessionId); + enrichedLog = enrichedLog.WithContext(NetclawLogProperties.SessionId, parentSessionId); if (!string.IsNullOrWhiteSpace(scopeId)) - enrichedLog = enrichedLog.WithContext("SubSessionId", scopeId); + enrichedLog = enrichedLog.WithContext(NetclawLogProperties.SubSessionId, scopeId); _log = enrichedLog; // The run is bounded by a two-phase inactivity watchdog re-armed on @@ -676,16 +686,19 @@ private void FireLlmCall(bool forceNoTools = false) var messages = new List(_history); var callId = ++_llmCallId; - ChatOptions? options = null; + // Carry the parent session id (when known) so the chat-client decorators group this + // sub-agent's LLM-pipeline lines under the spawning session in OTEL, plus the sub-session + // id so the file-logger partitions them into the sub-agent's OWN session.log. Direct/test + // callers leave both unset. + ChatOptions? options = _parentSessionId is { Length: > 0 } sid + ? new SessionScopedChatOptions { SessionId = sid, SubSessionId = _subSessionId } + : null; if (!forceNoTools && _aiTools.Count > 0) { - options = new ChatOptions - { - Tools = [.. _aiTools] - }; + options ??= new ChatOptions(); + options.Tools = [.. _aiTools]; } - var sessionId = _toolExecutionContext.SessionId is null ? (SessionId?)null : new SessionId(_toolExecutionContext.SessionId); _log.Info( "SubAgent [{AgentName}] LLM call start callId={CallId} iteration={Iteration} messages={MessageCount} toolsEnabled={ToolsEnabled} forceNoTools={ForceNoTools}", _definition.Name, @@ -694,7 +707,7 @@ private void FireLlmCall(bool forceNoTools = false) messages.Count, options?.Tools?.Count > 0, forceNoTools); - _ = InvokeLlmAsync(client, messages, options, sessionId, self, callId, _executionCts?.Token ?? CancellationToken.None); + _ = InvokeLlmAsync(client, messages, options, self, callId, _executionCts?.Token ?? CancellationToken.None); } private IReadOnlyList ResolveExposedAiTools() @@ -913,27 +926,20 @@ internal static Task InvokeLlmAsync( IChatClient client, List messages, ChatOptions? options, - SessionId? sessionId, IActorRef self, CancellationToken ct) - => InvokeLlmAsync(client, messages, options, sessionId, self, callId: 0, ct); + => InvokeLlmAsync(client, messages, options, self, callId: 0, ct); internal static async Task InvokeLlmAsync( IChatClient client, List messages, ChatOptions? options, - SessionId? sessionId, IActorRef self, long callId, CancellationToken ct) { try { - // Sub-agents share the parent's diagnostics scope: SessionDiagnosticsContext - // strips the "/subagent/..." suffix back to the parent id. Null is intentional - // for sub-agents that run outside any session. - using var diagnosticsScope = SessionDiagnosticsContext.Push(sessionId?.Value); - // Use streaming to match the main session path. The non-streaming // GetResponseAsync path drops reasoning content for some providers // (e.g., Qwen emits blocks that surface as TextReasoningContent diff --git a/src/Netclaw.Actors/SubAgents/SubAgentSessionScope.cs b/src/Netclaw.Actors/SubAgents/SubAgentSessionScope.cs new file mode 100644 index 000000000..0f0cc7caf --- /dev/null +++ b/src/Netclaw.Actors/SubAgents/SubAgentSessionScope.cs @@ -0,0 +1,33 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +namespace Netclaw.Actors.SubAgents; + +/// +/// Derives the owning (parent) session id from a sub-agent's composite scope id +/// ({parentId}/subagent/{name}/{runId}) by stripping the /subagent/… suffix. +/// Sub-agent log lines are tagged with this parent id as their SessionId so OTEL/Seq +/// groups a sub-agent's activity under the session that spawned it; the full composite id is +/// carried separately as SubSessionId and is what partitions the local +/// session.log into the sub-agent's OWN file. Keep this in sync with how sub-agent ids +/// are constructed in the spawner. ToolApprovalActor also calls this to walk a +/// sub-agent's approvals up to its parent, so this is the single owner of the /subagent/ +/// split. +/// +internal static class SubAgentSessionScope +{ + public static string? NormalizeSessionId(string? sessionId) + { + if (string.IsNullOrWhiteSpace(sessionId)) + return null; + + var value = sessionId.Trim(); + var subAgentMarker = value.IndexOf("/subagent/", StringComparison.Ordinal); + if (subAgentMarker > 0) + value = value[..subAgentMarker]; + + return value; + } +} diff --git a/src/Netclaw.Actors/SubAgents/SubAgentSpawnBreadcrumbs.cs b/src/Netclaw.Actors/SubAgents/SubAgentSpawnBreadcrumbs.cs new file mode 100644 index 000000000..6ec48cec0 --- /dev/null +++ b/src/Netclaw.Actors/SubAgents/SubAgentSpawnBreadcrumbs.cs @@ -0,0 +1,112 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using Microsoft.Extensions.Logging; +using Netclaw.Actors.Protocol; +using Netclaw.Configuration; +using Netclaw.Tools; + +namespace Netclaw.Actors.SubAgents; + +/// +/// Centralizes the sub-agent spawn-lifecycle log lines. Each is an ordinary structured log call +/// wrapped in a SessionId scope, so the file-logger partitions it into the spawning +/// session's session.log (and the OTLP exporter sees the session id as an attribute). +/// and are plain classes — no actor +/// WithContext — so the scope is what carries the id. The is +/// nullable so tool call sites with an optional logger can share these emitters. +/// +internal static class SubAgentSpawnBreadcrumbs +{ + public static void SpawnRequested(ILogger? logger, ToolExecutionContext context, string agentName, int taskChars) + { + using (BeginSessionScope(logger, context.SessionId)) + logger?.LogInformation( + "SubAgent [{AgentName}] spawn requested (taskChars={TaskChars})", + agentName, taskChars); + } + + public static void NoSessionContext(ILogger? logger, ToolExecutionContext context, string agentName) + { + using (BeginSessionScope(logger, context.SessionId)) + logger?.LogWarning( + "SubAgent [{AgentName}] cannot spawn — no session context available", + agentName); + } + + public static void NoToolsAvailable(ILogger? logger, ToolExecutionContext context, string agentName) + { + using (BeginSessionScope(logger, context.SessionId)) + logger?.LogWarning( + "SubAgent [{AgentName}] has no tools available under the parent audience policy — cannot spawn", + agentName); + } + + public static void ChildSpawnFailed(ILogger? logger, ToolExecutionContext context, string agentName, SubAgentRunId runId, Exception ex) + { + using (BeginSessionScope(logger, context.SessionId)) + logger?.LogError( + ex, "SubAgent [{AgentName}] failed to spawn child actor (runId={RunId})", + agentName, runId.Value); + } + + public static void ChildSpawned(ILogger? logger, ToolExecutionContext context, string agentName, SubAgentRunId runId) + { + using (BeginSessionScope(logger, context.SessionId)) + logger?.LogInformation( + "SubAgent [{AgentName}] child actor spawned (runId={RunId}); dispatching RunSubAgent", + agentName, runId.Value); + } + + public static void Completed(ILogger? logger, ToolExecutionContext context, string agentName, SubAgentRunId runId, bool success, long durationMs) + { + using (BeginSessionScope(logger, context.SessionId)) + logger?.LogInformation( + "SubAgent [{AgentName}] completed (runId={RunId}, success={Success}, duration={Duration}ms)", + agentName, runId.Value, success, durationMs); + } + + public static void RunFailed(ILogger? logger, ToolExecutionContext context, string agentName, SubAgentRunId runId, Exception ex) + { + using (BeginSessionScope(logger, context.SessionId)) + logger?.LogError( + ex, "SubAgent [{AgentName}] run failed (runId={RunId})", + agentName, runId.Value); + } + + public static void ToolDenied(ILogger? logger, ToolExecutionContext context, string agentName, string toolName) + { + // INFO so tool denials are visible in production: a sub-agent missing a tool may be unable + // to finish its task. Routed via the SessionId scope like every other breadcrumb (no + // hand-rolled (session=…) field), so it lands in the session.log without string drift. + using (BeginSessionScope(logger, context.SessionId)) + logger?.LogInformation( + "SubAgent [{AgentName}] tool '{ToolName}' denied by SubAgentToolPolicy", + agentName, toolName); + } + + public static void SpawnRefused(ILogger? logger, ToolExecutionContext context, string agentName, TrustAudience audience, bool subsystemEnabled) + { + using (BeginSessionScope(logger, context.SessionId)) + logger?.LogWarning( + "spawn_agent refused (agent={Agent}, audience={Audience}, subsystemEnabled={Enabled})", + agentName, audience, subsystemEnabled); + } + + public static void UnknownAgentRefused(ILogger? logger, ToolExecutionContext context, string agentName, int availableCount) + { + using (BeginSessionScope(logger, context.SessionId)) + logger?.LogWarning( + "spawn_agent refused: agent '{Agent}' not found or not user-facing (availableCount={Count})", + agentName, availableCount); + } + + // Opens the SessionId scope the file-logger routes on. No-op (null using) when there is no + // logger or no session — the line then falls through to daemon.log as a sessionless line. + private static IDisposable? BeginSessionScope(ILogger? logger, string? sessionId) => + logger is not null && !string.IsNullOrWhiteSpace(sessionId) + ? logger.BeginScope(new[] { new KeyValuePair(NetclawLogProperties.SessionId, sessionId) }) + : null; +} diff --git a/src/Netclaw.Actors/SubAgents/SubAgentSpawner.cs b/src/Netclaw.Actors/SubAgents/SubAgentSpawner.cs index 780bc02ba..de2b673ea 100644 --- a/src/Netclaw.Actors/SubAgents/SubAgentSpawner.cs +++ b/src/Netclaw.Actors/SubAgents/SubAgentSpawner.cs @@ -67,19 +67,14 @@ public async Task SpawnAsync( string? systemPromptOverlay = null, ChannelWriter? activitySink = null) { - // Session-scoped breadcrumbs: the sub-agent's own actor logs go through - // Akka's async logger bridge, where the diagnostics AsyncLocal is gone, so - // they never reach session.log. These parent-side lines run synchronously - // under the parent session scope, so the spawn lifecycle (request → outcome, - // including every early rejection) is always visible in the session transcript. - using var diagnosticsScope = SessionDiagnosticsContext.Push(context.SessionId); - _logger.LogInformation( - "SubAgent [{AgentName}] spawn requested (taskChars={TaskChars})", - profile.Name, task.Length); + // Parent-side spawn breadcrumbs — each event is fanned out to daemon.log/Seq and + // the parent's session.log from one place (see SubAgentSpawnBreadcrumbs), covering + // request → outcome plus early rejections that happen before the child even exists. + SubAgentSpawnBreadcrumbs.SpawnRequested(_logger, context, profile.Name, task.Length); if (context.SpawnChildActor is null) { - _logger.LogWarning("SubAgent [{AgentName}] cannot spawn — no session context available", profile.Name); + SubAgentSpawnBreadcrumbs.NoSessionContext(_logger, context, profile.Name); activitySink?.TryComplete(); return new SubAgentResult { @@ -94,9 +89,7 @@ public async Task SpawnAsync( var tools = ResolveTools(profile, context); if (tools.Count == 0) { - _logger.LogWarning( - "SubAgent [{AgentName}] has no tools available under the parent audience policy — cannot spawn", - profile.Name); + SubAgentSpawnBreadcrumbs.NoToolsAvailable(_logger, context, profile.Name); activitySink?.TryComplete(); return new SubAgentResult { @@ -158,9 +151,7 @@ public async Task SpawnAsync( // The child actor was never created (session actor ActorOf failed or the // spawn ask timed out). Record it to the session transcript before the // exception propagates to the tool pipeline. - _logger.LogError( - ex, "SubAgent [{AgentName}] failed to spawn child actor (runId={RunId})", - profile.Name, runId); + SubAgentSpawnBreadcrumbs.ChildSpawnFailed(_logger, context, profile.Name, runId, ex); // Balance the IsStarted=true notification above: the non-streaming path // (activitySink is null) relies solely on OnSubAgentActivity, so without // a terminal event the session UI shows a sub-agent stuck in "Started". @@ -177,9 +168,7 @@ public async Task SpawnAsync( throw; } - _logger.LogInformation( - "SubAgent [{AgentName}] child actor spawned (runId={RunId}); dispatching RunSubAgent", - profile.Name, runId); + SubAgentSpawnBreadcrumbs.ChildSpawned(_logger, context, profile.Name, runId); var sw = Stopwatch.StartNew(); try @@ -238,9 +227,7 @@ public async Task SpawnAsync( Findings = result.Findings }); - _logger.LogInformation( - "SubAgent [{AgentName}] completed (runId={RunId}, success={Success}, duration={Duration}ms)", - profile.Name, runId, result.Success, sw.ElapsedMilliseconds); + SubAgentSpawnBreadcrumbs.Completed(_logger, context, profile.Name, runId, result.Success, sw.ElapsedMilliseconds); return result with { @@ -265,7 +252,7 @@ public async Task SpawnAsync( Duration = sw.Elapsed }); - _logger.LogError(ex, "SubAgent [{AgentName}] run failed (runId={RunId})", profile.Name, runId); + SubAgentSpawnBreadcrumbs.RunFailed(_logger, context, profile.Name, runId, ex); return new SubAgentResult { Success = false, @@ -299,12 +286,7 @@ private IReadOnlyList ResolveTools(SubAgentProfile profile, ToolEx } else { - // Log at INFO so tool denials are visible in production logs. - // Sub-agents without certain tools may be unable to complete - // their tasks, and this information is important for debugging. - _logger.LogInformation( - "SubAgent [{AgentName}] tool '{ToolName}' denied by SubAgentToolPolicy", - profile.Name, tool.Name); + SubAgentSpawnBreadcrumbs.ToolDenied(_logger, context, profile.Name, tool.Name); } } diff --git a/src/Netclaw.Actors/Tools/ToolApprovalActor.cs b/src/Netclaw.Actors/Tools/ToolApprovalActor.cs index b150968a6..8fccdc6f1 100644 --- a/src/Netclaw.Actors/Tools/ToolApprovalActor.cs +++ b/src/Netclaw.Actors/Tools/ToolApprovalActor.cs @@ -6,6 +6,7 @@ using Akka.Actor; using Akka.Event; using Netclaw.Actors.Protocol; +using Netclaw.Actors.SubAgents; using Netclaw.Configuration; using Netclaw.Security; using Netclaw.Tools; @@ -104,11 +105,14 @@ private bool IsSessionApproved(SessionId sessionId, TrustAudience audience, Tool return true; } - var subagentMarker = scopeId.IndexOf("/subagent/", StringComparison.Ordinal); - if (subagentMarker <= 0) + // Walk to the parent session so a sub-agent inherits its parent's approvals. + // SubAgentSessionScope.NormalizeSessionId owns the "/subagent/" split (one + // implementation shared with log routing); break once there is nothing left to strip. + var parent = SubAgentSessionScope.NormalizeSessionId(scopeId); + if (string.IsNullOrEmpty(parent) || parent == scopeId) break; - scopeId = scopeId[..subagentMarker]; + scopeId = parent; } return false; diff --git a/src/Netclaw.Channels.Discord/DiscordSessionBindingActor.cs b/src/Netclaw.Channels.Discord/DiscordSessionBindingActor.cs index 918bd42c0..4c9bb6a6b 100644 --- a/src/Netclaw.Channels.Discord/DiscordSessionBindingActor.cs +++ b/src/Netclaw.Channels.Discord/DiscordSessionBindingActor.cs @@ -116,7 +116,7 @@ public DiscordSessionBindingActor( _log = Context.GetLogger() .WithContext("Adapter", "discord") - .WithContext("SessionId", _sessionId.Value) + .WithContext(NetclawLogProperties.SessionId, _sessionId.Value) .WithContext("DiscordChannelId", _channelId.Value) .WithContext("DiscordThreadOrMessageId", _threadOrMessageId.Value); diff --git a/src/Netclaw.Channels.Mattermost/MattermostSessionBindingActor.cs b/src/Netclaw.Channels.Mattermost/MattermostSessionBindingActor.cs index d372126e8..e2b4c5f19 100644 --- a/src/Netclaw.Channels.Mattermost/MattermostSessionBindingActor.cs +++ b/src/Netclaw.Channels.Mattermost/MattermostSessionBindingActor.cs @@ -110,7 +110,7 @@ public MattermostSessionBindingActor( _log = Context.GetLogger() .WithContext("Adapter", "mattermost") - .WithContext("SessionId", _sessionId.Value) + .WithContext(NetclawLogProperties.SessionId, _sessionId.Value) .WithContext("MattermostChannelId", _channelId.Value) .WithContext("MattermostRootPostId", _rootPostId.Value); diff --git a/src/Netclaw.Channels.Slack/SlackConversationActor.cs b/src/Netclaw.Channels.Slack/SlackConversationActor.cs index bf676ee56..1afdd5d91 100644 --- a/src/Netclaw.Channels.Slack/SlackConversationActor.cs +++ b/src/Netclaw.Channels.Slack/SlackConversationActor.cs @@ -117,7 +117,7 @@ public SlackConversationActor(SlackChannelId conversationId, SlackGatewayDepende : message.EventId.Value; var log = _log .WithContext("SlackThreadTs", threadTs.Value) - .WithContext("SessionId", sessionId.Value) + .WithContext(NetclawLogProperties.SessionId, sessionId.Value) .WithContext("TurnId", turnId) .WithContext("SlackEventId", message.EventId.Value); diff --git a/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs b/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs index bd18a387c..6a3435edb 100644 --- a/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs +++ b/src/Netclaw.Channels.Slack/SlackThreadBindingActor.cs @@ -101,7 +101,7 @@ public SlackThreadBindingActor( _handle = new SessionPipelineHandle(dependencies.Pipeline, Context.GetLogger(), "slack-thread"); _log = Context.GetLogger() .WithContext("Adapter", "slack") - .WithContext("SessionId", _sessionId.Value) + .WithContext(NetclawLogProperties.SessionId, _sessionId.Value) .WithContext("SlackChannelId", _channelId) .WithContext("SlackThreadTs", _threadTs); diff --git a/src/Netclaw.Channels/ChannelConversationActor.cs b/src/Netclaw.Channels/ChannelConversationActor.cs index 678703785..6dc0783da 100644 --- a/src/Netclaw.Channels/ChannelConversationActor.cs +++ b/src/Netclaw.Channels/ChannelConversationActor.cs @@ -208,7 +208,7 @@ private void HandleGatewayMessage(TMessage message) var log = Log .WithContext(ThreadLogContextKey, threadKey) - .WithContext("SessionId", sessionId.Value) + .WithContext(NetclawLogProperties.SessionId, sessionId.Value) .WithContext("TurnId", turnId) .WithContext(EventLogContextKey, eventId); diff --git a/src/Netclaw.Configuration/SessionDiagnosticsContext.cs b/src/Netclaw.Configuration/SessionDiagnosticsContext.cs deleted file mode 100644 index e5c2ec363..000000000 --- a/src/Netclaw.Configuration/SessionDiagnosticsContext.cs +++ /dev/null @@ -1,62 +0,0 @@ -// ----------------------------------------------------------------------- -// -// Copyright (C) 2026 - 2026 Petabridge, LLC -// -// ----------------------------------------------------------------------- -namespace Netclaw.Configuration; - -/// -/// Ambient session context used by diagnostics sinks that need to distinguish -/// daemon-global logs from session-owned logs. -/// -public static class SessionDiagnosticsContext -{ - private static readonly AsyncLocal Current = new(); - - public static string? SessionId - { - get => Current.Value; - set => Current.Value = NormalizeSessionId(value); - } - - public static IDisposable Push(string? sessionId) - { - var prior = Current.Value; - Current.Value = NormalizeSessionId(sessionId); - return new RestoreScope(prior); - } - - /// - /// Trims whitespace and collapses any nested sub-agent suffix back to the - /// owning session id. Sub-agents run as ephemeral children of a parent - /// session and reuse its session.log for the audit trail; treating - /// their composite ids ({parentId}/subagent/{agentName}) as - /// distinct sessions would scatter sub-agent diagnostics into per-agent - /// files that operators do not monitor. The split here is the only - /// place that decision is made — keep it in sync with how sub-agent - /// ids are constructed inside the sub-agent spawner. - /// - public static string? NormalizeSessionId(string? sessionId) - { - if (string.IsNullOrWhiteSpace(sessionId)) - return null; - - var value = sessionId.Trim(); - var subAgentMarker = value.IndexOf("/subagent/", StringComparison.Ordinal); - if (subAgentMarker > 0) - value = value[..subAgentMarker]; - - return value; - } - - private sealed class RestoreScope(string? prior) : IDisposable - { - private string? _prior = prior; - - public void Dispose() - { - Current.Value = _prior; - _prior = null; - } - } -} diff --git a/src/Netclaw.Daemon.Tests/Configuration/LoggingChatClientTests.cs b/src/Netclaw.Daemon.Tests/Configuration/LoggingChatClientTests.cs index 9a7ec0c0e..11549c5b7 100644 --- a/src/Netclaw.Daemon.Tests/Configuration/LoggingChatClientTests.cs +++ b/src/Netclaw.Daemon.Tests/Configuration/LoggingChatClientTests.cs @@ -5,7 +5,7 @@ // ----------------------------------------------------------------------- using Microsoft.Extensions.AI; using Microsoft.Extensions.Logging; -using Netclaw.Configuration; +using Netclaw.Actors.Sessions; using Netclaw.Daemon.Configuration; using Xunit; // Netclaw's LoggingChatClient collides with Microsoft.Extensions.AI.LoggingChatClient @@ -75,6 +75,7 @@ public async Task Streaming_LogsPromptSummaryInDebugMode() ] }, TestContext.Current.CancellationToken)) { + // Drain the stream to completion so the pipeline (scope/retry/logging) runs; updates aren't asserted here. } Assert.Contains(logs, l => l.Contains("LLM prompt summary")); @@ -95,27 +96,36 @@ public async Task Streaming_LogsPromptDumpWhenTraceEnabled() } [Fact] - public async Task Streaming_attaches_SessionId_scope_from_diagnostics_context() + public async Task Streaming_attaches_SessionId_scope_from_options() { + // The decorator is session-agnostic; it learns the session from the call's + // SessionScopedChatOptions, replacing the deleted AsyncLocal diagnostics context. var logger = new ScopeCapturingLogger(); var client = new LoggingChatClient(new FakeChatClient(streaming: true), logger); - using (SessionDiagnosticsContext.Push("ch/thread")) + await foreach (var _ in client.GetStreamingResponseAsync( + [new ChatMessage(ChatRole.User, "hi")], + new SessionScopedChatOptions { SessionId = "ch/thread" }, + TestContext.Current.CancellationToken)) { - await Drain(client); + // Drain the stream to completion so the pipeline (scope/retry/logging) runs; updates aren't asserted here. } Assert.True(logger.HasSessionScope("ch/thread")); } [Fact] - public async Task Streaming_without_session_context_attaches_no_scope() + public async Task Streaming_with_plain_options_attaches_no_scope() { - Assert.Null(SessionDiagnosticsContext.SessionId); + // A sidecar/session-agnostic call carries a plain ChatOptions: no session scope. var logger = new ScopeCapturingLogger(); var client = new LoggingChatClient(new FakeChatClient(streaming: true), logger); - await Drain(client); + await foreach (var _ in client.GetStreamingResponseAsync( + [new ChatMessage(ChatRole.User, "hi")], new ChatOptions(), TestContext.Current.CancellationToken)) + { + // Drain the stream to completion so the pipeline (scope/retry/logging) runs; updates aren't asserted here. + } Assert.False(logger.HasAnySessionScope()); } @@ -125,6 +135,7 @@ private static async Task Drain(IChatClient client) await foreach (var _ in client.GetStreamingResponseAsync( [new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.CancellationToken)) { + // Drain the stream to completion so the pipeline (scope/retry/logging) runs; updates aren't asserted here. } } diff --git a/src/Netclaw.Daemon.Tests/Configuration/OpenAiCompatibleChatClientTests.cs b/src/Netclaw.Daemon.Tests/Configuration/OpenAiCompatibleChatClientTests.cs index ba41ca78a..8607e4896 100644 --- a/src/Netclaw.Daemon.Tests/Configuration/OpenAiCompatibleChatClientTests.cs +++ b/src/Netclaw.Daemon.Tests/Configuration/OpenAiCompatibleChatClientTests.cs @@ -98,6 +98,35 @@ [new ChatMessage(ChatRole.User, "hello")], Assert.Contains("\"required\":[\"query\"]", body, StringComparison.Ordinal); } + [Fact] + public async Task DoesNotLeakSessionId_FromSessionScopedChatOptions_ToWire() + { + // SessionScopedChatOptions carries the session id as a CLR property, NOT in + // AdditionalProperties — which this client forwards verbatim as top-level JSON. + // The session id must never appear in the outbound request body. + string? body = null; + using var handler = new RecordingHandler(req => + { + body = req.Content is null ? null : req.Content.ReadAsStringAsync().GetAwaiter().GetResult(); + return new HttpResponseMessage(HttpStatusCode.OK) + { + Content = new StringContent("{\"id\":\"1\",\"model\":\"test\",\"choices\":[{\"finish_reason\":\"stop\",\"message\":{\"role\":\"assistant\",\"content\":\"hi\"}}]}", Encoding.UTF8, "application/json") + }; + }); + using var httpClient = new HttpClient(handler) { BaseAddress = new Uri("http://localhost:8000") }; + var endpoint = OpenAiCompatibleEndpoint.FromBaseUrl("http://localhost:8000"); + var client = new OpenAiCompatibleChatClient(httpClient, endpoint, "test-model"); + + await client.GetResponseAsync( + [new ChatMessage(ChatRole.User, "hello")], + new Netclaw.Actors.Sessions.SessionScopedChatOptions { SessionId = "C123/167.42" }, + cancellationToken: TestContext.Current.CancellationToken); + + Assert.NotNull(body); + Assert.DoesNotContain("SessionId", body, StringComparison.OrdinalIgnoreCase); + Assert.DoesNotContain("C123/167.42", body, StringComparison.Ordinal); + } + [Fact] public async Task CollapsesMultipleSystemMessages_IntoSingleLeadingSystemMessage() { diff --git a/src/Netclaw.Daemon.Tests/Configuration/PipelineChatClientFactoryTests.cs b/src/Netclaw.Daemon.Tests/Configuration/PipelineChatClientFactoryTests.cs index b92516be7..a2491d4a6 100644 --- a/src/Netclaw.Daemon.Tests/Configuration/PipelineChatClientFactoryTests.cs +++ b/src/Netclaw.Daemon.Tests/Configuration/PipelineChatClientFactoryTests.cs @@ -3,9 +3,12 @@ // Copyright (C) 2026 - 2026 Petabridge, LLC // // ----------------------------------------------------------------------- +using System.Net; +using System.Runtime.CompilerServices; using Microsoft.Extensions.AI; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; +using Netclaw.Actors.Sessions; using Netclaw.Configuration; using Netclaw.Daemon.Configuration; using Xunit; @@ -52,6 +55,117 @@ [new ChatMessage(ChatRole.User, "hi")], cancellationToken: TestContext.Current.C Assert.Contains(logs, l => l.Contains("LLM streaming call completed")); // Logging middleware wired } + [Fact] + public async Task Compose_streaming_tags_SessionId_scope_through_pipeline() + { + // Cross-cutting invariant: SessionScopedChatOptions must survive *by reference* + // through the composed Logging -> Retry pipeline (no decorator clones it down to a + // base ChatOptions), so the streaming production path still surfaces SessionId as a + // Seq scope. A future decorator that rebuilt options would break this test, not just + // a unit decorator tested in isolation. + var logger = new ScopeCapturingLogger(); + var pipeline = PipelineChatClientFactory.Compose( + new FakeChatClient(streaming: true), _policy, new SingleLoggerFactory(logger), TimeProvider.System); + + await foreach (var _ in pipeline.GetStreamingResponseAsync( + [new ChatMessage(ChatRole.User, "hi")], + new SessionScopedChatOptions { SessionId = "ch/thread" }, + TestContext.Current.CancellationToken)) + { + // Drain the stream to completion so the pipeline (scope/retry/logging) runs; updates aren't asserted here. + } + + Assert.True(logger.HasSessionScope("ch/thread")); + } + + [Fact] + public async Task Compose_streaming_retry_warning_inherits_SessionId_scope() + { + // RetryingChatClient deliberately opens no SessionId scope of its own — the retry + // warning must still be correlated, inheriting the enclosing LoggingChatClient + // streaming scope. Prove it at the message level: the retry-warning line is emitted + // while the session id is the active scope. + var logger = new MessageScopeLogger(); + var attempts = 0; + var leaf = new FakeChatClient(streamHandler: (_, _, ct) => + { + attempts++; + return ThrowFirstThenYield(shouldThrow: attempts < 2, ct); + }); + var pipeline = PipelineChatClientFactory.Compose( + leaf, _policy, new SingleLoggerFactory(logger), TimeProvider.System); + + await foreach (var _ in pipeline.GetStreamingResponseAsync( + [new ChatMessage(ChatRole.User, "hi")], + new SessionScopedChatOptions { SessionId = "ch/thread" }, + TestContext.Current.CancellationToken)) + { + // Drain the stream to completion so the pipeline (scope/retry/logging) runs; updates aren't asserted here. + } + + var retryWarning = Assert.Single( + logger.Entries, e => e.Message.Contains("LLM call failed (attempt", StringComparison.Ordinal)); + Assert.Equal("ch/thread", retryWarning.SessionId); + } + + private static async IAsyncEnumerable ThrowFirstThenYield( + bool shouldThrow, [EnumeratorCancellation] CancellationToken cancellationToken) + { + await Task.Yield(); + cancellationToken.ThrowIfCancellationRequested(); + if (shouldThrow) + throw new HttpRequestException("transient", null, HttpStatusCode.TooManyRequests); + + yield return new ChatResponseUpdate { Role = ChatRole.Assistant, Contents = [new TextContent("ok")] }; + } + + // Records, for each log call, the session id on the active scope stack — so a test can + // assert that a specific line (not just the call as a whole) was emitted in scope. + private sealed class MessageScopeLogger : ILogger + { + private readonly List _scopes = []; + public List<(string Message, string? SessionId)> Entries { get; } = []; + + public IDisposable BeginScope(TState state) where TState : notnull + { + _scopes.Add(state); + return new Pop(_scopes); + } + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, EventId eventId, TState state, + Exception? exception, Func formatter) + => Entries.Add((formatter(state, exception), ActiveSessionId())); + + private string? ActiveSessionId() + { + for (var i = _scopes.Count - 1; i >= 0; i--) + if (_scopes[i] is IEnumerable> kvps) + foreach (var kv in kvps) + if (kv.Key == Netclaw.Actors.Protocol.NetclawLogProperties.SessionId && kv.Value is string s) + return s; + return null; + } + + private sealed class Pop(List scopes) : IDisposable + { + public void Dispose() + { + if (scopes.Count > 0) + scopes.RemoveAt(scopes.Count - 1); + } + } + } + + private sealed class SingleLoggerFactory(ILogger logger) : ILoggerFactory + { + public ILogger CreateLogger(string categoryName) => logger; + public void AddProvider(ILoggerProvider provider) { } + public void Dispose() { } + } + private sealed class ListLoggerFactory : ILoggerFactory { private readonly List _logs; diff --git a/src/Netclaw.Daemon.Tests/Configuration/RetryingChatClientTests.cs b/src/Netclaw.Daemon.Tests/Configuration/RetryingChatClientTests.cs index 6441ff682..9035e013c 100644 --- a/src/Netclaw.Daemon.Tests/Configuration/RetryingChatClientTests.cs +++ b/src/Netclaw.Daemon.Tests/Configuration/RetryingChatClientTests.cs @@ -117,6 +117,12 @@ await Assert.ThrowsAsync(() => Assert.Equal(4, attempts); } + // NOTE: RetryingChatClient deliberately does NOT open its own SessionId scope — it + // inherits the enclosing LoggingChatClient scope in the composed pipeline. The retry + // warning's session correlation is therefore covered by + // PipelineChatClientFactoryTests.Compose_streaming_retry_warning_inherits_SessionId_scope, + // which exercises the real composition, not this decorator in isolation. + [Fact] public async Task DoesNotRetryNonTransientErrors() { diff --git a/src/Netclaw.Daemon.Tests/Configuration/RollingFileLoggerPartitionTests.cs b/src/Netclaw.Daemon.Tests/Configuration/RollingFileLoggerPartitionTests.cs new file mode 100644 index 000000000..396e2e807 --- /dev/null +++ b/src/Netclaw.Daemon.Tests/Configuration/RollingFileLoggerPartitionTests.cs @@ -0,0 +1,303 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using Akka.Hosting; +using Akka.Hosting.TestKit; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Time.Testing; +using Netclaw.Actors.Protocol; +using Netclaw.Daemon.Configuration; +using Xunit; + +namespace Netclaw.Daemon.Tests.Configuration; + +/// +/// The provider owns the LOCAL partition of the log stream: a line tagged with a session id by +/// session-SERVING code (an actor's WithContext, which the Akka bridge surfaces alongside a +/// LogSource; or a BeginScope) goes to that session's session.log (Tell'd as a +/// to the dispatcher) and NOT to daemon.log. A daemon-service +/// line that merely NAMES a session in its message template (a bare {SessionId} field with no +/// LogSource) stays in daemon.log, as does everything sessionless. The session-log writer's own +/// lines are excluded so a write-failure log cannot recurse. +/// +public sealed class RollingFileLoggerPartitionTests : TestKit +{ + private static readonly DateTimeOffset FixedNow = DateTimeOffset.Parse("2026-05-07T12:00:00Z"); + + protected override void ConfigureAkka(AkkaConfigurationBuilder builder, IServiceProvider provider) + { + } + + [Fact] + public async Task Session_tagged_line_routes_to_session_log_not_daemon_log() + { + var (dir, daemonPath) = TempPaths(); + var dispatcher = CreateTestProbe("dispatcher"); + + using (var provider = new RollingFileLoggerProvider(daemonPath, new FakeTimeProvider(FixedNow))) + { + provider.AttachSessionDispatcher(Task.FromResult(dispatcher.Ref)); + provider.CreateLogger("Akka.Actor.ActorSystem") + .Log(LogLevel.Information, new EventId(0), ActorState("C1/T1"), null, (_, _) => "spawn requested"); + + var diag = await dispatcher.ExpectMsgAsync(cancellationToken: TestContext.Current.CancellationToken); + Assert.Equal("C1/T1", diag.SessionId.Value); + Assert.Contains("spawn requested", diag.Line, StringComparison.Ordinal); + } + + // Partition: the session-tagged line is NOT also in daemon.log. + Assert.DoesNotContain("spawn requested", ReadDaemonLog(dir), StringComparison.Ordinal); + Cleanup(dir); + } + + [Fact] + public async Task Session_id_carried_in_a_scope_routes() + { + var (dir, daemonPath) = TempPaths(); + var dispatcher = CreateTestProbe("dispatcher"); + var provider = new RollingFileLoggerProvider(daemonPath, new FakeTimeProvider(FixedNow)); + + // A real LoggerFactory wires the external scope provider into the sink, mirroring the + // chat-client decorators that carry the session id via BeginScope rather than the message. + var factory = LoggerFactory.Create(b => b.AddProvider(provider)); + provider.AttachSessionDispatcher(Task.FromResult(dispatcher.Ref)); + + var logger = factory.CreateLogger("Netclaw.LlmPipeline"); + using (logger.BeginScope(new[] { new KeyValuePair(NetclawLogProperties.SessionId, "C2/T2") })) + logger.LogInformation("LLM streaming call completed"); + + var diag = await dispatcher.ExpectMsgAsync(cancellationToken: TestContext.Current.CancellationToken); + Assert.Equal("C2/T2", diag.SessionId.Value); + Assert.Contains("LLM streaming call completed", diag.Line, StringComparison.Ordinal); + + factory.Dispose(); // disposes the provider, draining daemon.log + Assert.DoesNotContain("LLM streaming call completed", ReadDaemonLog(dir), StringComparison.Ordinal); + Cleanup(dir); + } + + [Fact] + public async Task Sub_agent_line_routes_to_its_own_file_keyed_by_sub_session_id() + { + var (dir, daemonPath) = TempPaths(); + var dispatcher = CreateTestProbe("dispatcher"); + + using (var provider = new RollingFileLoggerProvider(daemonPath, new FakeTimeProvider(FixedNow))) + { + provider.AttachSessionDispatcher(Task.FromResult(dispatcher.Ref)); + + // A bridged sub-agent line carries the parent SessionId (for OTEL grouping) AND the + // sub-session id; the LOCAL file is partitioned by the sub-session. + var state = ActorState("C1/T1", "C1/T1/subagent/summarizer/ab12"); + provider.CreateLogger("Akka.Actor.ActorSystem") + .Log(LogLevel.Information, new EventId(0), state, null, (_, _) => "sub-agent did work"); + + var diag = await dispatcher.ExpectMsgAsync(cancellationToken: TestContext.Current.CancellationToken); + Assert.Equal("C1/T1/subagent/summarizer/ab12", diag.SessionId.Value); // sub file, not the parent + Assert.Contains("sub-agent did work", diag.Line, StringComparison.Ordinal); + } + + Cleanup(dir); + } + + [Fact] + public async Task Sub_session_id_carried_in_a_scope_routes_to_the_sub_agent_file() + { + var (dir, daemonPath) = TempPaths(); + var dispatcher = CreateTestProbe("dispatcher"); + var provider = new RollingFileLoggerProvider(daemonPath, new FakeTimeProvider(FixedNow)); + var factory = LoggerFactory.Create(b => b.AddProvider(provider)); + provider.AttachSessionDispatcher(Task.FromResult(dispatcher.Ref)); + + var logger = factory.CreateLogger("Netclaw.LlmPipeline"); + using (logger.BeginScope(new[] + { + new KeyValuePair(NetclawLogProperties.SessionId, "C2/T2"), + new KeyValuePair(NetclawLogProperties.SubSessionId, "C2/T2/subagent/coder/cd34"), + })) + logger.LogInformation("sub-agent LLM streaming call completed"); + + var diag = await dispatcher.ExpectMsgAsync(cancellationToken: TestContext.Current.CancellationToken); + Assert.Equal("C2/T2/subagent/coder/cd34", diag.SessionId.Value); + Assert.Contains("sub-agent LLM streaming call completed", diag.Line, StringComparison.Ordinal); + + factory.Dispose(); + Cleanup(dir); + } + + [Fact] + public async Task Sessionless_line_goes_to_daemon_log() + { + var (dir, daemonPath) = TempPaths(); + var dispatcher = CreateTestProbe("dispatcher"); + + using (var provider = new RollingFileLoggerProvider(daemonPath, new FakeTimeProvider(FixedNow))) + { + provider.AttachSessionDispatcher(Task.FromResult(dispatcher.Ref)); + provider.CreateLogger("Netclaw.Daemon").LogInformation("daemon listening on port 8080"); + + await dispatcher.ExpectNoMsgAsync(TimeSpan.FromMilliseconds(300), TestContext.Current.CancellationToken); + } + + Assert.Contains("daemon listening on port 8080", ReadDaemonLog(dir), StringComparison.Ordinal); + Cleanup(dir); + } + + [Fact] + public async Task Daemon_service_line_naming_a_session_in_its_message_stays_in_daemon_log() + { + var (dir, daemonPath) = TempPaths(); + var dispatcher = CreateTestProbe("dispatcher"); + + using (var provider = new RollingFileLoggerProvider(daemonPath, new FakeTimeProvider(FixedNow))) + { + provider.AttachSessionDispatcher(Task.FromResult(dispatcher.Ref)); + + // A plain daemon-service ILogger line that merely names a session in its message + // template — no actor LogSource, no scope. It is daemon infrastructure, not the + // session's own work, so it must NOT be diverted into session.log. + provider.CreateLogger("Netclaw.Daemon.Gateway.SessionCatalogService") + .LogWarning("Failed to mark session {SessionId} active", "C7/T7"); + + await dispatcher.ExpectNoMsgAsync(TimeSpan.FromMilliseconds(300), TestContext.Current.CancellationToken); + } + + Assert.Contains("Failed to mark session", ReadDaemonLog(dir), StringComparison.Ordinal); + Cleanup(dir); + } + + [Fact] + public async Task Session_log_writers_own_line_is_not_routed_back() + { + var (dir, daemonPath) = TempPaths(); + var dispatcher = CreateTestProbe("dispatcher"); + + using (var provider = new RollingFileLoggerProvider(daemonPath, new FakeTimeProvider(FixedNow))) + { + provider.AttachSessionDispatcher(Task.FromResult(dispatcher.Ref)); + + // Shape of a bridged Akka log from the SessionLogActor: it carries a SessionId but its + // ActorPath is under the session-log dispatcher. It must NOT route (feedback guard). + var state = new List> + { + new(NetclawLogProperties.SessionId, "C3/T3"), + new("ActorPath", "akka://netclaw/user/session-log-dispatcher/C3%2FT3"), + }; + provider.CreateLogger("Akka.Actor.ActorSystem") + .Log(LogLevel.Warning, new EventId(0), state, null, (_, _) => "Failed to flush session.log"); + + await dispatcher.ExpectNoMsgAsync(TimeSpan.FromMilliseconds(300), TestContext.Current.CancellationToken); + } + + Assert.Contains("Failed to flush session.log", ReadDaemonLog(dir), StringComparison.Ordinal); + Cleanup(dir); + } + + [Fact] + public async Task Session_line_before_dispatcher_resolves_falls_back_to_daemon_log() + { + var (dir, daemonPath) = TempPaths(); + var dispatcher = CreateTestProbe("dispatcher"); + var pending = new TaskCompletionSource(); + + using (var provider = new RollingFileLoggerProvider(daemonPath, new FakeTimeProvider(FixedNow))) + { + provider.AttachSessionDispatcher(pending.Task); // never resolves during this test + provider.CreateLogger("Akka.Actor.ActorSystem") + .Log(LogLevel.Information, new EventId(0), ActorState("C4/T4"), null, (_, _) => "before resolve"); + + // No buffering: a routable line logged before the dispatcher resolves is not held for + // it; it goes straight to daemon.log (the dispatcher never sees it). + await dispatcher.ExpectNoMsgAsync(TimeSpan.FromMilliseconds(300), TestContext.Current.CancellationToken); + } + + Assert.Contains("before resolve", ReadDaemonLog(dir), StringComparison.Ordinal); + Cleanup(dir); + } + + [Fact] + public async Task Dispatcher_resolution_failure_falls_back_and_beacons_to_daemon_log() + { + var (dir, daemonPath) = TempPaths(); + var pending = new TaskCompletionSource(); + var provider = new RollingFileLoggerProvider(daemonPath, new FakeTimeProvider(FixedNow)); + try + { + provider.AttachSessionDispatcher(pending.Task); // dispatcher not resolved + provider.CreateLogger("Akka.Actor.ActorSystem") + .Log(LogLevel.Warning, new EventId(0), ActorState("C9/T9"), null, (_, _) => "before failure"); + + pending.SetException(new InvalidOperationException("dispatcher never registered")); + + // The session line fell back to daemon.log at log time (dispatcher still null); on + // failure a single beacon records that per-session routing is disabled. + await AwaitAssertAsync( + async () => + { + var text = await ReadDaemonSharedAsync(dir, TestContext.Current.CancellationToken); + Assert.Contains("before failure", text, StringComparison.Ordinal); + Assert.Contains("per-session routing disabled", text, StringComparison.Ordinal); + }, + TimeSpan.FromSeconds(5), + cancellationToken: TestContext.Current.CancellationToken); + } + finally + { + provider.Dispose(); + } + + Cleanup(dir); + } + + private static async Task ReadDaemonSharedAsync(string dir, CancellationToken ct) + { + var files = Directory.GetFiles(dir, "daemon-*.log"); + if (files.Length == 0) + return string.Empty; + await using var stream = new FileStream(files[0], FileMode.Open, FileAccess.Read, FileShare.ReadWrite | FileShare.Delete); + using var reader = new StreamReader(stream); + return await reader.ReadToEndAsync(ct); + } + + // Shape of a bridged Akka actor log: WithContext("SessionId") surfaces alongside the actor's + // LogSource. The LogSource is what marks the line as session-SERVING (vs a daemon service that + // merely names a session in its message), so the router treats the session id as routable. + private static List> ActorState(string sessionId, string? subSessionId = null) + { + var state = new List> + { + new(NetclawLogProperties.SessionId, sessionId), + new("LogSource", "[akka://netclaw/user/session-manager#1]"), + }; + if (subSessionId is not null) + state.Add(new(NetclawLogProperties.SubSessionId, subSessionId)); + return state; + } + + private static (string Dir, string DaemonPath) TempPaths() + { + var dir = Path.Join(Path.GetTempPath(), $"netclaw-partition-{Guid.NewGuid():N}"); + Directory.CreateDirectory(dir); + return (dir, Path.Join(dir, "daemon.log")); + } + + private static string ReadDaemonLog(string dir) + { + var files = Directory.GetFiles(dir, "daemon-*.log"); + return files.Length == 0 ? string.Empty : File.ReadAllText(files[0]); + } + + private static void Cleanup(string dir) + { + try + { + if (Directory.Exists(dir)) + Directory.Delete(dir, recursive: true); + } + catch (IOException ex) + { + Console.Error.WriteLine($"[RollingFileLoggerPartitionTests] cleanup failed: {ex.Message}"); + } + } +} diff --git a/src/Netclaw.Daemon.Tests/Configuration/RollingFileLoggerProviderTests.cs b/src/Netclaw.Daemon.Tests/Configuration/RollingFileLoggerProviderTests.cs index 8d54108be..f25c6507d 100644 --- a/src/Netclaw.Daemon.Tests/Configuration/RollingFileLoggerProviderTests.cs +++ b/src/Netclaw.Daemon.Tests/Configuration/RollingFileLoggerProviderTests.cs @@ -3,138 +3,42 @@ // Copyright (C) 2026 - 2026 Petabridge, LLC // // ----------------------------------------------------------------------- -using Akka.Actor; -using Akka.Hosting; -using Akka.Hosting.TestKit; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Time.Testing; -using Netclaw.Actors.Protocol; -using Netclaw.Configuration; using Netclaw.Daemon.Configuration; using Xunit; namespace Netclaw.Daemon.Tests.Configuration; -public sealed class RollingFileLoggerProviderTests : TestKit, IDisposable +public sealed class RollingFileLoggerProviderTests : IDisposable { - protected override void ConfigureAkka(AkkaConfigurationBuilder builder, IServiceProvider provider) - { - } - - private readonly string _basePath = Path.Combine(Path.GetTempPath(), $"netclaw-rolling-logger-tests-{Guid.NewGuid():N}"); - - [Fact] - public async Task Session_scoped_log_routes_diagnostic_to_dispatcher() - { - var timeProvider = new FakeTimeProvider(DateTimeOffset.Parse("2026-05-07T12:34:56Z")); - var daemonLogPath = Path.Combine(_basePath, "logs", "daemon.log"); - Directory.CreateDirectory(Path.GetDirectoryName(daemonLogPath)!); - var probe = CreateTestProbe(); - - using (var provider = new RollingFileLoggerProvider(daemonLogPath, timeProvider)) - { - provider.AttachSessionDispatcher(Task.FromResult(probe.Ref)); - var logger = provider.CreateLogger("Netclaw.Tests"); - - using (SessionDiagnosticsContext.Push("channel/thread")) - { - logger.LogInformation("session scoped message"); - } - - var diagnostic = await probe.ExpectMsgAsync( - cancellationToken: TestContext.Current.CancellationToken); - Assert.Equal("channel/thread", diagnostic.SessionId.Value); - Assert.Contains("session scoped message", diagnostic.Line, StringComparison.Ordinal); - Assert.Contains("Diagnostic:", diagnostic.Line, StringComparison.Ordinal); - } - } + private readonly string _basePath = Path.Join(Path.GetTempPath(), $"netclaw-rolling-logger-tests-{Guid.NewGuid():N}"); [Fact] - public async Task Daemon_scoped_log_does_not_route_to_dispatcher() + public async Task Writes_log_lines_to_the_daily_rolling_daemon_log() { - var timeProvider = new FakeTimeProvider(DateTimeOffset.Parse("2026-05-07T12:34:56Z")); - var daemonLogPath = Path.Combine(_basePath, "logs", "daemon.log"); + var time = new FakeTimeProvider(DateTimeOffset.Parse("2026-05-07T12:34:56Z")); + var daemonLogPath = Path.Join(_basePath, "logs", "daemon.log"); Directory.CreateDirectory(Path.GetDirectoryName(daemonLogPath)!); - var probe = CreateTestProbe(); - using (var provider = new RollingFileLoggerProvider(daemonLogPath, timeProvider)) + // Dispose drains the writer thread, so the line is flushed by the time we read. + using (var provider = new RollingFileLoggerProvider(daemonLogPath, time)) { - provider.AttachSessionDispatcher(Task.FromResult(probe.Ref)); var logger = provider.CreateLogger("Netclaw.Tests"); - logger.LogInformation("daemon message"); - - await probe.ExpectNoMsgAsync( - TimeSpan.FromMilliseconds(200), - cancellationToken: TestContext.Current.CancellationToken); + logger.LogInformation("hello daemon log {SessionId}", "channel/thread"); } - var daemonLog = Directory.GetFiles(Path.Combine(_basePath, "logs"), "daemon-*.log", SearchOption.TopDirectoryOnly).Single(); - var daemonText = await File.ReadAllTextAsync(daemonLog, TestContext.Current.CancellationToken); - Assert.Contains("daemon message", daemonText, StringComparison.Ordinal); - } - - [Fact] - public async Task Pre_resolution_diagnostics_buffer_and_drain_to_dispatcher() - { - var timeProvider = new FakeTimeProvider(DateTimeOffset.Parse("2026-05-07T12:34:56Z")); - var daemonLogPath = Path.Combine(_basePath, "logs", "daemon.log"); - Directory.CreateDirectory(Path.GetDirectoryName(daemonLogPath)!); - var probe = CreateTestProbe(); - var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - - using (var provider = new RollingFileLoggerProvider(daemonLogPath, timeProvider)) - { - provider.AttachSessionDispatcher(tcs.Task); - var logger = provider.CreateLogger("Netclaw.Tests"); - - using (SessionDiagnosticsContext.Push("channel/thread")) - { - logger.LogInformation("pre-resolution one"); - logger.LogInformation("pre-resolution two"); - } - - await probe.ExpectNoMsgAsync( - TimeSpan.FromMilliseconds(100), - cancellationToken: TestContext.Current.CancellationToken); - - tcs.SetResult(probe.Ref); - - var first = await probe.ExpectMsgAsync( - cancellationToken: TestContext.Current.CancellationToken); - var second = await probe.ExpectMsgAsync( - cancellationToken: TestContext.Current.CancellationToken); - Assert.Contains("pre-resolution one", first.Line, StringComparison.Ordinal); - Assert.Contains("pre-resolution two", second.Line, StringComparison.Ordinal); - } - } - - [Fact] - public async Task Diagnostic_routes_correctly_when_logged_from_async_continuation() - { - var timeProvider = new FakeTimeProvider(DateTimeOffset.Parse("2026-05-07T12:34:56Z")); - var daemonLogPath = Path.Combine(_basePath, "logs", "daemon.log"); - Directory.CreateDirectory(Path.GetDirectoryName(daemonLogPath)!); - var probe = CreateTestProbe(); - - using (var provider = new RollingFileLoggerProvider(daemonLogPath, timeProvider)) - { - provider.AttachSessionDispatcher(Task.FromResult(probe.Ref)); - var logger = provider.CreateLogger("Netclaw.Tests"); - - using (SessionDiagnosticsContext.Push("channel/thread")) - { - await Task.Yield(); - logger.LogInformation("post-await"); - } - - var diagnostic = await probe.ExpectMsgAsync( - cancellationToken: TestContext.Current.CancellationToken); - Assert.Equal("channel/thread", diagnostic.SessionId.Value); - Assert.Contains("post-await", diagnostic.Line, StringComparison.Ordinal); - } + // No dispatcher was attached, so per-session routing is off and even a session-tagged + // line falls back to daemon.log. (When routing IS attached the {SessionId} line is + // partitioned to session.log — see RollingFileLoggerPartitionTests.) + var daemonLog = Directory.GetFiles(Path.Join(_basePath, "logs"), "daemon-*.log").Single(); + var text = await File.ReadAllTextAsync(daemonLog, TestContext.Current.CancellationToken); + Assert.Contains("hello daemon log", text, StringComparison.Ordinal); + Assert.Contains("Netclaw.Tests", text, StringComparison.Ordinal); + Assert.Contains("channel/thread", text, StringComparison.Ordinal); } - void IDisposable.Dispose() + public void Dispose() { try { diff --git a/src/Netclaw.Daemon.Tests/Configuration/RoutingChatClientTests.cs b/src/Netclaw.Daemon.Tests/Configuration/RoutingChatClientTests.cs index 4670c0904..00e3db65f 100644 --- a/src/Netclaw.Daemon.Tests/Configuration/RoutingChatClientTests.cs +++ b/src/Netclaw.Daemon.Tests/Configuration/RoutingChatClientTests.cs @@ -6,6 +6,7 @@ using System.Runtime.CompilerServices; using Microsoft.Extensions.AI; using Microsoft.Extensions.Logging.Abstractions; +using Netclaw.Actors.Sessions; using Netclaw.Configuration; using Netclaw.Daemon.Configuration; using Xunit; @@ -80,6 +81,56 @@ await Assert.ThrowsAsync(() => Assert.DoesNotContain(sink.Alerts, a => a.Category == AlertType.ProviderFailover); } + [Fact] + public async Task Failover_log_carries_SessionId_scope_from_options() + { + // Provider failover/outage is logged outside any per-pipeline LoggingChatClient + // scope, so RoutingChatClient must attach the session id (from the call options) + // itself — otherwise outages are uncorrelatable to the affected session in Seq. + var sink = new CapturingSink(); + var logger = new ScopeCapturingLogger(); + var primary = new FakeChatClient((_, _, _) => throw new HttpRequestException("primary down")); + var fallback = new FakeChatClient((_, _, _) => + Task.FromResult(new ChatResponse([new ChatMessage(ChatRole.Assistant, "fallback")]))); + var client = new RoutingChatClient( + new StubRouter([primary, fallback]), + new ChatRoutingContext { Role = ModelRole.Main }, + sink, logger, TimeProvider.System); + + await client.GetResponseAsync( + [new ChatMessage(ChatRole.User, "hi")], + new SessionScopedChatOptions { SessionId = "ch/thread" }, + TestContext.Current.CancellationToken); + + Assert.True(logger.HasSessionScope("ch/thread")); + } + + [Fact] + public async Task Streaming_failover_log_carries_SessionId_scope_from_options() + { + // Production takes the streaming failover path (StreamingResponseReader), so the + // streaming branch — not the non-streaming one above — is the one that must tag + // provider-failover/outage warnings with the session id. + var sink = new CapturingSink(); + var logger = new ScopeCapturingLogger(); + var primary = new FakeChatClient(streamHandler: (_, _, ct) => ThrowBeforeFirstChunkAsync(true, ct)); + var fallback = new FakeChatClient(streamHandler: (_, _, ct) => SingleTextUpdateAsync("fallback", ct)); + var client = new RoutingChatClient( + new StubRouter([primary, fallback]), + new ChatRoutingContext { Role = ModelRole.Main }, + sink, logger, TimeProvider.System); + + await foreach (var _ in client.GetStreamingResponseAsync( + [new ChatMessage(ChatRole.User, "hi")], + new SessionScopedChatOptions { SessionId = "ch/thread" }, + TestContext.Current.CancellationToken)) + { + // Drain the stream to completion so the pipeline (scope/retry/logging) runs; updates aren't asserted here. + } + + Assert.True(logger.HasSessionScope("ch/thread")); + } + [Fact] public async Task DoesNotFailover_OnCancellation() { @@ -204,7 +255,7 @@ public async Task Streaming_DoesNotFailover_OnCancellation() await Assert.ThrowsAnyAsync(async () => { await foreach (var _ in Client(sink, primary, fallback).GetStreamingResponseAsync( - [new ChatMessage(ChatRole.User, "hi")], cancellationToken: cts.Token)) { } + [new ChatMessage(ChatRole.User, "hi")], cancellationToken: cts.Token)) { /* drain the stream so the pipeline runs */ } }); Assert.Equal(0, fallbackCalls); diff --git a/src/Netclaw.Daemon.Tests/Configuration/ScopeCapturingLogger.cs b/src/Netclaw.Daemon.Tests/Configuration/ScopeCapturingLogger.cs index 2b93e2bba..075e71bf1 100644 --- a/src/Netclaw.Daemon.Tests/Configuration/ScopeCapturingLogger.cs +++ b/src/Netclaw.Daemon.Tests/Configuration/ScopeCapturingLogger.cs @@ -4,6 +4,7 @@ // // ----------------------------------------------------------------------- using Microsoft.Extensions.Logging; +using Netclaw.Actors.Protocol; namespace Netclaw.Daemon.Tests.Configuration; @@ -15,8 +16,6 @@ namespace Netclaw.Daemon.Tests.Configuration; /// internal sealed class ScopeCapturingLogger : ILogger { - private const string SessionIdKey = "SessionId"; - public List Scopes { get; } = []; public void Log( @@ -36,10 +35,10 @@ public void Log( /// True if a scope tagging the given session id was opened. public bool HasSessionScope(string expectedId) => Scopes.Any(s => s is IEnumerable> kvps - && kvps.Any(kv => kv.Key == SessionIdKey && kv.Value is string v && v == expectedId)); + && kvps.Any(kv => kv.Key == NetclawLogProperties.SessionId && kv.Value is string v && v == expectedId)); /// True if any session-id scope (regardless of value) was opened. public bool HasAnySessionScope() => Scopes.Any(s => s is IEnumerable> kvps - && kvps.Any(kv => kv.Key == SessionIdKey)); + && kvps.Any(kv => kv.Key == NetclawLogProperties.SessionId)); } diff --git a/src/Netclaw.Daemon.Tests/Configuration/SessionLogPartitionIntegrationTests.cs b/src/Netclaw.Daemon.Tests/Configuration/SessionLogPartitionIntegrationTests.cs new file mode 100644 index 000000000..4323ef9a8 --- /dev/null +++ b/src/Netclaw.Daemon.Tests/Configuration/SessionLogPartitionIntegrationTests.cs @@ -0,0 +1,122 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using Akka.Actor; +using Akka.Event; +using Akka.Hosting; +using Akka.Hosting.TestKit; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Time.Testing; +using Netclaw.Actors.Hosting; +using Netclaw.Actors.Protocol; +using Netclaw.Daemon.Configuration; +using Xunit; + +namespace Netclaw.Daemon.Tests.Configuration; + +/// +/// End-to-end proof through the REAL Akka→MEL bridge: an actor that tags its logger with +/// WithContext("SessionId", …) has its line partitioned into that session's +/// session.log (not daemon.log) by . This +/// exercises the live AkkaLogState shape that the unit tests can only simulate, plus the +/// real SessionLogDispatcher and SessionLogActor file write. +/// +public sealed class SessionLogPartitionIntegrationTests : TestKit +{ + private readonly string _logDir = Path.Join(Path.GetTempPath(), $"netclaw-logpart-int-{Guid.NewGuid():N}"); + private readonly FakeTimeProvider _time = new(DateTimeOffset.Parse("2026-05-07T12:00:00Z")); + private RollingFileLoggerProvider _provider = null!; + + private string DaemonPath => Path.Join(_logDir, "daemon.log"); + private string SessionsDir => Path.Join(_logDir, "sessions"); + + protected override void ConfigureAkka(AkkaConfigurationBuilder builder, IServiceProvider provider) + { + builder + .ConfigureLoggers(setup => + { + setup.ClearLoggers(); + setup.AddLoggerFactory(); + setup.LogLevel = Akka.Event.LogLevel.DebugLevel; + }) + .WithSessionLogDispatcher(SessionsDir, _time); + } + + protected override void ConfigureServices(HostBuilderContext context, IServiceCollection services) + { + Directory.CreateDirectory(_logDir); + _provider = new RollingFileLoggerProvider(DaemonPath, _time); + // Wire our provider into the host's ILoggerFactory so AddLoggerFactory bridges every + // Akka actor log to it — the same path production uses. + services.AddSingleton(_provider); + } + + [Fact] + public async Task Actor_log_tagged_with_session_context_lands_in_session_log_not_daemon_log() + { + // Attach the provider to the real dispatcher, as SessionLogDispatcherWiringService does. + var dispatcher = ActorRegistry.Get(); + _provider.AttachSessionDispatcher(Task.FromResult(dispatcher)); + + var sessionId = new SessionId("intgr-channel/intgr-thread"); + const string marker = "hello from a real actor (partition integration)"; + + var probe = Sys.ActorOf(Props.Create(() => new SessionTaggedLogger()), "session-tagged-logger"); + probe.Tell(new LogIt(sessionId.Value, marker)); + + var sessionLogPath = SessionLogFile.GetLogPath(sessionId, SessionsDir); + await AwaitAssertAsync( + async () => + { + Assert.True(File.Exists(sessionLogPath), "session.log was not created for the tagged actor log"); + var text = await ReadSharedAsync(sessionLogPath, TestContext.Current.CancellationToken); + Assert.Contains(marker, text, StringComparison.Ordinal); + }, + TimeSpan.FromSeconds(10), + cancellationToken: TestContext.Current.CancellationToken); + + // Partition: the session-tagged line is NOT in daemon.log (daemon writes flush per line). + var daemonFiles = Directory.GetFiles(_logDir, "daemon-*.log"); + var daemonText = daemonFiles.Length == 0 + ? string.Empty + : await ReadSharedAsync(daemonFiles[0], TestContext.Current.CancellationToken); + Assert.DoesNotContain(marker, daemonText, StringComparison.Ordinal); + + TryCleanup(); + } + + private static async Task ReadSharedAsync(string path, CancellationToken ct) + { + await using var stream = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.ReadWrite | FileShare.Delete); + using var reader = new StreamReader(stream); + return await reader.ReadToEndAsync(ct); + } + + private void TryCleanup() + { + try + { + if (Directory.Exists(_logDir)) + Directory.Delete(_logDir, recursive: true); + } + catch (IOException ex) + { + Console.Error.WriteLine($"[SessionLogPartitionIntegrationTests] cleanup failed: {ex.Message}"); + } + } + + private sealed record LogIt(string SessionId, string Message); + + private sealed class SessionTaggedLogger : ReceiveActor + { + public SessionTaggedLogger() + { + Receive(m => + Context.GetLogger().WithContext("SessionId", m.SessionId).Info(m.Message)); + } + } +} diff --git a/src/Netclaw.Daemon/Configuration/ChatClientSessionScope.cs b/src/Netclaw.Daemon/Configuration/ChatClientSessionScope.cs new file mode 100644 index 000000000..baa97bb43 --- /dev/null +++ b/src/Netclaw.Daemon/Configuration/ChatClientSessionScope.cs @@ -0,0 +1,48 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using Microsoft.Extensions.AI; +using Microsoft.Extensions.Logging; +using Netclaw.Actors.Protocol; +using Netclaw.Actors.Sessions; + +namespace Netclaw.Daemon.Configuration; + +/// +/// Opens a logging scope tagging a chat-client decorator's log lines with the owning +/// SessionId, read from on the call's +/// . The OTLP exporter has IncludeScopes enabled, so the +/// scope surfaces SessionId as a filterable attribute in Seq — matching the +/// WithContext("SessionId", …) key the session/channel actors already use, so +/// LLM-pipeline diagnostics (timing, retries, provider failover/outage) correlate to the +/// same session as the actor logs. +/// +/// The decorators are session-agnostic singletons, so the id must arrive on the call +/// seam: this reads it from the options object rather than ambient context (which would +/// not flow across the actor mailbox boundary). Returns null — a no-op +/// using — for sidecar/session-agnostic calls that carry a plain +/// . +/// +internal static class ChatClientSessionScope +{ + public static IDisposable? Begin(ILogger logger, ChatOptions? options) + { + if (options is not SessionScopedChatOptions { SessionId: { Length: > 0 } sessionId } scoped) + return null; + + // A sub-agent's call also carries its SubSessionId, so the file-logger routes the line + // into the sub-agent's own session.log while SessionId keeps OTEL grouping under the parent. + return string.IsNullOrWhiteSpace(scoped.SubSessionId) + ? logger.BeginScope(new[] + { + new KeyValuePair(NetclawLogProperties.SessionId, sessionId), + }) + : logger.BeginScope(new[] + { + new KeyValuePair(NetclawLogProperties.SessionId, sessionId), + new KeyValuePair(NetclawLogProperties.SubSessionId, scoped.SubSessionId), + }); + } +} diff --git a/src/Netclaw.Daemon/Configuration/LoggingChatClient.cs b/src/Netclaw.Daemon/Configuration/LoggingChatClient.cs index ffa618b33..dcf6c75e9 100644 --- a/src/Netclaw.Daemon/Configuration/LoggingChatClient.cs +++ b/src/Netclaw.Daemon/Configuration/LoggingChatClient.cs @@ -8,15 +8,19 @@ using System.Text; using Microsoft.Extensions.AI; using Microsoft.Extensions.Logging; +using Netclaw.Actors.Sessions; namespace Netclaw.Daemon.Configuration; /// -/// Decorates an with logging for elapsed time, token -/// usage, and errors, and tags log lines with the ambient session id. Stateless -/// and safe to share across sessions. Netclaw issues only streaming requests, so -/// only the streaming path is instrumented; the inherited non-streaming -/// pass-through is unused. +/// Decorates an with logging for elapsed time, token usage, and +/// errors. The decorator is session-agnostic but tags each line with the owning +/// SessionId via a logging scope — read from on +/// the call (). The file-logger partitions scoped lines into +/// that session's session.log (and they carry SessionId as a Seq/OTLP field); a +/// call with no session scope falls through to daemon.log. Stateless and safe to share +/// across sessions. Netclaw issues only streaming requests, so only the streaming path is +/// instrumented; the inherited non-streaming pass-through is unused. /// public sealed class LoggingChatClient : DelegatingChatClient { @@ -39,7 +43,9 @@ public override async IAsyncEnumerable GetStreamingResponseA [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default) { - using var sessionScope = SessionLoggingScope.Begin(_logger); + // Tag every line this call emits (prompt summary, timing, token usage, errors) + // with the owning session so they correlate in Seq. No-op for sidecar calls. + using var sessionScope = ChatClientSessionScope.Begin(_logger, options); var messageList = messages as IReadOnlyList ?? messages.ToList(); LogPromptDiagnostics(messageList, options); diff --git a/src/Netclaw.Daemon/Configuration/LoggingRegistrationExtensions.cs b/src/Netclaw.Daemon/Configuration/LoggingRegistrationExtensions.cs index d1ee4633f..baef41848 100644 --- a/src/Netclaw.Daemon/Configuration/LoggingRegistrationExtensions.cs +++ b/src/Netclaw.Daemon/Configuration/LoggingRegistrationExtensions.cs @@ -25,14 +25,14 @@ public static LogLevel ConfigureNetclawLogging(this WebApplicationBuilder builde if (consoleEnabled) builder.Logging.AddSimpleConsole(options => options.SingleLine = true); - // Always write to a rolling log file in ~/.netclaw/logs/ + // This provider owns the local partition of the log stream: session-tagged lines go + // to per-session session.log files, everything else to daemon.log (see + // RollingFileLoggerProvider). It must be constructed eagerly so MEL sees it via + // AddProvider — a Services.AddSingleton(factory) registration here is + // not picked up by the LoggerFactory in this hosting setup. The session-log dispatcher + // is wired in post-build by SessionLogDispatcherWiringService once Akka.Hosting has + // registered the actor system. Directory.CreateDirectory(resolvedPaths.LogsDirectory); - - // Provider must be constructed eagerly so MEL can see it via AddProvider — - // a Services.AddSingleton(factory) registration here is - // not picked up by the LoggerFactory in this hosting setup. The session - // log dispatcher is wired in post-build by SessionLogDispatcherWiringService - // once Akka.Hosting has registered the actor system. var provider = new RollingFileLoggerProvider(resolvedPaths.DaemonLogPath); builder.Logging.AddProvider(provider); builder.Services.AddSingleton(provider); @@ -53,8 +53,9 @@ private static LogLevel ResolveLogLevel(IConfiguration configuration) } /// -/// Hooks the session log dispatcher into -/// once Akka.Hosting has registered SessionLogDispatcherActorKey. +/// Hooks the session-log dispatcher into once +/// Akka.Hosting has registered SessionLogDispatcherActorKey. The provider is built during +/// host construction (before Akka), so the dispatcher can only be attached post-start. /// internal sealed class SessionLogDispatcherWiringService : IHostedService { diff --git a/src/Netclaw.Daemon/Configuration/RetryingChatClient.cs b/src/Netclaw.Daemon/Configuration/RetryingChatClient.cs index 2a658e9bd..2ea0d8c4e 100644 --- a/src/Netclaw.Daemon/Configuration/RetryingChatClient.cs +++ b/src/Netclaw.Daemon/Configuration/RetryingChatClient.cs @@ -122,14 +122,23 @@ public override async IAsyncEnumerable GetStreamingResponseA private async Task BackoffAsync(Exception ex, int attempt, CancellationToken cancellationToken) { var delay = _policy.GetDelay(attempt); - // Retry is the hottest failure path; tag the warning with the session id (like - // the other chat-client decorators) so retry storms correlate by session in Seq. - using (SessionLoggingScope.Begin(_logger)) - { - _logger.LogWarning(ex, - "LLM call failed (attempt {Attempt}/{Max}), retrying in {Delay:F1}s", - attempt + 1, _policy.MaxRetries, delay.TotalSeconds); - } + // No SessionId scope is opened here: PipelineChatClientFactory.Compose always + // wraps this decorator inside LoggingChatClient (guarded by + // Compose_puts_Logging_outermost), whose streaming scope stays open for the whole + // enumeration that drives this retry loop — so the warning already inherits the + // session id. Re-opening an identical scope would be pure duplication. + // + // CAVEAT — this inheritance holds ONLY on the streaming path. LoggingChatClient + // instruments streaming only; its inherited non-streaming GetResponseAsync opens no + // scope. Netclaw issues only streaming requests today, so the non-streaming retry + // path above is unreachable — but if that ever changes, these retry warnings would + // carry no SessionId and the file-logger would route them to daemon.log instead of + // the owning session.log (and they'd be uncorrelated in Seq). The fix at that point + // is to open a SessionId scope here from `options` (ChatClientSessionScope.Begin), + // or to instrument LoggingChatClient.GetResponseAsync. See RetryingChatClientTests. + _logger.LogWarning(ex, + "LLM call failed (attempt {Attempt}/{Max}), retrying in {Delay:F1}s", + attempt + 1, _policy.MaxRetries, delay.TotalSeconds); await Task.Delay(delay, _timeProvider, cancellationToken); } diff --git a/src/Netclaw.Daemon/Configuration/RollingFileLoggerProvider.cs b/src/Netclaw.Daemon/Configuration/RollingFileLoggerProvider.cs index d24755418..80b75e17a 100644 --- a/src/Netclaw.Daemon/Configuration/RollingFileLoggerProvider.cs +++ b/src/Netclaw.Daemon/Configuration/RollingFileLoggerProvider.cs @@ -8,40 +8,51 @@ using Akka.Actor; using Microsoft.Extensions.Logging; using Netclaw.Actors.Protocol; -using Netclaw.Configuration; +using Netclaw.Actors.Sessions; namespace Netclaw.Daemon.Configuration; /// -/// Simple file-based logger that writes to a daily rolling log file. -/// Uses a background queue to avoid blocking callers. -/// -/// Session-scoped lines (emitted under a populated -/// ) are mirrored to a per-session -/// session.log by routing through the SessionLogDispatcher -/// actor. The dispatcher serializes all writes for a given session through -/// a single mailbox, replacing the in-process file lock that previously -/// coordinated concurrent writers. -/// -/// The dispatcher is wired in via -/// post-construction (typically from an IHostedService that runs -/// after the actor system starts) because the provider is constructed -/// during host build, before Akka. +/// File-based logger that owns the LOCAL partition of the one log stream. Every line is +/// written to exactly one place on disk: +/// +/// A line tagged with a session id by session-serving code is routed to that +/// session's session.log via the SessionLogDispatcher and is NOT written to +/// daemon.log. "Serving" is the key: the id must ride in on an actor context — an +/// actor's WithContext("SessionId", …), which the Akka→MEL bridge surfaces alongside a +/// LogSource — or a BeginScope (the chat-client decorators and spawn breadcrumbs). +/// Everything else goes to daemon.log: genuinely daemon-wide lines (startup, +/// config, session lifecycle, global errors) AND a daemon-infrastructure line that merely +/// names a session in its message template (a bare {SessionId} field with no actor +/// context — e.g. the gateway/catalog/drain "failed to … session X" warnings). Those are daemon +/// functionality, not the session's own work, so they stay where an operator triages the daemon. +/// +/// The full stream still reaches OTEL via the separate OTLP exporter (with the session id as +/// an attribute); the OTEL receiver does the global slicing/distilling. The dispatcher is +/// attached post-construction via (the provider is built +/// during host build, before Akka). Lines emitted by the itself +/// are forced to daemon.log so a write-failure log can never recurse into the file that +/// just failed. /// -internal sealed class RollingFileLoggerProvider : ILoggerProvider +internal sealed class RollingFileLoggerProvider : ILoggerProvider, ISupportExternalScope { private const long MaxFileSizeBytes = 10 * 1024 * 1024; // 10MB per file - private const int PreResolutionBufferLimit = 1000; + private const int DaemonLogFlushBatch = 256; // flush cap under a burst; idle flushes every line + private const long RollFlushMarginBytes = 128 * 1024; // flush near the size cap so rolls aren't late private readonly string _basePath; private readonly TimeProvider _timeProvider; private readonly ConcurrentDictionary _loggers = new(); private readonly BlockingCollection _queue = new(1024); private readonly Thread _writerThread; - private ConcurrentQueue? _pendingDiagnostics; + private IExternalScopeProvider? _scopeProvider; + + // The dispatcher reference is the only shared state on the routing path: null until + // IRequiredActor.GetAsync resolves it, then written once (Volatile). No buffer/lock — a line + // that finds it still null (the brief startup window, or a resolution failure) falls back to + // daemon.log, which is the documented routing-off behavior. private IActorRef? _sessionDispatcher; - private int _pendingCount; - private int _sessionRoutingEnabled; + private int _attached; private StreamWriter? _writer; private string _currentDate = ""; @@ -60,97 +71,155 @@ public RollingFileLoggerProvider(string basePath, TimeProvider? timeProvider = n public ILogger CreateLogger(string categoryName) => _loggers.GetOrAdd(categoryName, name => new RollingFileLogger(name, this)); + // MEL hands every scope-aware provider the same shared scope provider; the chat-client + // decorators carry the session id via BeginScope, so we read it from here at emit time. + public void SetScopeProvider(IExternalScopeProvider scopeProvider) => _scopeProvider = scopeProvider; + /// - /// Enables session-scoped log routing through the dispatcher actor. - /// The provider buffers session-scoped diagnostic lines emitted between - /// this call and dispatcher resolution into a small bounded queue, then - /// drains them in order. Subsequent emits go straight to the dispatcher. - /// Failures during resolution surface a single ERR line in the daemon log - /// and disable session routing for the rest of the process. + /// Resolves the per-session log dispatcher in the background (the caller passes + /// IRequiredActor<SessionLogDispatcherActorKey>.GetAsync()) and publishes it. + /// Resolving in the background — rather than blocking — keeps this safe to call from a hosted + /// service that starts before Akka. Session-tagged lines logged before it resolves (a brief + /// startup window in which session actors do not yet exist) fall back to daemon.log, as + /// do all session-tagged lines if resolution fails (a single ERR beacon records that). /// public void AttachSessionDispatcher(Task dispatcherTask) { - if (Interlocked.Exchange(ref _sessionRoutingEnabled, 1) == 1) + if (Interlocked.Exchange(ref _attached, 1) == 1) return; - _pendingDiagnostics = new ConcurrentQueue(); _ = ResolveSessionDispatcherAsync(dispatcherTask); } - internal void Enqueue(string message) + /// + /// Partitions one already-formatted line: routes it to its session's session.log when + /// it carries a session id (and is not the session-log writer's own line) and the dispatcher + /// is resolved; otherwise writes it to daemon.log. is + /// the id found on the log event's state; the active scopes are consulted only as a fallback. + /// + internal void Route(string line, string? stateSessionId, string? stateSubSessionId, bool fromSessionLogActor) { - _queue.TryAdd(message); - - if (_sessionRoutingEnabled == 0) + // Carve-out: the session-log writer's own lines go to daemon.log so a failed write that + // logs an error cannot route back into the same (failing) session.log — an infinite loop. + if (fromSessionLogActor) + { + _queue.TryAdd(line); return; + } - var sessionId = SessionDiagnosticsContext.SessionId; - if (string.IsNullOrWhiteSpace(sessionId)) - return; + var sessionId = stateSessionId; + var subSessionId = stateSubSessionId; + if (sessionId is null) + { + // Only chat-client lines lack a state session id; they carry both ids via BeginScope. + // An actor line already names its ids in state, so we never consult scopes for it — + // otherwise an unrelated ambient SubSessionId scope could hijack its routing. + FindScopeIds(out var scopeSessionId, out var scopeSubSessionId); + sessionId = scopeSessionId; + subSessionId ??= scopeSubSessionId; + } - var dispatcher = Volatile.Read(ref _sessionDispatcher); - if (dispatcher is null && Volatile.Read(ref _pendingCount) >= PreResolutionBufferLimit) + // Partition the LOCAL file: a sub-agent's lines (which carry a SubSessionId) get their own + // session.log keyed by the sub-session; everything else goes to its session's file. The + // line still carries the parent SessionId, so OTEL groups it under the parent regardless. + var routingId = subSessionId ?? sessionId; + if (!string.IsNullOrWhiteSpace(routingId) && Volatile.Read(ref _sessionDispatcher) is { } dispatcher) + { + dispatcher.Tell(new SessionLogDiagnostic(new SessionId(routingId), line)); return; + } + + // No session id, the dispatcher hasn't resolved yet (startup window), or resolution + // failed → daemon.log. + _queue.TryAdd(line); + } - var diagnostic = new SessionLogDiagnostic( - new SessionId(sessionId), - $"[{_timeProvider.GetUtcNow():o}] Diagnostic: {message}"); + private void FindScopeIds(out string? sessionId, out string? subSessionId) + { + sessionId = null; + subSessionId = null; - if (dispatcher is not null) - { - dispatcher.Tell(diagnostic); + var scopeProvider = _scopeProvider; + if (scopeProvider is null) return; - } - Interlocked.Increment(ref _pendingCount); - _pendingDiagnostics!.Enqueue(diagnostic); + // static lambda + a single mutable holder: the delegate is cached and nothing is + // captured, so the chat-client diagnostic hot path doesn't allocate a closure per line. + var ids = new ScopeIds(); + scopeProvider.ForEachScope( + static (scope, state) => + { + if (state.SessionId is not null && state.SubSessionId is not null) + return; + + if (scope is IEnumerable> kvps) + { + foreach (var kv in kvps) + { + if (state.SessionId is null + && kv.Key == NetclawLogProperties.SessionId + && kv.Value?.ToString() is { Length: > 0 } id) + state.SessionId = id; + else if (state.SubSessionId is null + && kv.Key == NetclawLogProperties.SubSessionId + && kv.Value?.ToString() is { Length: > 0 } subId) + state.SubSessionId = subId; + } + } + }, + ids); + + sessionId = ids.SessionId; + subSessionId = ids.SubSessionId; + } + + private sealed class ScopeIds + { + public string? SessionId; + public string? SubSessionId; } private async Task ResolveSessionDispatcherAsync(Task dispatcherTask) { - IActorRef dispatcher; try { - dispatcher = await dispatcherTask.ConfigureAwait(false); + // Publish once. After this, every Route fast-path reads it and Tells directly. + Volatile.Write(ref _sessionDispatcher, await dispatcherTask.ConfigureAwait(false)); } catch (Exception ex) { - // Resolution failed permanently. Drop the buffer and switch the - // session-log path off — daemon-global logging continues normally, - // but session-scoped diagnostics will not appear in session.log - // for the remainder of this process. Surface a single loud line - // so operators see this in the daemon log instead of silently - // accumulating drops. - _queue.TryAdd($"{GetTimestamp()} [ERR] Netclaw.Logging: session log dispatcher resolution failed; session-scoped diagnostics disabled. {ex.Message}"); - _pendingDiagnostics = null; - return; - } - - // Publish the ref BEFORE draining so producers racing with the drainer - // see the dispatcher and Tell directly rather than enqueueing into a - // queue that we are about to abandon. - Volatile.Write(ref _sessionDispatcher, dispatcher); - - while (_pendingDiagnostics!.TryDequeue(out var pending)) - { - Interlocked.Decrement(ref _pendingCount); - dispatcher.Tell(pending); + // Resolution failed: _sessionDispatcher stays null, so session-tagged lines keep + // falling back to daemon.log. One loud beacon, with a stderr fallback so the single + // failure signal survives even a saturated daemon-log queue. + var beacon = $"{GetTimestamp()} [ERR] Netclaw.Logging: session log dispatcher resolution failed; per-session routing disabled. {ex.Message}"; + if (!_queue.TryAdd(beacon)) + Console.Error.WriteLine(beacon); } } private void ProcessQueue() { + var batched = 0; foreach (var message in _queue.GetConsumingEnumerable()) { try { EnsureWriter(); _writer!.WriteLine(message); - _writer.Flush(); + + // Flush when the queue has drained — so sparse daemon.log lines (startup, config, + // lifecycle, alerts) stay immediately durable — or after a burst batch, so a + // sustained burst doesn't pay an fsync per line. The final tail is flushed by + // Dispose() when the writer thread exits. + if (_queue.Count == 0 || ++batched >= DaemonLogFlushBatch) + { + _writer.Flush(); + batched = 0; + } } catch (Exception ex) { - // Last-resort: write to stderr to avoid silent swallow + // Last-resort: write to stderr to avoid silent swallow. Console.Error.WriteLine($"[NetclawLogWriter] Failed to write log: {ex.Message}"); } } @@ -161,6 +230,13 @@ private void EnsureWriter() var today = _timeProvider.GetUtcNow().ToString("yyyy-MM-dd", CultureInfo.InvariantCulture); if (_writer is not null && _currentDate == today) { + // With AutoFlush off (batched writes), BaseStream.Length lags the buffered bytes. Flush + // once we're within a batch of the cap so the roll decision sees the true size — paying + // the fsync only near the threshold, not per line — then the file can't overshoot 10MB + // by a batch's worth of buffered data. + if (_writer.BaseStream.Length >= MaxFileSizeBytes - RollFlushMarginBytes) + _writer.Flush(); + // Roll if file exceeds size limit if (_writer.BaseStream.Length >= MaxFileSizeBytes) { @@ -217,6 +293,13 @@ public void Log(LogLevel logLevel, EventId eventId, TState state, Except if (!IsEnabled(logLevel)) return; + // Single pass over the event's structured state: pick up the session id and (for a + // sub-agent's lines) the sub-session id (for routing), the Akka log source (for a useful + // per-line label — every actor shares the generic MEL category "Akka.Actor.ActorSystem"), + // and whether this line is the session-log writer's own (so we never route it back into + // the file it writes). + ScanState(state, out var sessionId, out var subSessionId, out var logSource, out var fromSessionLogActor); + var timestamp = _provider.GetTimestamp(); var level = logLevel switch { @@ -227,11 +310,77 @@ public void Log(LogLevel logLevel, EventId eventId, TState state, Except _ => "DBG" }; + var source = logSource ?? _category; var message = formatter(state, exception); - var line = $"{timestamp} [{level}] {_category}: {message}"; + var line = $"{timestamp} [{level}] {source}: {message}"; if (exception is not null) line += Environment.NewLine + exception; - _provider.Enqueue(line); + // A session id found in the message STATE is a routing trigger only when it rode in on an + // Akka actor context: the Akka→MEL bridge's AkkaLogState carries a LogSource alongside the + // WithContext("SessionId", …). A bare {SessionId} message field on a plain daemon-service + // ILogger (gateway/catalog/drain) has no LogSource — that is daemon infrastructure + // merely naming a session, so it stays in daemon.log. Session-serving non-actor producers + // (chat-client decorators, spawn breadcrumbs) tag via BeginScope instead, which Route still + // consults as a fallback. + var stateSessionId = logSource is not null ? sessionId : null; + var stateSubSessionId = logSource is not null ? subSessionId : null; + + _provider.Route(line, stateSessionId, stateSubSessionId, fromSessionLogActor); } + + // Read the fields the producer already put on the log event. The Akka→MEL bridge passes an + // AkkaLogState carrying WithContext("SessionId", …) plus "LogSource"/"ActorPath"; MEL's own + // structured logging passes FormattedLogValues carrying a {SessionId} field. Both surface + // their fields as KeyValuePair sequences (the nullable-annotated and + // unannotated forms are the same runtime type), so one branch reads both — no Akka internals. + private static void ScanState(TState state, out string? sessionId, out string? subSessionId, out string? logSource, out bool fromSessionLogActor) + { + sessionId = null; + subSessionId = null; + logSource = null; + fromSessionLogActor = false; + + if (state is IEnumerable> fields) + { + foreach (var field in fields) + Apply(field.Key, field.Value, ref sessionId, ref subSessionId, ref logSource, ref fromSessionLogActor); + } + } + + private static void Apply(string key, object? value, ref string? sessionId, ref string? subSessionId, ref string? logSource, ref bool fromSessionLogActor) + { + if (value is null) + return; + + if (key == NetclawLogProperties.SessionId) + { + if (value.ToString() is { Length: > 0 } id) + sessionId = id; + } + else if (key == NetclawLogProperties.SubSessionId) + { + if (value.ToString() is { Length: > 0 } subId) + subSessionId = subId; + } + else if (key == "LogSource") + { + logSource = value.ToString(); + if (IsSessionLogActorSource(logSource)) + fromSessionLogActor = true; + } + else if (key == "ActorPath") + { + if (IsSessionLogActorSource(value.ToString())) + fromSessionLogActor = true; + } + } + + // The session-log writer (and its dispatcher) must never have its own lines routed back to + // session.log. All actors share one MEL category, so identify it by Akka log source instead: + // the actor type (SessionLogActor) or the dispatcher path ("session-log-dispatcher"). + private static bool IsSessionLogActorSource(string? source) => + source is not null + && (source.Contains(nameof(SessionLogActor), StringComparison.Ordinal) + || source.Contains("session-log-dispatcher", StringComparison.Ordinal)); } diff --git a/src/Netclaw.Daemon/Configuration/RoutingChatClient.cs b/src/Netclaw.Daemon/Configuration/RoutingChatClient.cs index bffd4dc1c..337e9e71d 100644 --- a/src/Netclaw.Daemon/Configuration/RoutingChatClient.cs +++ b/src/Netclaw.Daemon/Configuration/RoutingChatClient.cs @@ -64,12 +64,12 @@ public async Task GetResponseAsync( catch (Exception ex) when (!cancellationToken.IsCancellationRequested && !isLast) { EmitFailover(ex); - LogWithSession(LogLevel.Warning, ex, "LLM provider failed, failing over to next candidate"); + LogWithSession(LogLevel.Warning, ex, "LLM provider failed, failing over to next candidate", options); } catch (Exception ex) when (!cancellationToken.IsCancellationRequested) { EmitUnreachable(ex, candidates.Count); - LogWithSession(LogLevel.Error, ex, UnreachableMessage(candidates.Count)); + LogWithSession(LogLevel.Error, ex, UnreachableMessage(candidates.Count), options); throw; } } @@ -137,12 +137,12 @@ public async IAsyncEnumerable GetStreamingResponseAsync( if (isLast) { EmitUnreachable(failure, candidates.Count); - LogWithSession(LogLevel.Error, failure, UnreachableMessage(candidates.Count)); + LogWithSession(LogLevel.Error, failure, UnreachableMessage(candidates.Count), options); ExceptionDispatchInfo.Capture(failure).Throw(); } EmitFailover(failure); - LogWithSession(LogLevel.Warning, failure, "LLM provider failed, failing over to next candidate"); + LogWithSession(LogLevel.Warning, failure, "LLM provider failed, failing over to next candidate", options); // outer loop advances to the next candidate } } @@ -180,11 +180,12 @@ private void EmitUnreachable(Exception ex, int candidateCount) => AlertSeverity.Critical, context: new Dictionary { ["error"] = ex.Message })); - // Failover/outage events are logged here, outside any per-pipeline LoggingChatClient - // scope, so attach the session id so they correlate by session in Seq. - private void LogWithSession(LogLevel level, Exception ex, string message) + // Failover/outage events are logged outside any per-pipeline LoggingChatClient + // scope, so attach the session id (carried on the call's options) here so they + // correlate by session in Seq. + private void LogWithSession(LogLevel level, Exception ex, string message, ChatOptions? options) { - using (SessionLoggingScope.Begin(_logger)) + using (ChatClientSessionScope.Begin(_logger, options)) _logger.Log(level, ex, message); } } diff --git a/src/Netclaw.Daemon/Configuration/SessionLoggingScope.cs b/src/Netclaw.Daemon/Configuration/SessionLoggingScope.cs deleted file mode 100644 index 14e1da57c..000000000 --- a/src/Netclaw.Daemon/Configuration/SessionLoggingScope.cs +++ /dev/null @@ -1,44 +0,0 @@ -// ----------------------------------------------------------------------- -// -// Copyright (C) 2026 - 2026 Petabridge, LLC -// -// ----------------------------------------------------------------------- -using Microsoft.Extensions.Logging; -using Netclaw.Configuration; - -namespace Netclaw.Daemon.Configuration; - -/// -/// Shared helper for tagging chat-client log lines with the ambient session id. -/// -/// The id lives in (an AsyncLocal pushed by -/// every session-owned path). The OTLP log exporter has IncludeScopes enabled -/// but cannot read that AsyncLocal directly, so without a logging scope LLM logs reach -/// Seq with no session correlation. Opening a scope keyed SessionId surfaces it -/// as a filterable attribute — matching the WithContext("SessionId", …) key the -/// session/channel actors already use, so a single attribute correlates actor and -/// chat-client logs. -/// -/// MEL scopes are per-, so each chat-client decorator that emits -/// its own log lines (e.g. and the routing/failover -/// client) opens its own scope from this helper. -/// -internal static class SessionLoggingScope -{ - private const string SessionIdKey = "SessionId"; - - /// - /// Opens a scope tagging subsequent log lines on with the - /// ambient session id, or returns null when no session is in scope (a no-op - /// using). A single-entry array avoids a per-call dictionary allocation while - /// still presenting as IEnumerable<KeyValuePair>, which is what the OTLP - /// exporter projects into log attributes. - /// - public static IDisposable? Begin(ILogger logger) - { - var sessionId = SessionDiagnosticsContext.SessionId; - return sessionId is null - ? null - : logger.BeginScope(new[] { new KeyValuePair(SessionIdKey, sessionId) }); - } -} diff --git a/src/Netclaw.Daemon/Gateway/SignalRSessionActor.cs b/src/Netclaw.Daemon/Gateway/SignalRSessionActor.cs index 2c9a635cc..20a206d20 100644 --- a/src/Netclaw.Daemon/Gateway/SignalRSessionActor.cs +++ b/src/Netclaw.Daemon/Gateway/SignalRSessionActor.cs @@ -55,7 +55,7 @@ public SignalRSessionActor( _hubContext = hubContext; _log = Context.GetLogger() .WithContext("Adapter", "signalr") - .WithContext("SessionId", _sessionId.Value); + .WithContext(NetclawLogProperties.SessionId, _sessionId.Value); _handle = new SessionPipelineHandle(pipeline, _log, "signalr"); Initializing();