From 684167292fd746aed3c47acd8d911bb6b65db3c3 Mon Sep 17 00:00:00 2001 From: Oliver Kieselbach Date: Thu, 1 Oct 2026 21:45:10 +0200 Subject: [PATCH] Gather rules keep their place across agent restarts; byte-exact logparser reader; MCP filters by gather event types - Logparser positions per (rule, file) and the on_change values persist in State\gather-rule-state.json, bound to the session id. A reboot, hard kill or WhiteGlove Part 2 no longer replays a log from byte 0 or re-emits an unchanged on_change result. A replaced log (fingerprint of its first 4 KB) is read from the beginning. Positions are saved after the events went out; restoring the file is fail-soft. --run-gather-rules stays in memory. - LogLineReader: a maxLines stop loses no line, an unfinished last line is never split (interval rules wait one run, one-shot triggers read it again next run), UTF-16/UTF-32 logs are read by their byte order mark from any position, line ends as StreamReader.ReadLine, logLineNumber counts from the start of the file, lines over 32 MB are skipped. A maxLines value that is no positive number falls back to 1000. - UserProfileResolver no longer caches "no user signed in yet" (detection retried at most every 30 s); guarded callers expand the token with the profile they checked. - MCP get_session_events, search_sessions_by_event and query_raw_events accept a gather rule's own outputEventType; an empty first-page result for an uncatalogued type carries eventTypeNote with the closest built-in types instead of being rejected up front. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../Gather/GatherRuleCollectorGuardTests.cs | 9 +- .../Gather/GatherRuleDebugLogTests.cs | 2 +- .../Gather/GatherRuleExecutorRestartTests.cs | 220 ++++++++++ .../Gather/GatherRuleSessionStateTests.cs | 278 ++++++++++++ .../Monitoring/Gather/LogLineReaderTests.cs | 188 ++++++++ .../Gather/LogParserCollectorDebugLogTests.cs | 2 +- .../Gather/LogParserCollectorReadingTests.cs | 225 ++++++++++ .../RegistryCollectorEmitOnlyIfExistsTests.cs | 2 +- .../Runtime/DiagnosticsPackageServiceTests.cs | 1 + .../Runtime/UserProfileResolverTests.cs | 96 +++++ ...DefaultComponentFactoryGatherStateTests.cs | 103 +++++ .../Enrollment/Ime/LogFilePositionTracker.cs | 6 +- .../Runtime/DiagnosticsPackageService.cs | 2 +- .../Monitoring/Runtime/UserProfileResolver.cs | 76 +++- .../Gather/Collectors/FileCollector.cs | 2 +- .../Gather/Collectors/JsonCollector.cs | 2 +- .../Gather/Collectors/LogLineReader.cs | 313 ++++++++++++++ .../Gather/Collectors/LogParserCollector.cs | 404 +++++++++++------- .../Gather/Collectors/XmlCollector.cs | 2 +- .../Telemetry/Gather/GatherRuleContext.cs | 10 +- .../Telemetry/Gather/GatherRuleExecutor.cs | 65 ++- .../Gather/GatherRuleSessionState.cs | 399 +++++++++++++++++ .../Gather/GatherRuleStatePersistence.cs | 131 ++++++ .../Orchestration/DefaultComponentFactory.cs | 3 +- .../Hosts/GatherRuleExecutorHost.cs | 5 +- .../Program.GatherRulesMode.cs | 3 + .../src/__tests__/auto-exhaust.test.ts | 81 ++-- .../src/__tests__/event-type-filter.test.ts | 101 +++++ .../autopilot-monitor-mcp/src/client.ts | 2 +- .../src/resource-catalog.ts | 71 ++- .../autopilot-monitor-mcp/src/tools/admin.ts | 14 +- .../src/tools/sessions.ts | 28 +- 32 files changed, 2557 insertions(+), 289 deletions(-) create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/GatherRuleExecutorRestartTests.cs create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/GatherRuleSessionStateTests.cs create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/LogLineReaderTests.cs create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/LogParserCollectorReadingTests.cs create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Runtime/UserProfileResolverTests.cs create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/DefaultComponentFactoryGatherStateTests.cs create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/LogLineReader.cs create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/GatherRuleSessionState.cs create mode 100644 src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/GatherRuleStatePersistence.cs create mode 100644 src/McpServer/autopilot-monitor-mcp/src/__tests__/event-type-filter.test.ts diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/GatherRuleCollectorGuardTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/GatherRuleCollectorGuardTests.cs index c38d9f0b..79f1e4e4 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/GatherRuleCollectorGuardTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/GatherRuleCollectorGuardTests.cs @@ -41,15 +41,18 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring.Gather } private GatherRuleContext Context(bool unrestricted = false, string? imeLogPathOverride = null) - => new GatherRuleContext( - new AgentLogger(_tmp.Path, AgentLogLevel.Info), + { + var logger = new AgentLogger(_tmp.Path, AgentLogLevel.Info); + return new GatherRuleContext( + logger, "sess", "tenant", evt => _events.Add(evt), imeLogPathOverride, - new LogFilePositionTracker()) + new GatherRuleSessionState("sess", null, logger)) { UnrestrictedMode = unrestricted }; + } private static GatherRule Rule(string collectorType, string target, Dictionary? parameters = null) => new GatherRule diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/GatherRuleDebugLogTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/GatherRuleDebugLogTests.cs index 271e1575..3149511d 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/GatherRuleDebugLogTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/GatherRuleDebugLogTests.cs @@ -94,7 +94,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring.Gather public void Context_DebugLog_is_noop_without_writer() { var context = new GatherRuleContext( - _logger, "sess", "tenant", _ => { }, null, new LogFilePositionTracker()); + _logger, "sess", "tenant", _ => { }, null, new GatherRuleSessionState("sess", null, _logger)); context.DebugLog("RULE-1", GatherRuleDebugLog.StageScope, "should not throw, should not write"); diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/GatherRuleExecutorRestartTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/GatherRuleExecutorRestartTests.cs new file mode 100644 index 00000000..5dba3ce9 --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/GatherRuleExecutorRestartTests.cs @@ -0,0 +1,220 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using AutopilotMonitor.Agent.V2.Core.Logging; +using AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather; +using AutopilotMonitor.Agent.V2.Core.Tests.Harness; +using AutopilotMonitor.Shared.Models; +using Newtonsoft.Json; +using Xunit; + +namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring.Gather +{ + /// + /// An agent restart inside one enrollment session — a new on the + /// same state directory — continues where the previous run stopped: logparser rules do not + /// replay their log, on_change rules do not re-emit an unchanged result. Regression anchor: + /// session c8106021, where every restart replayed the whole HP Image Assistant log (70 events). + /// + [Collection("SerialThreading")] // startup rules execute on the shared ThreadPool + public sealed class GatherRuleExecutorRestartTests : IDisposable + { + private readonly TempDirectory _tmp = new TempDirectory(); + private readonly TempDirectory _stateDir = new TempDirectory(); + private readonly List _events = new List(); + private readonly object _eventsGate = new object(); + private readonly List _executors = new List(); + + // %TEMP% lives under C:\Users (hard-blocked) — log targets sit beside the test assembly. + private readonly string _logDir = Path.Combine( + AppContext.BaseDirectory, "gather-restart-tests-" + Guid.NewGuid().ToString("N")); + + public GatherRuleExecutorRestartTests() + { + Directory.CreateDirectory(_logDir); + } + + public void Dispose() + { + foreach (var executor in _executors) + executor.Dispose(); + _tmp.Dispose(); + _stateDir.Dispose(); + try { Directory.Delete(_logDir, recursive: true); } catch { /* best effort */ } + } + + private string StateFile => Path.Combine(_stateDir.Path, GatherRuleStatePersistence.FileName); + + private List Events + { + get { lock (_eventsGate) return _events.ToList(); } + } + + /// One agent run: a fresh executor whose startup rules run once. Not disposed — a hard kill. + private GatherRuleExecutor AgentRun(GatherRule rule, string sessionId = "session-a", string? stateDirectory = "default", + Action? onEvent = null, string? debugLogPath = null) + { + var executor = new GatherRuleExecutor( + sessionId, "tenant", + evt => + { + onEvent?.Invoke(evt); + lock (_eventsGate) _events.Add(evt); + }, + new AgentLogger(_tmp.Path, AgentLogLevel.Info), + debugLogPath: debugLogPath, + stateDirectory: stateDirectory == "default" ? _stateDir.Path : stateDirectory) + { + UnrestrictedMode = true + }; + _executors.Add(executor); + executor.UpdateRules(new List { rule }); + Assert.True(executor.WaitForStartupRules(30)); + return executor; + } + + private static GatherRule LogRule(string path, string id = "hpia-log-collect") => new GatherRule + { + RuleId = id, + Title = id, + CollectorType = "logparser", + Target = path, + Parameters = new Dictionary { ["pattern"] = "^(?Update .+)$", ["format"] = "text" }, + Trigger = "startup", + OutputEventType = "HPiA-UpdateStatus", + Enabled = true, + }; + + private static GatherRule OnChangeFileRule(string path) => new GatherRule + { + RuleId = "GATHER-OC-FILE", + Title = "file presence", + CollectorType = "file", + Target = path, + Trigger = "startup", + OutputEventType = "gather_file_presence", + Enabled = true, + EmitMode = "on_change", + }; + + private string WriteLog(string name, string content) + { + var path = Path.Combine(_logDir, name); + File.WriteAllText(path, content); + return path; + } + + private int StatusEvents => Events.Count(e => e.EventType == "HPiA-UpdateStatus"); + + [Fact] + public void Restart_does_not_replay_the_log_and_new_lines_arrive_once() + { + var path = WriteLog("HP Image Assistant.log", "Update one\nUpdate two\nnoise\n"); + + AgentRun(LogRule(path)); + Assert.Equal(2, StatusEvents); + + AgentRun(LogRule(path)); // reboot, nothing new + Assert.Equal(2, StatusEvents); + + File.AppendAllText(path, "Update three\n"); + AgentRun(LogRule(path)); // next restart picks up only the new line + Assert.Equal(3, StatusEvents); + Assert.Equal("Update three", Events.Last(e => e.EventType == "HPiA-UpdateStatus").Data["status"]); + } + + [Fact] + public void Position_is_on_disk_right_after_the_run_without_a_clean_shutdown() + { + var path = WriteLog("hardkill.log", "Update one\n"); + + AgentRun(LogRule(path)); // never disposed + + var saved = JsonConvert.DeserializeObject(File.ReadAllText(StateFile))!; + var position = Assert.Single(saved.LogPositions!); + Assert.Equal(new FileInfo(path).Length, position.Position); + Assert.Equal("session-a", saved.SessionId); + } + + [Fact] + public void Position_is_saved_only_after_the_events_went_out() + { + var path = WriteLog("ordering.log", "Update one\n"); + var savedDuringEmit = new List(); + + AgentRun(LogRule(path), onEvent: _ => savedDuringEmit.Add(File.Exists(StateFile))); + + Assert.Equal(new[] { false }, savedDuringEmit); + Assert.True(File.Exists(StateFile)); + } + + [Fact] + public void A_new_session_reads_the_log_from_the_beginning() + { + var path = WriteLog("rotated.log", "Update one\nUpdate two\n"); + AgentRun(LogRule(path), sessionId: "session-a"); + + AgentRun(LogRule(path), sessionId: "session-b"); // session rotated on the same device + + Assert.Equal(4, StatusEvents); + } + + [Fact] + public void A_log_replaced_during_the_reboot_is_read_from_the_beginning() + { + var path = WriteLog("replaced.log", "Update one\n"); + AgentRun(LogRule(path, "replaced")); + + // The tool writes a new log after the reboot; it is already longer than the old position. + File.WriteAllText(path, "Update fresh A\nUpdate fresh B\nUpdate fresh C\n"); + AgentRun(LogRule(path, "replaced")); + + Assert.Equal(new[] { "Update one", "Update fresh A", "Update fresh B", "Update fresh C" }, + Events.Where(e => e.EventType == "HPiA-UpdateStatus").Select(e => (string)e.Data["status"]).ToArray()); + } + + [Fact] + public void An_unchanged_on_change_result_stays_silent_after_a_restart() + { + var watched = Path.Combine(_logDir, "marker.txt"); + + AgentRun(OnChangeFileRule(watched)); // absent → first result emits + AgentRun(OnChangeFileRule(watched)); // reboot, still absent → silent + Assert.Single(Events, e => e.EventType == "gather_file_presence"); + + File.WriteAllText(watched, "present"); + AgentRun(OnChangeFileRule(watched)); // changed → emits + + Assert.Equal(2, Events.Count(e => e.EventType == "gather_file_presence")); + } + + [Fact] + public void Without_a_state_directory_every_run_starts_over_and_writes_nothing() + { + var path = WriteLog("diagnostic.log", "Update one\n"); + + AgentRun(LogRule(path), stateDirectory: null); + AgentRun(LogRule(path), stateDirectory: null); + + Assert.Equal(2, StatusEvents); + Assert.False(File.Exists(StateFile)); + } + + [Fact] + public void Debug_trace_says_what_was_restored_and_where_reading_continues() + { + var path = WriteLog("traced.log", "Update one\n"); + var trace = Path.Combine(_tmp.Path, "gather-rules-debug.log"); + AgentRun(LogRule(path)); + File.AppendAllText(path, "Update two\n"); + + AgentRun(LogRule(path), debugLogPath: trace); + + var text = File.ReadAllText(trace); + Assert.Contains("gather state: restored 1 logparser position(s) for 1 rule(s)", text); + Assert.Contains("hpia-log-collect | config | logparser position restored: traced.log", text); + Assert.Contains("traced.log: continuing at position", text); + } + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/GatherRuleSessionStateTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/GatherRuleSessionStateTests.cs new file mode 100644 index 00000000..61c8ce12 --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/GatherRuleSessionStateTests.cs @@ -0,0 +1,278 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Text; +using System.Threading.Tasks; +using AutopilotMonitor.Agent.V2.Core.Logging; +using AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather; +using AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather.Collectors; +using AutopilotMonitor.Agent.V2.Core.Tests.Harness; +using Newtonsoft.Json; +using Xunit; + +namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring.Gather +{ + /// + /// and : what is + /// written, what a restart restores, what it refuses, and when the file is (not) written. + /// + public sealed class GatherRuleSessionStateTests : IDisposable + { + private readonly TempDirectory _tmp = new TempDirectory(); + private readonly TempDirectory _stateDir = new TempDirectory(); + private readonly AgentLogger _logger; + + public GatherRuleSessionStateTests() + { + _logger = new AgentLogger(_tmp.Path, AgentLogLevel.Info); + } + + public void Dispose() + { + _tmp.Dispose(); + _stateDir.Dispose(); + } + + private string StateFile => Path.Combine(_stateDir.Path, GatherRuleStatePersistence.FileName); + + private GatherRuleSessionState NewState(string sessionId = "session-a") => + new GatherRuleSessionState(sessionId, new GatherRuleStatePersistence(_stateDir.Path, _logger), _logger); + + private static LogFileHead Head(string text) => LogFileHead.FromBytes(Encoding.UTF8.GetBytes(text)); + + // ── restore ───────────────────────────────────────────────────────── + + [Fact] + public void Positions_and_emitted_hashes_survive_a_restart_of_the_same_session() + { + var first = NewState(); + first.CommitLogPosition("RULE-1", @"C:\Logs\a.log", 120, 7, Head("2026-09-28 a.log start"), heldTailStart: -1); + first.CommitEmittedHash("RULE-2", "hash-value"); + + var second = NewState(); + + var position = second.PeekLogPosition("RULE-1", @"C:\Logs\a.log"); + Assert.NotNull(position); + Assert.Equal(120, position!.Position); + Assert.Equal(7, position.LineNumber); + Assert.True(position.FromPreviousRun); + Assert.Equal("hash-value", second.GetEmittedHashes()["RULE-2"]); + Assert.StartsWith("restored 1 logparser position(s) for 1 rule(s) and 1 on_change value(s)", second.RestoreSummary); + } + + [Fact] + public void State_of_another_session_is_ignored() + { + NewState("session-a").CommitLogPosition("RULE-1", @"C:\Logs\a.log", 120, 7, Head("start"), -1); + + var rotated = NewState("session-b"); + + Assert.Null(rotated.PeekLogPosition("RULE-1", @"C:\Logs\a.log")); + Assert.Contains("saved for session session-a, current session is session-b", rotated.RestoreSummary); + } + + [Fact] + public void Missing_file_is_a_first_run_and_corrupt_file_a_fresh_start() + { + Assert.StartsWith("no saved state", NewState().RestoreSummary); + + File.WriteAllText(StateFile, "not json {{{"); + var afterCorruption = NewState(); + + Assert.StartsWith("saved state discarded", afterCorruption.RestoreSummary); + Assert.Empty(afterCorruption.GetEmittedHashes()); + } + + [Fact] + public void Tampered_entries_are_dropped_and_the_rest_restored() + { + var data = new GatherRuleStateData + { + SessionId = "session-a", + SavedAtUtc = DateTime.UtcNow, + LogPositions = new List + { + new GatherLogPositionData { RuleId = "OK", Path = @"C:\a.log", Position = 10, LineNumber = 1, HeadLength = 10, HeadHash = "h" }, + new GatherLogPositionData { RuleId = "NEG", Path = @"C:\b.log", Position = -5, HeadLength = 10, HeadHash = "h" }, + new GatherLogPositionData { RuleId = "", Path = @"C:\c.log", Position = 1, HeadLength = 1, HeadHash = "h" }, + new GatherLogPositionData { RuleId = "BIGHEAD", Path = @"C:\d.log", Position = 1, HeadLength = 5000, HeadHash = "h" }, + new GatherLogPositionData { RuleId = "NOHEAD", Path = @"C:\e.log", Position = 50, HeadLength = 0 }, + }, + OnChangeHashes = new Dictionary { ["GOOD"] = "x", ["EMPTY"] = "" }, + }; + File.WriteAllText(StateFile, JsonConvert.SerializeObject(data)); + + var state = NewState(); + + Assert.NotNull(state.PeekLogPosition("OK", @"C:\a.log")); + Assert.Null(state.PeekLogPosition("NEG", @"C:\b.log")); + Assert.Null(state.PeekLogPosition("BIGHEAD", @"C:\d.log")); + Assert.Null(state.PeekLogPosition("NOHEAD", @"C:\e.log")); + Assert.Equal(new[] { "GOOD" }, state.GetEmittedHashes().Keys.ToArray()); + Assert.Contains("5 invalid entries dropped", state.RestoreSummary); + } + + [Fact] + public void A_path_no_real_file_can_have_is_dropped_and_never_stops_the_executor() + { + // Path APIs throw on these characters under .NET Framework; a throw here would keep + // every collector host down at every restart of the session. + var data = new GatherRuleStateData + { + SessionId = "session-a", + LogPositions = new List + { + new GatherLogPositionData { RuleId = "BAD", Path = "C:\\Logs\\x.|og", Position = 1, HeadLength = 1, HeadHash = "h" }, + new GatherLogPositionData { RuleId = "WILD", Path = "C:\\Logs\\*.log", Position = 1, HeadLength = 1, HeadHash = "h" }, + new GatherLogPositionData { RuleId = "OK", Path = "C:\\Logs\\ok.log", Position = 1, HeadLength = 1, HeadHash = "h" }, + }, + }; + File.WriteAllText(StateFile, JsonConvert.SerializeObject(data)); + + var state = NewState(); + + Assert.NotNull(state.PeekLogPosition("OK", "C:\\Logs\\ok.log")); + Assert.Null(state.PeekLogPosition("BAD", "C:\\Logs\\x.|og")); + Assert.Contains("2 invalid entries dropped", state.RestoreSummary); + } + + // ── where reading starts ──────────────────────────────────────────── + + [Fact] + public void Reading_continues_while_the_file_head_still_matches() + { + var state = NewState(); + state.CommitLogPosition("R", @"C:\a.log", 30, 3, Head("2026-09-28 first line\n"), -1); + + var start = GatherRuleSessionState.ResolveStart(state.PeekLogPosition("R", @"C:\a.log"), 200, + Head("2026-09-28 first line\nsecond line\nthird"), out var how); + + Assert.Equal(30, start); + Assert.Equal(LogReadStart.Continue, how); + } + + [Fact] + public void A_shrunk_file_and_a_replaced_file_are_read_from_the_beginning() + { + var state = NewState(); + state.CommitLogPosition("R", @"C:\a.log", 30, 3, Head("2026-09-28 first line\n"), -1); + var known = state.PeekLogPosition("R", @"C:\a.log"); + + Assert.Equal(0, GatherRuleSessionState.ResolveStart(known, 20, Head("2026-09-28 fir"), out var shrunk)); + Assert.Equal(LogReadStart.Shrunk, shrunk); + + // Replaced and already longer than the old position: only the head tells. + Assert.Equal(0, GatherRuleSessionState.ResolveStart(known, 500, Head("2026-09-30 a new log entirely, much longer"), out var replaced)); + Assert.Equal(LogReadStart.Replaced, replaced); + + // Shorter than the recorded head although not shorter than the position. + state.CommitLogPosition("S", @"C:\b.log", 5, 1, Head("0123456789abcdef"), -1); + Assert.Equal(0, GatherRuleSessionState.ResolveStart(state.PeekLogPosition("S", @"C:\b.log"), 8, Head("01234567"), out var cut)); + Assert.Equal(LogReadStart.Replaced, cut); + } + + [Fact] + public void Positions_are_kept_per_rule_for_the_same_file() + { + var state = NewState(); + state.CommitLogPosition("RULE-A", @"C:\shared.log", 100, 10, Head("x"), -1); + + Assert.Null(state.PeekLogPosition("RULE-B", @"C:\shared.log")); + Assert.NotNull(state.PeekLogPosition("RULE-A", @"C:\SHARED.LOG")); // paths compare case-insensitively + } + + // ── when the file is written ──────────────────────────────────────── + + [Fact] + public void An_unchanged_commit_does_not_write_the_file() + { + var state = NewState(); + state.CommitLogPosition("R", @"C:\a.log", 10, 1, Head("head"), -1); + File.Delete(StateFile); + + state.CommitLogPosition("R", @"C:\a.log", 10, 1, Head("head"), -1); + state.CommitEmittedHash("H", "v"); + File.Delete(StateFile); + state.CommitEmittedHash("H", "v"); + + Assert.False(File.Exists(StateFile)); + } + + [Fact] + public void A_failed_save_is_retried_at_the_next_commit() + { + var blocked = Path.Combine(_stateDir.Path, "blocked"); + File.WriteAllText(blocked, "a file where the state directory should be"); + var state = new GatherRuleSessionState("session-a", new GatherRuleStatePersistence(blocked, _logger), _logger); + + state.CommitEmittedHash("H", "v"); // Directory.CreateDirectory fails + File.Delete(blocked); + state.CommitEmittedHash("H", "v"); // unchanged, but the earlier save is still owed + + Assert.True(File.Exists(Path.Combine(blocked, GatherRuleStatePersistence.FileName))); + } + + [Fact] + public void A_closed_state_no_longer_writes() + { + var state = NewState(); + state.Close(); + + state.CommitEmittedHash("H", "v"); + + Assert.False(File.Exists(StateFile)); + } + + [Fact] + public void Without_persistence_nothing_is_written() + { + var state = new GatherRuleSessionState("session-a", null, _logger); + + state.CommitLogPosition("R", @"C:\a.log", 10, 1, Head("head"), -1); + + Assert.StartsWith("persistence off", state.RestoreSummary); + Assert.Empty(Directory.GetFiles(_stateDir.Path)); + } + + // ── bounds and concurrency ────────────────────────────────────────── + + [Fact] + public void A_rule_forgets_its_least_recently_touched_file_beyond_the_cap() + { + var state = NewState(); + for (var i = 0; i < GatherRuleSessionState.MaxLogFilesPerRule; i++) + state.CommitLogPosition("R", $@"C:\logs\{i}.log", 1, 1, Head("h"), -1); + state.PeekLogPosition("R", @"C:\logs\0.log"); // touched again — survives + + state.CommitLogPosition("R", @"C:\logs\new.log", 1, 1, Head("h"), -1); + + Assert.NotNull(state.PeekLogPosition("R", @"C:\logs\0.log")); + Assert.Null(state.PeekLogPosition("R", @"C:\logs\1.log")); + Assert.NotNull(state.PeekLogPosition("R", @"C:\logs\new.log")); + } + + [Fact] + public void Parallel_commits_leave_a_consistent_file() + { + var state = NewState(); + + Parallel.For(0, 8, worker => + { + for (var i = 0; i < 50; i++) + { + state.CommitLogPosition($"RULE-{worker}", $@"C:\logs\{worker}.log", i + 1, i, Head("h"), -1); + state.PeekLogPosition($"RULE-{worker}", $@"C:\logs\{worker}.log"); + state.CommitEmittedHash($"RULE-{worker}", "hash-" + i); + } + }); + + var restored = NewState(); + for (var worker = 0; worker < 8; worker++) + { + Assert.Equal(50, restored.PeekLogPosition($"RULE-{worker}", $@"C:\logs\{worker}.log")!.Position); + Assert.Equal("hash-49", restored.GetEmittedHashes()[$"RULE-{worker}"]); + } + } + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/LogLineReaderTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/LogLineReaderTests.cs new file mode 100644 index 00000000..c5b620ed --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/LogLineReaderTests.cs @@ -0,0 +1,188 @@ +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Text; +using AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather.Collectors; +using Xunit; + +namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring.Gather +{ + /// + /// , and : + /// exact byte positions, StreamReader-compatible line ends, byte order marks only at offset 0, + /// and the code-unit view of UTF-16/UTF-32 files. + /// + public sealed class LogLineReaderTests + { + private static byte[] Bytes(Encoding encoding, string text, bool withBom = true) + { + var body = encoding.GetBytes(text); + return withBom ? encoding.GetPreamble().Concat(body).ToArray() : body; + } + + private static (List Lines, List EndPositions, List Terminated) ReadAll( + byte[] file, long start = 0, int maxLineBytes = 1024) + { + var encoding = LogFileHead.FromBytes(file.Take(LogFileHead.MaxBytes).ToArray()).Encoding; + using (var stream = new MemoryStream(file)) + { + stream.Seek(start, SeekOrigin.Begin); + var reader = new LogLineReader(stream, encoding, maxLineBytes); + var lines = new List(); + var ends = new List(); + var terminated = new List(); + string line; + while ((line = reader.ReadLine()) != null) + { + lines.Add(line); + ends.Add(reader.Position); + terminated.Add(reader.LastLineTerminated); + } + return (lines, ends, terminated); + } + } + + [Fact] + public void Line_ends_match_StreamReader_including_a_lone_carriage_return() + { + var file = Encoding.UTF8.GetBytes("one\ntwo\r\nthree\rfour"); + + var result = ReadAll(file); + + Assert.Equal(new[] { "one", "two", "three", "four" }, result.Lines); + Assert.Equal(new long[] { 4, 9, 15, 19 }, result.EndPositions); + Assert.Equal(new[] { true, true, true, false }, result.Terminated); + } + + [Fact] + public void Position_after_each_line_is_the_offset_to_resume_from() + { + var file = Encoding.UTF8.GetBytes("alpha\nbeta\ngamma\n"); + + var first = ReadAll(file); + var resumed = ReadAll(file, start: first.EndPositions[0]); + + Assert.Equal(new[] { "beta", "gamma" }, resumed.Lines); + Assert.Equal(file.Length, resumed.EndPositions.Last()); + } + + [Fact] + public void Utf8_byte_order_mark_is_skipped_only_at_offset_0() + { + var file = Bytes(Encoding.UTF8, "first\nsecond\n"); + + var result = ReadAll(file); + + Assert.Equal(new[] { "first", "second" }, result.Lines); + Assert.Equal(LogTextEncoding.Utf8WithBom, LogFileHead.FromBytes(file).Encoding); + } + + [Theory] + [InlineData("utf-16le")] + [InlineData("utf-16be")] + [InlineData("utf-32le")] + [InlineData("utf-32be")] + public void Unicode_files_are_read_in_their_code_units_from_any_resume_point(string name) + { + var encoding = name == "utf-16le" ? Encoding.Unicode + : name == "utf-16be" ? Encoding.BigEndianUnicode + : name == "utf-32le" ? Encoding.UTF32 + : new UTF32Encoding(bigEndian: true, byteOrderMark: true); + var file = Bytes(encoding, "ERROR first\r\nERROR zweite Zeile mit Umlaut äöü\r\n"); + + var whole = ReadAll(file); + var resumed = ReadAll(file, start: whole.EndPositions[0]); + + Assert.Equal(name, LogFileHead.FromBytes(file).Encoding.Name); + Assert.Equal(new[] { "ERROR first", "ERROR zweite Zeile mit Umlaut äöü" }, whole.Lines); + Assert.Equal(new[] { "ERROR zweite Zeile mit Umlaut äöü" }, resumed.Lines); + Assert.Equal(file.Length, whole.EndPositions.Last()); + } + + [Fact] + public void File_without_byte_order_mark_is_utf8_and_ansi_bytes_decode_as_before() + { + var file = new byte[] { (byte)'M', (byte)'a', (byte)'n', 0xFC, (byte)'\n' }; // "Man" + Windows-1252 'ü' + + var result = ReadAll(file); + + Assert.Equal(LogTextEncoding.Utf8, LogFileHead.FromBytes(file).Encoding); + Assert.Equal(new StreamReader(new MemoryStream(file), Encoding.UTF8).ReadLine(), result.Lines.Single()); + } + + [Fact] + public void Carriage_return_at_end_of_file_ends_the_line_and_its_late_line_feed_is_skipped() + { + var firstWrite = Encoding.UTF8.GetBytes("done\r"); + var later = Encoding.UTF8.GetBytes("done\r\nnext\n"); + + var first = ReadAll(firstWrite); + var resumed = ReadAll(later, start: first.EndPositions.Last()); + + Assert.Equal(new[] { "done" }, first.Lines); + Assert.True(first.Terminated.Single()); + Assert.Equal(new[] { "next" }, resumed.Lines); + } + + [Fact] + public void Overlong_line_is_consumed_without_being_kept_and_reading_continues_behind_it() + { + var file = Encoding.UTF8.GetBytes(new string('x', 50) + "\nshort\n"); + using (var stream = new MemoryStream(file)) + { + var reader = new LogLineReader(stream, LogTextEncoding.Utf8, maxLineBytes: 10); + + Assert.Equal(string.Empty, reader.ReadLine()); + Assert.True(reader.LastLineTruncated); + Assert.Equal(51, reader.Position); + Assert.Equal("short", reader.ReadLine()); + Assert.False(reader.LastLineTruncated); + Assert.Null(reader.ReadLine()); + } + } + + [Fact] + public void Line_spanning_many_read_buffers_is_returned_whole() + { + var longLine = new string('y', 200_000); + var file = Encoding.UTF8.GetBytes(longLine + "\nend\n"); + + var result = ReadAll(file, maxLineBytes: 1_000_000); + + Assert.Equal(new[] { longLine, "end" }, result.Lines); + } + + [Fact] + public void Half_written_utf16_character_at_end_of_file_stays_in_the_unterminated_line() + { + var full = Bytes(Encoding.Unicode, "ok\r\nhalf"); + var file = full.Concat(new byte[] { 0x41 }).ToArray(); // first byte of the next UTF-16 unit + + var result = ReadAll(file); + + Assert.Equal("ok", result.Lines[0]); + Assert.False(result.Terminated.Last()); + Assert.Equal(file.Length, result.EndPositions.Last()); + } + + [Fact] + public void AlignDown_snaps_to_code_units_behind_the_byte_order_mark() + { + Assert.Equal(0, LogTextEncoding.Utf16LittleEndian.AlignDown(1)); + Assert.Equal(2, LogTextEncoding.Utf16LittleEndian.AlignDown(3)); + Assert.Equal(6, LogTextEncoding.Utf16LittleEndian.AlignDown(7)); + Assert.Equal(8, LogTextEncoding.Utf32LittleEndian.AlignDown(11)); + Assert.Equal(7, LogTextEncoding.Utf8.AlignDown(7)); + } + + [Fact] + public void Head_hash_of_a_prefix_matches_the_hash_of_the_shorter_file() + { + var shortFile = LogFileHead.FromBytes(Encoding.UTF8.GetBytes("2026-09-28 start")); + var grown = LogFileHead.FromBytes(Encoding.UTF8.GetBytes("2026-09-28 start\nmore lines")); + + Assert.Equal(shortFile.Hash(shortFile.Length), grown.Hash(shortFile.Length)); + Assert.NotEqual(shortFile.Hash(shortFile.Length), grown.Hash(grown.Length)); + } + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/LogParserCollectorDebugLogTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/LogParserCollectorDebugLogTests.cs index 14b43c2e..d264f05b 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/LogParserCollectorDebugLogTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/LogParserCollectorDebugLogTests.cs @@ -37,7 +37,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring.Gather logger, "sess", "tenant", evt => _events.Add(evt), null, - new LogFilePositionTracker(), + new GatherRuleSessionState("sess", null, logger), new GatherRuleDebugLog(_debugLogPath, logger)) { UnrestrictedMode = true diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/LogParserCollectorReadingTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/LogParserCollectorReadingTests.cs new file mode 100644 index 00000000..f63d887a --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/LogParserCollectorReadingTests.cs @@ -0,0 +1,225 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Text; +using AutopilotMonitor.Agent.V2.Core.Logging; +using AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime; +using AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather; +using AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather.Collectors; +using AutopilotMonitor.Agent.V2.Core.Tests.Harness; +using AutopilotMonitor.Shared.Models; +using Xunit; + +namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring.Gather +{ + /// + /// Reading semantics of across runs on one context: every + /// line of a growing log is matched exactly once — no line lost at a maxLines cut, no line + /// split by a writer that is mid-line, UTF-16 logs readable after the first run, positions + /// kept per rule, line numbers counted from the start of the file. + /// + public sealed class LogParserCollectorReadingTests : IDisposable + { + private readonly TempDirectory _tmp = new TempDirectory(); + private readonly List _events = new List(); + private readonly GatherRuleContext _context; + + // %TEMP% lives under C:\Users (hard-blocked) — parse targets sit beside the + // test assembly instead, admitted via UnrestrictedMode. + private readonly string _logDir = Path.Combine( + AppContext.BaseDirectory, "gather-reading-tests-" + Guid.NewGuid().ToString("N")); + + public LogParserCollectorReadingTests() + { + Directory.CreateDirectory(_logDir); + _context = NewContext(); + } + + public void Dispose() + { + _tmp.Dispose(); + try { Directory.Delete(_logDir, recursive: true); } catch { /* best effort */ } + } + + private GatherRuleContext NewContext() + { + var logger = new AgentLogger(_tmp.Path, AgentLogLevel.Info); + return new GatherRuleContext(logger, "sess", "tenant", + evt => _events.Add(evt), null, new GatherRuleSessionState("sess", null, logger)) + { + UnrestrictedMode = true + }; + } + + private string LogPath(string name) => Path.Combine(_logDir, name); + + private static GatherRule TextRule(string ruleId, string target, string pattern, + string trigger = "interval", int? maxLines = null) + { + var parameters = new Dictionary { ["pattern"] = pattern, ["format"] = "text" }; + if (maxLines.HasValue) parameters["maxLines"] = maxLines.Value.ToString(); + return new GatherRule + { + RuleId = ruleId, + Title = ruleId, + CollectorType = "logparser", + Target = target, + Parameters = parameters, + Trigger = trigger, + IntervalSeconds = trigger == "interval" ? 30 : (int?)null, + OutputEventType = "logparser_reading_test", + }; + } + + private List Captured(string group) => + _events.Select(e => e.Data.TryGetValue(group, out var v) ? (string)v : "").ToList(); + + [Fact] + public void Two_rules_on_one_file_each_read_the_whole_file() + { + var path = LogPath("shared.log"); + File.WriteAllText(path, "ERROR one\nWARN two\nERROR three\n"); + var collector = new LogParserCollector(); + + collector.Execute(TextRule("RULE-ERRORS", path, "^ERROR (?\\w+)"), _context); + collector.Execute(TextRule("RULE-WARNINGS", path, "^WARN (?\\w+)"), _context); + + Assert.Equal(2, _events.Count(e => (string)e.Data["ruleId"] == "RULE-ERRORS")); + Assert.Single(_events, e => (string)e.Data["ruleId"] == "RULE-WARNINGS"); + } + + [Fact] + public void MaxLines_cut_loses_no_line_across_runs() + { + var path = LogPath("long.log"); + var sb = new StringBuilder(); + for (var i = 1; i <= 30; i++) + sb.Append("hit ").Append(i.ToString("D3")).Append(' ').Append('x', 90).Append('\n'); + File.WriteAllText(path, sb.ToString()); + var rule = TextRule("RULE-LONG", path, "^hit (?\\d{3})", maxLines: 10); + var collector = new LogParserCollector(); + + for (var run = 0; run < 6; run++) + collector.Execute(rule, _context); + + var expected = Enumerable.Range(1, 30).Select(i => i.ToString("D3")).ToList(); + Assert.Equal(expected, Captured("n")); + } + + [Fact] + public void Line_written_in_two_parts_is_matched_once_and_whole() + { + var path = LogPath("partial.log"); + File.WriteAllText(path, "ERR"); + var rule = TextRule("RULE-PARTIAL", path, "^ERROR (?0x[0-9A-F]+)$"); + var collector = new LogParserCollector(); + + collector.Execute(rule, _context); // the writer is mid-line + File.AppendAllText(path, "OR 0x1F\n"); + collector.Execute(rule, _context); + + Assert.Equal(new List { "0x1F" }, Captured("code")); + } + + [Fact] + public void Interval_rule_reads_a_line_still_unfinished_on_its_second_run_once() + { + var path = LogPath("never-terminated.log"); + File.WriteAllText(path, "ERROR 0x2A"); + var rule = TextRule("RULE-UNFINISHED", path, "^ERROR (?0x[0-9A-F]+)$"); + var collector = new LogParserCollector(); + + collector.Execute(rule, _context); // held: the writer may still be in the line + collector.Execute(rule, _context); // still unfinished — taken as it is + collector.Execute(rule, _context); // nothing new + + Assert.Equal(new List { "0x2A" }, Captured("code")); + } + + [Fact] + public void One_shot_rule_reads_an_unfinished_line_now_and_whole_on_its_next_run() + { + var path = LogPath("startup.log"); + File.WriteAllText(path, "ok\nERR"); + var rule = TextRule("RULE-STARTUP", path, "^ERR(?.*)$", trigger: "startup"); + var collector = new LogParserCollector(); + + collector.Execute(rule, _context); // agent start: the writer is mid-line + File.AppendAllText(path, "OR 0x1F\n"); + collector.Execute(rule, _context); // next agent start + + // Read twice, never split: the second read sees the whole line. + Assert.Equal(new List { "", "OR 0x1F" }, Captured("rest")); + Assert.Equal(new long[] { 2, 2 }, _events.Select(e => Convert.ToInt64(e.Data["logLineNumber"])).ToArray()); + } + + [Fact] + public void Line_feed_written_after_a_trailing_carriage_return_is_consumed_without_a_line() + { + var path = LogPath("cr-at-end.log"); + File.WriteAllText(path, "hit\r"); + var rule = TextRule("RULE-CR", path, "^hit$"); + var collector = new LogParserCollector(); + + collector.Execute(rule, _context); + File.AppendAllText(path, "\n"); + collector.Execute(rule, _context); + + Assert.Single(_events); + var position = _context.SessionState.PeekLogPosition("RULE-CR", path); + Assert.Equal(new FileInfo(path).Length, position!.Position); + Assert.Equal(1, position.LineNumber); + } + + [Fact] + public void Utf16_log_appended_after_the_first_run_is_still_matched() + { + var path = LogPath("transcript.log"); + File.WriteAllText(path, "ERROR first\r\n", Encoding.Unicode); // UTF-16 LE with BOM + var rule = TextRule("RULE-UTF16", path, "^ERROR (?\\w+)$"); + var collector = new LogParserCollector(); + + collector.Execute(rule, _context); + using (var stream = new FileStream(path, FileMode.Append, FileAccess.Write)) + { + var bytes = new UnicodeEncoding(bigEndian: false, byteOrderMark: false).GetBytes("ERROR second\r\n"); + stream.Write(bytes, 0, bytes.Length); + } + collector.Execute(rule, _context); + + Assert.Equal(new List { "first", "second" }, Captured("what")); + } + + [Theory] + [InlineData("abc")] + [InlineData("0")] + [InlineData("-5")] + public void MaxLines_that_is_no_positive_number_falls_back_to_the_default(string maxLines) + { + var path = LogPath("misconfigured.log"); + File.WriteAllText(path, "hit 1\nhit 2\n"); + var rule = TextRule("RULE-MISCONFIGURED", path, "^hit (?\\d)$"); + rule.Parameters["maxLines"] = maxLines; + + new LogParserCollector().Execute(rule, _context); + + Assert.Equal(new List { "1", "2" }, Captured("n")); + } + + [Fact] + public void Line_numbers_count_from_the_start_of_the_file() + { + var path = LogPath("numbers.log"); + File.WriteAllText(path, "a\nhit\nb\n"); + var rule = TextRule("RULE-NUMBERS", path, "^hit$"); + var collector = new LogParserCollector(); + + collector.Execute(rule, _context); + File.AppendAllText(path, "c\nhit\n"); + collector.Execute(rule, _context); + + Assert.Equal(new long[] { 2, 5 }, _events.Select(e => Convert.ToInt64(e.Data["logLineNumber"])).ToArray()); + } + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/RegistryCollectorEmitOnlyIfExistsTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/RegistryCollectorEmitOnlyIfExistsTests.cs index 002c1d15..18bfa2bf 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/RegistryCollectorEmitOnlyIfExistsTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Gather/RegistryCollectorEmitOnlyIfExistsTests.cs @@ -35,7 +35,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring.Gather tenantId: "tenant", onEventCollected: _ => { }, imeLogPathOverride: null, - filePositionTracker: new LogFilePositionTracker()) + sessionState: new GatherRuleSessionState("sess", null, logger)) { // Bypass the registry allowlist guard for the synthetic test path. UnrestrictedMode = true diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Runtime/DiagnosticsPackageServiceTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Runtime/DiagnosticsPackageServiceTests.cs index 043da53b..38f3ef6f 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Runtime/DiagnosticsPackageServiceTests.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Runtime/DiagnosticsPackageServiceTests.cs @@ -19,6 +19,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring.Runtime /// completion markers (whiteglove.complete, clean-exit) — V1 sessions had only AgentLogs/ /// and ImeLogs/, which left forensics blind to decision-engine state and pending uploads. /// + [Collection(UserProfileResolverCollection.Name)] // replaces UserProfileResolver's static state public sealed class DiagnosticsPackageServiceTests { private static AgentConfiguration Cfg(string sessionId = "S1") => new AgentConfiguration diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Runtime/UserProfileResolverTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Runtime/UserProfileResolverTests.cs new file mode 100644 index 00000000..d0f592be --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Monitoring/Runtime/UserProfileResolverTests.cs @@ -0,0 +1,96 @@ +using System; +using AutopilotMonitor.Agent.V2.Core.Monitoring.Runtime; +using Xunit; + +namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring.Runtime +{ + /// + /// Serialises the test classes that replace 's static state. + /// + [CollectionDefinition(Name)] + public sealed class UserProfileResolverCollection + { + public const string Name = "UserProfileResolver"; + } + + /// + /// The agent starts before anyone has signed in. A lookup in that window must not pin + /// "no user" for the rest of the process — the %LOGGED_ON_USER_PROFILE% gather rules and the + /// RealmJoin user logs of the diagnostics package would never resolve. + /// + [Collection(UserProfileResolverCollection.Name)] + public sealed class UserProfileResolverTests : IDisposable + { + private static readonly DateTime Start = new DateTime(2026, 9, 28, 12, 0, 0, DateTimeKind.Utc); + + public void Dispose() => UserProfileResolver.Reset(); + + [Fact] + public void User_who_signs_in_after_the_first_lookup_is_found_later() + { + string? signedIn = null; + var now = Start; + UserProfileResolver.SetDetectorForTesting(() => signedIn, () => now); + + Assert.Null(UserProfileResolver.GetLoggedOnUserProfilePath()); + + signedIn = @"C:\Users\enduser"; + now += UserProfileResolver.RetryInterval; + + Assert.Equal(@"C:\Users\enduser", UserProfileResolver.GetLoggedOnUserProfilePath()); + } + + [Fact] + public void No_user_is_detected_again_only_after_the_retry_interval() + { + var detections = 0; + var now = Start; + UserProfileResolver.SetDetectorForTesting(() => { detections++; return null; }, () => now); + + UserProfileResolver.GetLoggedOnUserProfilePath(); + now += TimeSpan.FromSeconds(10); + UserProfileResolver.GetLoggedOnUserProfilePath(); + UserProfileResolver.GetLoggedOnUserProfilePath(); + Assert.Equal(1, detections); + + now += UserProfileResolver.RetryInterval; + UserProfileResolver.GetLoggedOnUserProfilePath(); + Assert.Equal(2, detections); + } + + [Fact] + public void A_clock_stepping_backwards_does_not_delay_the_next_detection() + { + var detections = 0; + var now = Start; + UserProfileResolver.SetDetectorForTesting(() => { detections++; return null; }, () => now); + + UserProfileResolver.GetLoggedOnUserProfilePath(); + now -= TimeSpan.FromHours(1); // OOBE time sync + UserProfileResolver.GetLoggedOnUserProfilePath(); + + Assert.Equal(2, detections); + } + + [Fact] + public void A_found_user_is_kept_without_detecting_again() + { + var detections = 0; + UserProfileResolver.SetDetectorForTesting(() => { detections++; return @"C:\Users\enduser"; }, () => Start); + + UserProfileResolver.GetLoggedOnUserProfilePath(); + UserProfileResolver.GetLoggedOnUserProfilePath(); + + Assert.Equal(1, detections); + } + + [Fact] + public void Pinned_no_user_stays_null_without_real_detection() + { + UserProfileResolver.SetForTesting(null!); + + Assert.Null(UserProfileResolver.GetLoggedOnUserProfilePath()); + Assert.Null(UserProfileResolver.ExpandCustomTokens(@"%LOGGED_ON_USER_PROFILE%\AppData\Local\x.log")); + } + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/DefaultComponentFactoryGatherStateTests.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/DefaultComponentFactoryGatherStateTests.cs new file mode 100644 index 00000000..ed1980fb --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core.Tests/Orchestration/DefaultComponentFactoryGatherStateTests.cs @@ -0,0 +1,103 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Threading; +using AutopilotMonitor.Agent.V2.Core.Configuration; +using AutopilotMonitor.Agent.V2.Core.Logging; +using AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather; +using AutopilotMonitor.Agent.V2.Core.Orchestration; +using AutopilotMonitor.Agent.V2.Core.Tests.Harness; +using AutopilotMonitor.Shared.Models; +using Xunit; + +namespace AutopilotMonitor.Agent.V2.Core.Tests.Orchestration +{ + /// + /// The production factory hands its state directory to the gather executor. Without it every + /// unit test of the gather state passes while the agent keeps replaying logs after each reboot — + /// the one line in is the whole feature. + /// + [Collection("SerialThreading")] // the startup rule executes on the shared ThreadPool + public sealed class DefaultComponentFactoryGatherStateTests : IDisposable + { + private readonly TempDirectory _tmp = new TempDirectory(); + private readonly TempDirectory _stateDir = new TempDirectory(); + + // %TEMP% lives under C:\Users (hard-blocked) — the log sits beside the test assembly. + private readonly string _logDir = Path.Combine( + AppContext.BaseDirectory, "factory-gather-state-" + Guid.NewGuid().ToString("N")); + + public DefaultComponentFactoryGatherStateTests() => Directory.CreateDirectory(_logDir); + + public void Dispose() + { + _tmp.Dispose(); + _stateDir.Dispose(); + try { Directory.Delete(_logDir, recursive: true); } catch { /* best effort */ } + } + + [Fact] + public void Gather_host_built_by_the_factory_persists_logparser_positions_in_the_state_directory() + { + var log = Path.Combine(_logDir, "vendor.log"); + File.WriteAllText(log, "Update one\n"); + + var factory = new DefaultComponentFactory( + agentConfig: new AgentConfiguration + { + ApiBaseUrl = "https://example", + TenantId = "t", + SessionId = "s", + UnrestrictedMode = true, + }, + remoteConfig: new AgentConfigResponse + { + Collectors = CollectorConfiguration.CreateDefault(), + GatherRules = new List + { + new GatherRule + { + RuleId = "vendor-log", + Title = "vendor log", + CollectorType = "logparser", + Target = log, + Parameters = new Dictionary { ["pattern"] = "^Update", ["format"] = "text" }, + Trigger = "startup", + OutputEventType = "vendor_update", + Enabled = true, + }, + }, + }, + networkMetrics: null, + agentVersion: "test", + stateDirectory: _stateDir.Path, + startupEventGate: null); + + var hosts = factory.CreateCollectorHosts( + sessionId: "s", + tenantId: "t", + logger: new AgentLogger(_tmp.Path, AgentLogLevel.Info), + whiteGloveSealingPatternIds: Array.Empty(), + ingress: new FakeSignalIngressSink(), + clock: new VirtualClock(new DateTime(2026, 9, 28, 12, 0, 0, DateTimeKind.Utc)), + telemetrySpool: null, + timelineEvents: null).Hosts; + try + { + hosts.OfType().Single().Start(); + + var stateFile = Path.Combine(_stateDir.Path, GatherRuleStatePersistence.FileName); + Assert.True(SpinWait.SpinUntil(() => File.Exists(stateFile), TimeSpan.FromSeconds(10)), + "the gather host did not write its state into the factory's state directory"); + } + finally + { + foreach (var host in hosts) + { + try { host.Dispose(); } catch { /* best-effort cleanup */ } + } + } + } + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Enrollment/Ime/LogFilePositionTracker.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Enrollment/Ime/LogFilePositionTracker.cs index 81088d3e..7ff1463b 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Enrollment/Ime/LogFilePositionTracker.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Enrollment/Ime/LogFilePositionTracker.cs @@ -4,9 +4,9 @@ using System.Collections.Generic; namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime { /// - /// Tracks file read positions for incremental log file reading. - /// In-memory only - resets on agent restart, which is acceptable since - /// we start fresh per enrollment session. + /// Tracks the IME log tracker's file read positions for incremental log file reading. + /// The bookmarks survive agent restarts through the tracker's own state file + /// ( / ); the freshness timestamps do not. /// public class LogFilePositionTracker { diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Runtime/DiagnosticsPackageService.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Runtime/DiagnosticsPackageService.cs index 12dff547..8f720029 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Runtime/DiagnosticsPackageService.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Runtime/DiagnosticsPackageService.cs @@ -472,7 +472,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Runtime // Expand BEFORE the guard: the guard must judge the path that is read, // and an unresolved token is not a path it can judge. - var expandedPath = UserProfileResolver.ExpandCustomTokens(entry.Path); + var expandedPath = UserProfileResolver.ExpandCustomTokens(entry.Path, userProfilePath); if (expandedPath == null) { _logger.Warning($"Diagnostics path skipped (no user session for token): {entry.Path}"); diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Runtime/UserProfileResolver.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Runtime/UserProfileResolver.cs index 4f6dce60..d8b31250 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Runtime/UserProfileResolver.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Runtime/UserProfileResolver.cs @@ -1,6 +1,7 @@ using System; using System.Diagnostics; using System.IO; +using System.Threading; using AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment; using AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime; using AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals; @@ -14,7 +15,9 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Runtime /// The agent runs as SYSTEM, so standard environment variables like %USERPROFILE% or /// %LOCALAPPDATA% resolve to the SYSTEM profile — not the logged-on user. This class /// detects the real user via explorer.exe ownership (WTS session query, WMI fallback — - /// see ) and caches the result for the agent's lifetime. + /// see ) and caches the first user it finds for the + /// agent's lifetime. "No user yet" is not cached: the agent usually starts before anyone has + /// signed in, so detection is retried — at most every . /// /// Usage in paths: %LOGGED_ON_USER_PROFILE%\AppData\Local\RealmJoin\Logs\*.log /// @@ -23,9 +26,16 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Runtime // Spelled once, in Shared: the built-in diagnostics catalog uses the same token. public const string Token = Shared.Models.DiagnosticsBuiltInSections.UserProfileToken; + /// Minimum time between two detections while no user has been found. + internal static readonly TimeSpan RetryInterval = TimeSpan.FromSeconds(30); + private static readonly object Lock = new object(); - private static bool _resolved; - private static string _userProfilePath; // e.g. C:\Users\JohnDoe — null if no user detected + private static string _userProfilePath; // e.g. C:\Users\JohnDoe — null until a user is detected + private static DateTime _lastFailedDetectionUtc = DateTime.MinValue; + + // Seams for tests; production uses the real detection and clock. + private static Func _detector = DetectLoggedOnUserProfile; + private static Func _utcNow = () => DateTime.UtcNow; /// /// Returns the logged-on user's profile path (e.g. C:\Users\JohnDoe) or null @@ -34,17 +44,28 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Runtime /// public static string GetLoggedOnUserProfilePath() { - if (_resolved) - return _userProfilePath; + var cached = Volatile.Read(ref _userProfilePath); + if (cached != null) + return cached; lock (Lock) { - if (_resolved) + if (_userProfilePath != null) return _userProfilePath; - _userProfilePath = DetectLoggedOnUserProfile(); - _resolved = true; - return _userProfilePath; + // A clock that stepped backwards (common during OOBE) never delays the retry. + var now = _utcNow(); + var sinceLastFailure = now - _lastFailedDetectionUtc; + if (_lastFailedDetectionUtc != DateTime.MinValue + && sinceLastFailure >= TimeSpan.Zero && sinceLastFailure < RetryInterval) + return null; + + var detected = _detector(); + if (detected != null) + Volatile.Write(ref _userProfilePath, detected); + else + _lastFailedDetectionUtc = now; + return detected; } } @@ -53,18 +74,26 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Runtime /// Environment.ExpandEnvironmentVariables for standard tokens. /// Returns null if the path contains the token but no user is logged on. /// - public static string ExpandCustomTokens(string rawPath) + public static string ExpandCustomTokens(string rawPath) => + ExpandCustomTokens(rawPath, ContainsUserProfileToken(rawPath) ? GetLoggedOnUserProfilePath() : null); + + /// + /// Same, with the profile path the caller already resolved for its guard. Callers that + /// guard the expanded path must use this overload: while detection is still retrying, a + /// second lookup can find the user the first one missed, and the guard would then judge a + /// C:\Users path without the profile that admits it. + /// + public static string ExpandCustomTokens(string rawPath, string userProfilePath) { if (string.IsNullOrEmpty(rawPath)) return rawPath; if (rawPath.IndexOf(Token, StringComparison.OrdinalIgnoreCase) >= 0) { - var profilePath = GetLoggedOnUserProfilePath(); - if (profilePath == null) + if (userProfilePath == null) return null; // No user detected — caller should skip this path - rawPath = ReplaceCaseInsensitive(rawPath, Token, profilePath); + rawPath = ReplaceCaseInsensitive(rawPath, Token, userProfilePath); } return Environment.ExpandEnvironmentVariables(rawPath); @@ -80,26 +109,37 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Runtime } /// - /// Resets the cached state. Only used for testing. + /// Resets the cached state and the seams. Only used for testing. /// internal static void Reset() { lock (Lock) { - _resolved = false; _userProfilePath = null; + _lastFailedDetectionUtc = DateTime.MinValue; + _detector = DetectLoggedOnUserProfile; + _utcNow = () => DateTime.UtcNow; } } /// - /// Allows tests to inject a specific profile path without WMI. + /// Allows tests to inject a specific profile path without WMI. Null pins "no user signed in" + /// — detection keeps answering null instead of finding the developer's own session. /// internal static void SetForTesting(string profilePath) + { + SetDetectorForTesting(() => profilePath, () => DateTime.UtcNow); + } + + /// Replaces detection and clock, clearing what was cached. Only used for testing. + internal static void SetDetectorForTesting(Func detector, Func utcNow) { lock (Lock) { - _userProfilePath = profilePath; - _resolved = true; + _userProfilePath = null; + _lastFailedDetectionUtc = DateTime.MinValue; + _detector = detector ?? throw new ArgumentNullException(nameof(detector)); + _utcNow = utcNow ?? throw new ArgumentNullException(nameof(utcNow)); } } diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/FileCollector.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/FileCollector.cs index abe9c1ca..70bacac6 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/FileCollector.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/FileCollector.cs @@ -23,7 +23,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather.Collectors // Expand custom tokens (%LOGGED_ON_USER_PROFILE%) and standard environment variables var userProfilePath = UserProfileResolver.ContainsUserProfileToken(filePath) ? UserProfileResolver.GetLoggedOnUserProfilePath() : null; - filePath = UserProfileResolver.ExpandCustomTokens(filePath); + filePath = UserProfileResolver.ExpandCustomTokens(filePath, userProfilePath); if (filePath == null) return data; // Token present but no user logged on — skip silently diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/JsonCollector.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/JsonCollector.cs index b09f3463..24417895 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/JsonCollector.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/JsonCollector.cs @@ -24,7 +24,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather.Collectors // Expand custom tokens (%LOGGED_ON_USER_PROFILE%) and standard environment variables var userProfilePath = UserProfileResolver.ContainsUserProfileToken(filePath) ? UserProfileResolver.GetLoggedOnUserProfilePath() : null; - filePath = UserProfileResolver.ExpandCustomTokens(filePath); + filePath = UserProfileResolver.ExpandCustomTokens(filePath, userProfilePath); if (filePath == null) return data; // Token present but no user logged on — skip silently diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/LogLineReader.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/LogLineReader.cs new file mode 100644 index 00000000..5a203761 --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/LogLineReader.cs @@ -0,0 +1,313 @@ +using System; +using System.IO; +using System.Security.Cryptography; +using System.Text; + +namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather.Collectors +{ + /// + /// Text encoding of a log file as the logparser reads it. Chosen from the byte order mark at + /// the start of the file in 's detection order; without one the file + /// is UTF-8, which is also how ANSI logs were always read (invalid bytes decode to U+FFFD). + /// + internal sealed class LogTextEncoding + { + public static readonly LogTextEncoding Utf8 = new LogTextEncoding("utf-8", new UTF8Encoding(false), 1, 0); + public static readonly LogTextEncoding Utf8WithBom = new LogTextEncoding("utf-8 with bom", new UTF8Encoding(false), 1, 3); + public static readonly LogTextEncoding Utf16LittleEndian = new LogTextEncoding("utf-16le", new UnicodeEncoding(false, false), 2, 2); + public static readonly LogTextEncoding Utf16BigEndian = new LogTextEncoding("utf-16be", new UnicodeEncoding(true, false), 2, 2); + public static readonly LogTextEncoding Utf32LittleEndian = new LogTextEncoding("utf-32le", new UTF32Encoding(false, false), 4, 4); + public static readonly LogTextEncoding Utf32BigEndian = new LogTextEncoding("utf-32be", new UTF32Encoding(true, false), 4, 4); + + private LogTextEncoding(string name, Encoding encoding, int unitSize, int byteOrderMarkLength) + { + Name = name; + Encoding = encoding; + UnitSize = unitSize; + ByteOrderMarkLength = byteOrderMarkLength; + LineFeed = encoding.GetBytes("\n"); + CarriageReturn = encoding.GetBytes("\r"); + } + + public string Name { get; } + public Encoding Encoding { get; } + + /// Bytes per code unit; every line boundary lies on a unit boundary. + public int UnitSize { get; } + + /// Length of the byte order mark at offset 0 — skipped, never part of a line. + public int ByteOrderMarkLength { get; } + + public byte[] LineFeed { get; } + public byte[] CarriageReturn { get; } + + public static LogTextEncoding Detect(byte[] head) + { + if (StartsWith(head, 0xEF, 0xBB, 0xBF)) return Utf8WithBom; + if (StartsWith(head, 0xFF, 0xFE, 0x00, 0x00)) return Utf32LittleEndian; + if (StartsWith(head, 0xFF, 0xFE)) return Utf16LittleEndian; + if (StartsWith(head, 0xFE, 0xFF)) return Utf16BigEndian; + if (StartsWith(head, 0x00, 0x00, 0xFE, 0xFF)) return Utf32BigEndian; + return Utf8; + } + + /// The last code-unit boundary at or before . + public long AlignDown(long position) + { + if (position <= ByteOrderMarkLength) return position < ByteOrderMarkLength ? 0 : position; + return position - (position - ByteOrderMarkLength) % UnitSize; + } + + private static bool StartsWith(byte[] bytes, params byte[] prefix) + { + if (bytes.Length < prefix.Length) return false; + for (var i = 0; i < prefix.Length; i++) + if (bytes[i] != prefix[i]) return false; + return true; + } + } + + /// + /// The first bytes of a log file: they carry its byte order mark and identify the file, so a + /// file replaced under the same name is recognised even when it is already longer than the + /// position read so far. + /// + internal sealed class LogFileHead + { + internal const int MaxBytes = 4096; + + private readonly byte[] _bytes; + + private LogFileHead(byte[] bytes) + { + _bytes = bytes; + } + + /// Up to from offset 0. Leaves the stream at an undefined position. + public static LogFileHead Read(Stream stream) + { + stream.Seek(0, SeekOrigin.Begin); + var buffer = new byte[(int)Math.Min(MaxBytes, stream.Length)]; + var read = 0; + while (read < buffer.Length) + { + var n = stream.Read(buffer, read, buffer.Length - read); + if (n <= 0) break; + read += n; + } + if (read < buffer.Length) Array.Resize(ref buffer, read); + return new LogFileHead(buffer); + } + + internal static LogFileHead FromBytes(byte[] bytes) => new LogFileHead(bytes); + + public int Length => _bytes.Length; + + public LogTextEncoding Encoding => LogTextEncoding.Detect(_bytes); + + /// SHA-256 of the first bytes, Base64. + public string Hash(int length) + { + using (var sha = SHA256.Create()) + return Convert.ToBase64String(sha.ComputeHash(_bytes, 0, length)); + } + } + + /// + /// Byte-exact line reader for the logparser collector. is the file + /// offset right behind the last line returned, so a caller that stops after any line can + /// resume exactly there — only exposes its read-ahead boundary. + /// + /// Line ends are those of : "\r\n", "\n" and a lone "\r", + /// in code units of the file's . A "\n" directly after the start + /// position is skipped when the unit before it is "\r": the previous run ended on a "\r" whose + /// "\n" had not been written yet. A line at end of file without a line end is returned with + /// false; the caller decides whether the writer is still in + /// it. A line longer than the cap is consumed without being kept: + /// returns an empty string and sets . + /// + /// + internal sealed class LogLineReader + { + private const int ReadBufferSize = 64 * 1024; + + private readonly Stream _stream; + private readonly LogTextEncoding _encoding; + private readonly int _unit; + private readonly int _maxLineBytes; + private readonly byte[] _buffer = new byte[ReadBufferSize]; + private int _pos; + private int _len; + private long _bufferOffset; + private bool _eof; + private bool _skipLineFeedAfterCarriageReturn; + + private byte[] _line = new byte[4096]; + private int _lineLength; + + /// Seekable stream positioned at a code-unit boundary. + public LogLineReader(Stream stream, LogTextEncoding encoding, int maxLineBytes) + { + if (stream == null) throw new ArgumentNullException(nameof(stream)); + if (encoding == null) throw new ArgumentNullException(nameof(encoding)); + if (maxLineBytes <= 0) throw new ArgumentOutOfRangeException(nameof(maxLineBytes)); + _stream = stream; + _encoding = encoding; + _unit = encoding.UnitSize; + _maxLineBytes = maxLineBytes; + _bufferOffset = stream.Position; + _skipLineFeedAfterCarriageReturn = PrecededByCarriageReturn(); + } + + /// File offset of the next unread byte, exact. + public long Position => _bufferOffset + _pos; + + /// File offset of the first byte of the line most recently returned. + public long LastLineStart { get; private set; } + + /// False for a line that reached end of file without a line end. + public bool LastLineTerminated { get; private set; } + + /// True when the last line was longer than the cap; it was consumed, not returned. + public bool LastLineTruncated { get; private set; } + + /// The next line without its line end, or null at end of file. + public string ReadLine() + { + if (Position == 0 && _encoding.ByteOrderMarkLength > 0 && Ensure(_encoding.ByteOrderMarkLength)) + _pos += _encoding.ByteOrderMarkLength; + + if (_skipLineFeedAfterCarriageReturn) + { + _skipLineFeedAfterCarriageReturn = false; + if (Ensure(_unit) && UnitAt(_pos, _encoding.LineFeed)) + _pos += _unit; + } + + LastLineStart = Position; + LastLineTerminated = false; + LastLineTruncated = false; + _lineLength = 0; + + while (true) + { + if (!Ensure(_unit)) + { + // End of file. A partial code unit (a writer mid-character) belongs to the line. + var rest = _len - _pos; + if (rest > 0) + { + Append(_pos, rest); + _pos = _len; + } + return Position > LastLineStart ? Decode() : null; + } + + // Scan the whole code units in the buffer for the next line end and take the text + // before it in one copy. + var end = _pos + (_len - _pos) / _unit * _unit; + var scan = _pos; + while (scan < end && !IsLineEnd(scan)) + scan += _unit; + if (scan > _pos) + { + Append(_pos, scan - _pos); + _pos = scan; + } + if (scan == end) + continue; // no line end in the buffer yet — refill + + var carriageReturn = !UnitAt(_pos, _encoding.LineFeed); + _pos += _unit; + if (carriageReturn && Ensure(_unit) && UnitAt(_pos, _encoding.LineFeed)) + _pos += _unit; + LastLineTerminated = true; + return Decode(); + } + } + + private bool IsLineEnd(int offset) + { + if (_unit == 1) + { + var b = _buffer[offset]; + return b == (byte)'\n' || b == (byte)'\r'; + } + return UnitAt(offset, _encoding.LineFeed) || UnitAt(offset, _encoding.CarriageReturn); + } + + private bool PrecededByCarriageReturn() + { + var start = _bufferOffset; + if (start - _unit < _encoding.ByteOrderMarkLength) return false; + var previous = new byte[_unit]; + _stream.Seek(start - _unit, SeekOrigin.Begin); + var read = 0; + while (read < _unit) + { + var n = _stream.Read(previous, read, _unit - read); + if (n <= 0) break; + read += n; + } + _stream.Seek(start, SeekOrigin.Begin); + if (read < _unit) return false; + for (var i = 0; i < _unit; i++) + if (previous[i] != _encoding.CarriageReturn[i]) return false; + return true; + } + + /// At least unread bytes in the buffer, unless the file ends first. + private bool Ensure(int count) + { + while (_len - _pos < count) + { + if (_eof) return false; + var remaining = _len - _pos; + if (_pos > 0) + { + Buffer.BlockCopy(_buffer, _pos, _buffer, 0, remaining); + _bufferOffset += _pos; + _pos = 0; + _len = remaining; + } + var read = _stream.Read(_buffer, _len, _buffer.Length - _len); + if (read <= 0) + { + _eof = true; + return _len - _pos >= count; + } + _len += read; + } + return true; + } + + private bool UnitAt(int offset, byte[] unit) + { + for (var i = 0; i < _unit; i++) + if (_buffer[offset + i] != unit[i]) return false; + return true; + } + + private void Append(int offset, int count) + { + var room = _maxLineBytes - _lineLength; + if (count > room) + { + LastLineTruncated = true; + count = room; + if (count <= 0) return; + } + if (_lineLength + count > _line.Length) + { + var grown = new byte[Math.Max(_line.Length * 2, _lineLength + count)]; + Buffer.BlockCopy(_line, 0, grown, 0, _lineLength); + _line = grown; + } + Buffer.BlockCopy(_buffer, offset, _line, _lineLength, count); + _lineLength += count; + } + + private string Decode() => + _lineLength == 0 || LastLineTruncated ? string.Empty : _encoding.Encoding.GetString(_line, 0, _lineLength); + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/LogParserCollector.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/LogParserCollector.cs index 6a5fedb9..8b125ccd 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/LogParserCollector.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/LogParserCollector.cs @@ -18,6 +18,15 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather.Collectors { public string CollectorType => "logparser"; + /// A wildcard target reads at most this many files per run, the newest first. + internal const int MaxFilesPerRun = 20; + + /// Lines one run reads from one file unless the rule sets maxLines. + internal const int DefaultMaxLines = 1000; + + /// The IME tracker's line bound: a longer physical line is skipped, never matched. + internal const int MaxLineBytes = 32 * 1024 * 1024; + /// /// Executes the log parser rule. Returns null because logparser emits events directly /// via context.OnEventCollected rather than returning a single result dictionary. @@ -34,7 +43,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather.Collectors // Expand custom tokens (%LOGGED_ON_USER_PROFILE%) and standard environment variables var userProfilePath = UserProfileResolver.ContainsUserProfileToken(filePath) ? UserProfileResolver.GetLoggedOnUserProfilePath() : null; - filePath = UserProfileResolver.ExpandCustomTokens(filePath); + filePath = UserProfileResolver.ExpandCustomTokens(filePath, userProfilePath); if (filePath == null) { context.DebugLog(rule.RuleId, GatherRuleDebugLog.StageLogParser, @@ -84,10 +93,19 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather.Collectors if (rule.Parameters.TryGetValue("trackPosition", out trackPositionStr)) bool.TryParse(trackPositionStr, out trackPosition); + // A value that is no positive number keeps the default: a run capped at 0 lines would + // never advance its position, and the rule would read nothing for the whole session. string maxLinesStr; - int maxLines = 1000; + int maxLines = DefaultMaxLines; if (rule.Parameters.TryGetValue("maxLines", out maxLinesStr)) - int.TryParse(maxLinesStr, out maxLines); + { + int parsedMaxLines; + if (int.TryParse(maxLinesStr, out parsedMaxLines) && parsedMaxLines > 0) + maxLines = parsedMaxLines; + else + context.DebugLog(rule.RuleId, GatherRuleDebugLog.StageLogParser, + $"maxLines '{maxLinesStr}' is not a positive number — using {DefaultMaxLines}"); + } // Determine format: "cmtrace" (default) or "text" for plain text logs string formatStr; @@ -144,7 +162,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather.Collectors /// as the event data). Stops after matches. /// private static void TraceMatch(GatherRuleContext context, GatherRule rule, string fileName, - int lineNumber, Match match, Regex pattern, int matchNumber) + long lineNumber, Match match, Regex pattern, int matchNumber) { if (matchNumber > MaxTracedMatchesPerFile) { @@ -195,14 +213,14 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather.Collectors try { - // Return matched files sorted by last write time (newest first), capped at 20 + // Return matched files sorted by last write time (newest first), capped at MaxFilesPerRun var allMatches = Directory.GetFiles(directory, fileNamePart); - if (allMatches.Length > 20) + if (allMatches.Length > MaxFilesPerRun) context.DebugLog(ruleId, GatherRuleDebugLog.StageLogParser, - $"wildcard matched {allMatches.Length} files — capped to the 20 newest"); + $"wildcard matched {allMatches.Length} files — capped to the {MaxFilesPerRun} newest"); return allMatches .OrderByDescending(f => new FileInfo(f).LastWriteTimeUtc) - .Take(20) + .Take(MaxFilesPerRun) .ToList(); } catch (Exception ex) @@ -216,183 +234,248 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather.Collectors private static void ProcessLogFile(string filePath, GatherRule rule, Regex pattern, bool trackPosition, int maxLines, bool isTextMode, GatherRuleContext context) { + var fileName = Path.GetFileName(filePath); try { - var fileInfo = new FileInfo(filePath); - long startPosition = trackPosition - ? context.FilePositionTracker.GetSafePosition(filePath, fileInfo.Length) - : 0; + var state = context.SessionState; + var known = trackPosition ? state.PeekLogPosition(rule.RuleId, filePath) : null; - // Nothing new to read - if (startPosition >= fileInfo.Length) + // Nothing new since the last run — decided from the length alone, without opening the file. + var length = new FileInfo(filePath).Length; + if (known != null && known.Position == length) { context.DebugLog(rule.RuleId, GatherRuleDebugLog.StageLogParser, - $"{Path.GetFileName(filePath)}: no new content (position {startPosition} >= length {fileInfo.Length}) — trackPosition=true, nothing re-read"); + $"{fileName}: no new content (position {known.Position} >= length {length}) — trackPosition=true, nothing re-read" + + (known.FromPreviousRun ? "; position saved by a previous agent run" : "")); return; } + // A last line without a line end is usually one its writer is still in. An interval + // rule leaves it for its next run, once. A one-shot trigger may have no next run, so it + // reads the line now but saves the position before it: the next run reads it again, + // whole — at worst a match repeats, it is never split. + var isInterval = string.Equals(rule.Trigger, "interval", StringComparison.OrdinalIgnoreCase); + var mayHoldTail = trackPosition && isInterval; + var tailReadAgainNextRun = false; + int matchCount = 0; int linesRead = 0; + int committedLines = 0; int parseFailCount = 0; int timeoutCount = 0; - long endPosition = startPosition; + int oversizedCount = 0; + long startPosition; + long endPosition; + long lineNumberBase; + long fileLength; + long heldTailStart = -1; + LogFileHead head; + LogTextEncoding encoding; using (var stream = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite)) { + fileLength = stream.Length; + head = LogFileHead.Read(stream); + encoding = head.Encoding; + + LogReadStart start; + startPosition = encoding.AlignDown(GatherRuleSessionState.ResolveStart(known, fileLength, head, out start)); + lineNumberBase = start == LogReadStart.Continue ? known.LineNumber : 0; + TraceReadStart(context, rule, fileName, known, start, fileLength); + stream.Seek(startPosition, SeekOrigin.Begin); + var reader = new LogLineReader(stream, encoding, MaxLineBytes); + endPosition = startPosition; - using (var reader = new StreamReader(stream, Encoding.UTF8)) + while (linesRead < maxLines) { - string line; - while ((line = reader.ReadLine()) != null && linesRead < maxLines) + var line = reader.ReadLine(); + if (line == null) { - linesRead++; - - if (isTextMode) - { - // Text mode: match regex directly against the raw line - Match match; - try - { - match = pattern.Match(line); - } - catch (RegexMatchTimeoutException) - { - timeoutCount++; - continue; - } - if (!match.Success) - continue; - - var data = new Dictionary(); - foreach (var groupName in pattern.GetGroupNames()) - { - if (groupName == "0") continue; - var group = match.Groups[groupName]; - if (group.Success) - data[groupName] = group.Value; - } - - data["logLine"] = TruncateMessage(line, 500); - data["logLineNumber"] = linesRead; - data["logFile"] = Path.GetFileName(filePath); - data["ruleId"] = rule.RuleId; - data["ruleTitle"] = rule.Title; - GatherRuleExecutor.StampRuleMarkers(rule, data); - - var eventType = !string.IsNullOrEmpty(rule.OutputEventType) - ? rule.OutputEventType - : "logparser_match"; - var severity = GatherRuleContext.ParseSeverity(rule.OutputSeverity); - - context.OnEventCollected(new EnrollmentEvent - { - SessionId = context.SessionId, - TenantId = context.TenantId, - Timestamp = DateTime.UtcNow, - EventType = eventType, - Severity = severity, - Source = GatherRuleExecutor.SourceName, - Message = $"Gather: {rule.Title}", - Data = data - }); - - matchCount++; - TraceMatch(context, rule, Path.GetFileName(filePath), linesRead, match, pattern, matchCount); - } - else - { - // CMTrace mode: parse line as CMTrace format, match regex against message - CmTraceLogEntry entry; - if (!CmTraceLogParser.TryParseLine(line, out entry)) - { - parseFailCount++; - continue; - } - - Match match; - try - { - match = pattern.Match(entry.Message); - } - catch (RegexMatchTimeoutException) - { - timeoutCount++; - continue; - } - if (!match.Success) - continue; - - var data = new Dictionary(); - foreach (var groupName in pattern.GetGroupNames()) - { - if (groupName == "0") continue; - var group = match.Groups[groupName]; - if (group.Success) - data[groupName] = group.Value; - } - - // Gather-rule logs are arbitrary files this collector does not - // tail, so no writer offset can be measured for them; the - // reader-zone fallback carries its known defect (see - // ResolveUtcAssumingReaderZone). Behaviour unchanged from before - // the parser split. - var entryUtc = entry.TimestampUtc - ?? (entry.HasTimestamp - ? CmTraceLogParser.ResolveUtcAssumingReaderZone(entry.LocalTimestamp) - : DateTime.UtcNow); - data["logTimestamp"] = entryUtc.ToString("o"); - data["logComponent"] = entry.Component; - data["logType"] = entry.Type; - data["logMessage"] = TruncateMessage(entry.Message, 500); - data["logFile"] = Path.GetFileName(filePath); - data["ruleId"] = rule.RuleId; - data["ruleTitle"] = rule.Title; - GatherRuleExecutor.StampRuleMarkers(rule, data); - - var eventType = !string.IsNullOrEmpty(rule.OutputEventType) - ? rule.OutputEventType - : "logparser_match"; - var severity = MapCmTraceTypeToSeverity(entry.Type, rule.OutputSeverity); - - context.OnEventCollected(new EnrollmentEvent - { - SessionId = context.SessionId, - TenantId = context.TenantId, - Timestamp = entryUtc, - EventType = eventType, - Severity = severity, - Source = GatherRuleExecutor.SourceName, - Message = $"Gather: {rule.Title}", - Data = data - }); - - matchCount++; - TraceMatch(context, rule, Path.GetFileName(filePath), linesRead, match, pattern, matchCount); - } + // End of file. A "\n" the reader skipped behind the "\r" the previous run + // ended on is consumed too — unless a kept tail must be read again. + if (!tailReadAgainNextRun) + endPosition = reader.Position; + break; } - // Capture the final stream position after reading - endPosition = stream.Position; + var unfinished = !reader.LastLineTerminated; + if (unfinished && mayHoldTail + && (known == null || known.HeldTailStart != reader.LastLineStart)) + { + heldTailStart = reader.LastLineStart; + break; + } + + linesRead++; + var lineNumber = lineNumberBase + linesRead; + if (unfinished && trackPosition && !isInterval) + { + tailReadAgainNextRun = true; + } + else + { + endPosition = reader.Position; + committedLines = linesRead; + } + + if (reader.LastLineTruncated) + { + oversizedCount++; + continue; + } + + if (isTextMode) + { + // Text mode: match regex directly against the raw line + Match match; + try + { + match = pattern.Match(line); + } + catch (RegexMatchTimeoutException) + { + timeoutCount++; + continue; + } + if (!match.Success) + continue; + + var data = new Dictionary(); + foreach (var groupName in pattern.GetGroupNames()) + { + if (groupName == "0") continue; + var group = match.Groups[groupName]; + if (group.Success) + data[groupName] = group.Value; + } + + data["logLine"] = TruncateMessage(line, 500); + data["logLineNumber"] = lineNumber; + data["logFile"] = fileName; + data["ruleId"] = rule.RuleId; + data["ruleTitle"] = rule.Title; + GatherRuleExecutor.StampRuleMarkers(rule, data); + + var eventType = !string.IsNullOrEmpty(rule.OutputEventType) + ? rule.OutputEventType + : "logparser_match"; + var severity = GatherRuleContext.ParseSeverity(rule.OutputSeverity); + + context.OnEventCollected(new EnrollmentEvent + { + SessionId = context.SessionId, + TenantId = context.TenantId, + Timestamp = DateTime.UtcNow, + EventType = eventType, + Severity = severity, + Source = GatherRuleExecutor.SourceName, + Message = $"Gather: {rule.Title}", + Data = data + }); + + matchCount++; + TraceMatch(context, rule, fileName, lineNumber, match, pattern, matchCount); + } + else + { + // CMTrace mode: parse line as CMTrace format, match regex against message + CmTraceLogEntry entry; + if (!CmTraceLogParser.TryParseLine(line, out entry)) + { + parseFailCount++; + continue; + } + + Match match; + try + { + match = pattern.Match(entry.Message); + } + catch (RegexMatchTimeoutException) + { + timeoutCount++; + continue; + } + if (!match.Success) + continue; + + var data = new Dictionary(); + foreach (var groupName in pattern.GetGroupNames()) + { + if (groupName == "0") continue; + var group = match.Groups[groupName]; + if (group.Success) + data[groupName] = group.Value; + } + + // Gather-rule logs are arbitrary files this collector does not + // tail, so no writer offset can be measured for them; the + // reader-zone fallback carries its known defect (see + // ResolveUtcAssumingReaderZone). Behaviour unchanged from before + // the parser split. + var entryUtc = entry.TimestampUtc + ?? (entry.HasTimestamp + ? CmTraceLogParser.ResolveUtcAssumingReaderZone(entry.LocalTimestamp) + : DateTime.UtcNow); + data["logTimestamp"] = entryUtc.ToString("o"); + data["logComponent"] = entry.Component; + data["logType"] = entry.Type; + data["logMessage"] = TruncateMessage(entry.Message, 500); + data["logFile"] = fileName; + data["ruleId"] = rule.RuleId; + data["ruleTitle"] = rule.Title; + GatherRuleExecutor.StampRuleMarkers(rule, data); + + var eventType = !string.IsNullOrEmpty(rule.OutputEventType) + ? rule.OutputEventType + : "logparser_match"; + var severity = MapCmTraceTypeToSeverity(entry.Type, rule.OutputSeverity); + + context.OnEventCollected(new EnrollmentEvent + { + SessionId = context.SessionId, + TenantId = context.TenantId, + Timestamp = entryUtc, + EventType = eventType, + Severity = severity, + Source = GatherRuleExecutor.SourceName, + Message = $"Gather: {rule.Title}", + Data = data + }); + + matchCount++; + TraceMatch(context, rule, fileName, lineNumber, match, pattern, matchCount); + } } } + // Saved only now, after every event of this file went out: a process that dies + // before this line repeats the file's new lines after the restart, never skips them. if (trackPosition) - context.FilePositionTracker.SetPosition(filePath, endPosition); + state.CommitLogPosition(rule.RuleId, filePath, endPosition, lineNumberBase + committedLines, head, heldTailStart); if (matchCount > 0) - context.Logger.Debug($"LogParser rule {rule.RuleId}: {matchCount} matches from {linesRead} lines in {Path.GetFileName(filePath)}"); + context.Logger.Debug($"LogParser rule {rule.RuleId}: {matchCount} matches from {linesRead} lines in {fileName}"); // Per-file outcome — always written so a "matched 0" run is visible too. context.DebugLog(rule.RuleId, GatherRuleDebugLog.StageLogParser, - $"{Path.GetFileName(filePath)}: read {linesRead} lines from position {startPosition}->{endPosition}, " + - $"matched {matchCount}, parseFailures={parseFailCount}, regexTimeouts={timeoutCount}, mode={(isTextMode ? "text" : "cmtrace")}"); + $"{fileName}: read {linesRead} lines from position {startPosition}->{endPosition}, " + + $"matched {matchCount}, parseFailures={parseFailCount}, regexTimeouts={timeoutCount}, mode={(isTextMode ? "text" : "cmtrace")}, encoding={encoding.Name}" + + (oversizedCount > 0 ? $", skipped {oversizedCount} line(s) over {MaxLineBytes / (1024 * 1024)} MB" : "")); + + if (heldTailStart >= 0) + context.DebugLog(rule.RuleId, GatherRuleDebugLog.StageLogParser, + $"{fileName}: last line has no line end yet — left for the next run, read then even if still unfinished"); + else if (tailReadAgainNextRun) + context.DebugLog(rule.RuleId, GatherRuleDebugLog.StageLogParser, + $"{fileName}: last line has no line end yet — read now and again by the next run, position kept before it"); if (!isTextMode && linesRead > 0 && parseFailCount == linesRead) context.DebugLog(rule.RuleId, GatherRuleDebugLog.StageLogParser, "every line failed CMTrace parsing — if this is a plain-text log, set parameter format=text"); - if (linesRead == maxLines) + if (linesRead == maxLines && endPosition < fileLength) context.DebugLog(rule.RuleId, GatherRuleDebugLog.StageLogParser, $"stopped at maxLines={maxLines} — remaining content is deferred to the next run"); } @@ -403,6 +486,27 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather.Collectors } } + private static void TraceReadStart(GatherRuleContext context, GatherRule rule, string fileName, + LogPositionSnapshot known, LogReadStart start, long fileLength) + { + if (start == LogReadStart.Shrunk) + { + context.DebugLog(rule.RuleId, GatherRuleDebugLog.StageLogParser, + $"{fileName}: file is shorter than the saved position (length {fileLength} < {known.Position}) — truncated or recreated, reading from the beginning"); + } + else if (start == LogReadStart.Replaced) + { + context.Logger.Info($"LogParser rule {rule.RuleId}: {fileName} was replaced since the last read — reading it from the beginning"); + context.DebugLog(rule.RuleId, GatherRuleDebugLog.StageLogParser, + $"{fileName}: file start changed since the last read (replaced or rewritten) — saved position {known.Position} discarded, reading from the beginning"); + } + else if (start == LogReadStart.Continue && known.FromPreviousRun) + { + context.DebugLog(rule.RuleId, GatherRuleDebugLog.StageLogParser, + $"{fileName}: continuing at position {known.Position} (after line {known.LineNumber}) saved by a previous agent run"); + } + } + private static EventSeverity MapCmTraceTypeToSeverity(int cmTraceType, string ruleOverride) { // If the rule specifies a severity, use it diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/XmlCollector.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/XmlCollector.cs index e906523f..2dde9250 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/XmlCollector.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/Collectors/XmlCollector.cs @@ -26,7 +26,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather.Collectors // Expand custom tokens (%LOGGED_ON_USER_PROFILE%) and standard environment variables var userProfilePath = UserProfileResolver.ContainsUserProfileToken(filePath) ? UserProfileResolver.GetLoggedOnUserProfilePath() : null; - filePath = UserProfileResolver.ExpandCustomTokens(filePath); + filePath = UserProfileResolver.ExpandCustomTokens(filePath, userProfilePath); if (filePath == null) return data; // Token present but no user logged on — skip silently diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/GatherRuleContext.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/GatherRuleContext.cs index 23eca47f..405ff28b 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/GatherRuleContext.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/GatherRuleContext.cs @@ -4,7 +4,6 @@ using AutopilotMonitor.Agent.V2.Core.Logging; using AutopilotMonitor.Shared; using AutopilotMonitor.Shared.Models; using AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment; -using AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime; using AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals; namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather @@ -19,7 +18,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather public string TenantId { get; } public Action OnEventCollected { get; } public string ImeLogPathOverride { get; } - public LogFilePositionTracker FilePositionTracker { get; } + + /// Logparser positions per (rule, file) and on_change values; survives agent restarts when persisted. + public GatherRuleSessionState SessionState { get; } + public bool UnrestrictedMode { get; set; } /// @@ -34,7 +36,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather string tenantId, Action onEventCollected, string imeLogPathOverride, - LogFilePositionTracker filePositionTracker, + GatherRuleSessionState sessionState, GatherRuleDebugLog debugLog = null) { Logger = logger ?? throw new ArgumentNullException(nameof(logger)); @@ -42,7 +44,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather TenantId = tenantId ?? throw new ArgumentNullException(nameof(tenantId)); OnEventCollected = onEventCollected ?? throw new ArgumentNullException(nameof(onEventCollected)); ImeLogPathOverride = imeLogPathOverride; - FilePositionTracker = filePositionTracker ?? throw new ArgumentNullException(nameof(filePositionTracker)); + SessionState = sessionState ?? throw new ArgumentNullException(nameof(sessionState)); DebugWriter = debugLog; } diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/GatherRuleExecutor.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/GatherRuleExecutor.cs index 6299fd8c..3107733c 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/GatherRuleExecutor.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/GatherRuleExecutor.cs @@ -10,7 +10,6 @@ using AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather.Collectors; using AutopilotMonitor.Shared; using AutopilotMonitor.Shared.Models; using AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment; -using AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime; using AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.SystemSignals; namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather @@ -38,6 +37,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather private readonly AgentLogger _logger; private readonly GatherRuleDebugLog _debug; // null unless EnableGatherRuleDebugLog / --gather-debug-log + private readonly GatherRuleSessionState _state; private readonly GatherRuleContext _context; private readonly Dictionary _collectors; @@ -50,12 +50,15 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather // Phase-scope + emit-mode state (ActivePhases / ActiveFromPhase / EmitMode on GatherRule). // Guarded by _scopeLock; collector work always runs outside the lock. The state deliberately - // survives UpdateRules config refreshes — an agent restart causes at most one re-emit per - // on_change rule (acceptable under the per-enrollment lifecycle). + // survives UpdateRules config refreshes. Across agent restarts only the last emitted hash + // survives: it is seeded from _state and committed there after each emit, so an unchanged + // on_change result stays silent after a reboot. Phase latches, dedup sets and the + // suppression streak are process-scoped. private readonly object _scopeLock = new object(); private EnrollmentPhase _currentPhase = EnrollmentPhase.Unknown; private readonly HashSet _fromPhaseLatched = new HashSet(); private readonly Dictionary _lastEmittedHash = new Dictionary(); + private readonly HashSet _hashFromPreviousRun = new HashSet(); private readonly Dictionary _suppressed = new Dictionary(); private readonly HashSet _scopeBypassLogged = new HashSet(); @@ -77,9 +80,14 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather set { _context.UnrestrictedMode = value; } } + /// + /// Where logparser positions and on_change values are kept across agent restarts. Null keeps + /// them in memory — required for --run-gather-rules, which runs beside the live agent + /// under a throwaway session id and must not overwrite the session's file. + /// public GatherRuleExecutor(string sessionId, string tenantId, Action onEventCollected, AgentLogger logger, string imeLogPathOverride = null, string debugLogPath = null, - Action debugEcho = null) + Action debugEcho = null, string stateDirectory = null) { if (sessionId == null) throw new ArgumentNullException(nameof(sessionId)); if (tenantId == null) throw new ArgumentNullException(nameof(tenantId)); @@ -89,8 +97,19 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather _logger = logger; _debug = string.IsNullOrEmpty(debugLogPath) ? null : new GatherRuleDebugLog(debugLogPath, logger, debugEcho); - var filePositionTracker = new LogFilePositionTracker(); - _context = new GatherRuleContext(logger, sessionId, tenantId, onEventCollected, imeLogPathOverride, filePositionTracker, _debug); + var persistence = string.IsNullOrEmpty(stateDirectory) ? null : new GatherRuleStatePersistence(stateDirectory, logger); + _state = new GatherRuleSessionState(sessionId, persistence, logger); + _context = new GatherRuleContext(logger, sessionId, tenantId, onEventCollected, imeLogPathOverride, _state, _debug); + + foreach (var hash in _state.GetEmittedHashes()) + { + _lastEmittedHash[hash.Key] = hash.Value; + _hashFromPreviousRun.Add(hash.Key); + } + _logger.Info($"GatherRuleExecutor: gather state {_state.RestoreSummary}"); + DebugLog(null, GatherRuleDebugLog.StageConfig, $"gather state: {_state.RestoreSummary}"); + foreach (var detail in _state.RestoredDetails) + DebugLog(detail.Key, GatherRuleDebugLog.StageConfig, detail.Value); // Register all collector strategies var collectorList = new IGatherRuleCollector[] @@ -480,14 +499,17 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather // empty result never reaches this point, so an emitOnlyIfExists miss neither // emits nor updates the hash — composing to "one event on appearance, then // only on change". + string emittedHash = null; if (IsOnChangeMode(rule)) { string hashPrefix; int suppressedStreak; - if (!ShouldEmitOnChange(rule, result, out hashPrefix, out suppressedStreak)) + bool comparedWithPreviousRun; + if (!ShouldEmitOnChange(rule, result, out emittedHash, out hashPrefix, out suppressedStreak, out comparedWithPreviousRun)) { DebugLog(rule.RuleId, GatherRuleDebugLog.StageSuppress, - $"result unchanged (hash {hashPrefix}) — emit suppressed (on_change), suppressed streak={suppressedStreak}"); + $"result unchanged (hash {hashPrefix}) — emit suppressed (on_change), suppressed streak={suppressedStreak}" + + (comparedWithPreviousRun ? "; compared with the value a previous agent run emitted" : "")); return; } DebugLog(rule.RuleId, GatherRuleDebugLog.StageEmit, @@ -524,6 +546,12 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather Message = $"Gather: {rule.Title}", Data = result }); + + // Committed only after the event went out: a process that dies in between emits + // the result once more after the restart instead of never. + if (emittedHash != null) + _state.CommitEmittedHash(rule.RuleId, emittedHash); + DebugLog(rule.RuleId, GatherRuleDebugLog.StageEmit, $"emitted {eventType} (severity={severity}, {result.Count} fields)"); } @@ -679,9 +707,23 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather internal bool ShouldEmitOnChange(GatherRule rule, Dictionary result, out string hashPrefix, out int suppressedStreak) { - var hash = ComputeCanonicalHash(result); + string hash; + bool comparedWithPreviousRun; + return ShouldEmitOnChange(rule, result, out hash, out hashPrefix, out suppressedStreak, out comparedWithPreviousRun); + } + + /// + /// Full form: is the result's canonical hash, committed to the session + /// state once the event went out; says a suppression + /// compared against the value a previous agent run emitted. + /// + private bool ShouldEmitOnChange(GatherRule rule, Dictionary result, + out string hash, out string hashPrefix, out int suppressedStreak, out bool comparedWithPreviousRun) + { + hash = ComputeCanonicalHash(result); hashPrefix = hash.Length > 8 ? hash.Substring(0, 8) : hash; suppressedStreak = 0; + comparedWithPreviousRun = false; bool emit; lock (_scopeLock) { @@ -693,11 +735,13 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather ? (streak.Count + 1, streak.SinceUtc) : (1, DateTime.UtcNow); suppressedStreak = _suppressed[rule.RuleId].Count; + comparedWithPreviousRun = _hashFromPreviousRun.Contains(rule.RuleId); emit = false; } else { _lastEmittedHash[rule.RuleId] = hash; + _hashFromPreviousRun.Remove(rule.RuleId); (int Count, DateTime SinceUtc) streak; if (_suppressed.TryGetValue(rule.RuleId, out streak) && streak.Count > 0) { @@ -820,6 +864,9 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather StopAllTimers(); _startupRulesLatch?.Dispose(); _startupRulesLatch = null; + // A rule still finishing on a pool thread keeps its in-memory state but no longer writes: + // the agent may be removing its own folder right now. + _state.Close(); } } } diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/GatherRuleSessionState.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/GatherRuleSessionState.cs new file mode 100644 index 00000000..49ef4adc --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/GatherRuleSessionState.cs @@ -0,0 +1,399 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using AutopilotMonitor.Agent.V2.Core.Logging; +using AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather.Collectors; + +namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather +{ + /// + /// Gather-rule state of one enrollment session that survives agent restarts: where each + /// logparser rule stopped reading each file, and the on_change value each rule last emitted. + /// + /// Positions are kept per (rule, file) — two rules parsing the same log each read all of it. + /// Every position carries a fingerprint of the file's first bytes, so a file replaced under + /// the same name is read from the beginning even when it is already longer than the position. + /// + /// + /// Every commit is written to disk synchronously, AFTER the events it covers were handed on: + /// a process that dies in between repeats that one pass after the restart instead of losing it. + /// The lock guards memory and file write together and is never taken under the executor's + /// scope lock. Without a persistence (diagnostic mode, tests) the state lives in memory only. + /// + /// + public sealed class GatherRuleSessionState + { + /// + /// Files remembered per rule. A run reads at most + /// files; the least recently touched entry beyond the cap is dropped, and its file — should it + /// come back — is simply read again. + /// + internal const int MaxLogFilesPerRule = 5 * LogParserCollector.MaxFilesPerRun; + + private static readonly char[] WildcardChars = { '*', '?' }; + + private readonly object _gate = new object(); + private readonly string _sessionId; + private readonly GatherRuleStatePersistence _persistence; + private readonly AgentLogger _logger; + private readonly Dictionary> _logPositions = + new Dictionary>(StringComparer.Ordinal); + private readonly Dictionary _emittedHashes = new Dictionary(StringComparer.Ordinal); + private readonly List> _restoredDetails = new List>(); + private long _touchCounter; + private bool _unsaved; + private bool _saveFailureLogged; + private bool _closed; + + private sealed class Entry + { + public long Position; + public long LineNumber; + public int HeadLength; + public string HeadHash; + public bool FromPreviousRun; + public long HeldTailStart = -1; + public long LastTouched; + } + + public GatherRuleSessionState(string sessionId, GatherRuleStatePersistence persistence, AgentLogger logger) + { + _sessionId = sessionId ?? throw new ArgumentNullException(nameof(sessionId)); + _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + _persistence = persistence; + RestoreSummary = Restore(); + } + + /// One line on what the constructor found on disk — for the agent log and the debug trace. + public string RestoreSummary { get; } + + /// (ruleId, message) per restored position and on_change value — for the debug trace. + internal IReadOnlyList> RestoredDetails => _restoredDetails; + + // ===== logparser positions ===== + + /// The recorded position of for the rule, or null when it has none. + internal LogPositionSnapshot PeekLogPosition(string ruleId, string filePath) + { + lock (_gate) + { + Dictionary files; + Entry entry; + if (!_logPositions.TryGetValue(ruleId, out files) || !files.TryGetValue(filePath, out entry)) + return null; + entry.LastTouched = ++_touchCounter; + return new LogPositionSnapshot(entry.Position, entry.LineNumber, entry.HeadLength, entry.HeadHash, + entry.FromPreviousRun, entry.HeldTailStart); + } + } + + /// + /// Records where the rule stopped in the file and saves the state when it changed. + /// (in memory only) marks an unfinished last line left for + /// the next run; -1 when none. + /// + internal void CommitLogPosition(string ruleId, string filePath, long position, long lineNumber, + LogFileHead head, long heldTailStart) + { + var headLength = head.Length; + var headHash = headLength > 0 ? head.Hash(headLength) : null; + + lock (_gate) + { + Dictionary files; + if (!_logPositions.TryGetValue(ruleId, out files)) + { + files = new Dictionary(StringComparer.OrdinalIgnoreCase); + _logPositions[ruleId] = files; + } + + Entry entry; + var changed = false; + if (!files.TryGetValue(filePath, out entry)) + { + entry = new Entry(); + files[filePath] = entry; + changed = true; + } + + if (changed || entry.Position != position || entry.LineNumber != lineNumber + || entry.HeadLength != headLength || !string.Equals(entry.HeadHash, headHash, StringComparison.Ordinal)) + { + entry.Position = position; + entry.LineNumber = lineNumber; + entry.HeadLength = headLength; + entry.HeadHash = headHash; + entry.FromPreviousRun = false; + changed = true; + } + entry.HeldTailStart = heldTailStart; + entry.LastTouched = ++_touchCounter; + + if (files.Count > MaxLogFilesPerRule) + EvictLeastRecentlyTouched(ruleId, files); + + if (changed || _unsaved) + SaveLocked(); + } + } + + /// + /// Where to continue reading: the recorded position while the file is still the one it was + /// recorded for, otherwise 0 with the reason. + /// + internal static long ResolveStart(LogPositionSnapshot known, long fileLength, LogFileHead head, out LogReadStart start) + { + if (known == null) + { + start = LogReadStart.FirstRead; + return 0; + } + if (fileLength < known.Position) + { + start = LogReadStart.Shrunk; + return 0; + } + if (known.HeadLength > 0 + && (head.Length < known.HeadLength + || !string.Equals(head.Hash(known.HeadLength), known.HeadHash, StringComparison.Ordinal))) + { + start = LogReadStart.Replaced; + return 0; + } + start = LogReadStart.Continue; + return known.Position; + } + + // ===== on_change values ===== + + /// Rule id → hash of the result the rule last emitted. + internal Dictionary GetEmittedHashes() + { + lock (_gate) + return new Dictionary(_emittedHashes, StringComparer.Ordinal); + } + + /// Records the hash of a result that was just emitted and saves the state when it changed. + internal void CommitEmittedHash(string ruleId, string hash) + { + lock (_gate) + { + string current; + var changed = !_emittedHashes.TryGetValue(ruleId, out current) + || !string.Equals(current, hash, StringComparison.Ordinal); + _emittedHashes[ruleId] = hash; + if (changed || _unsaved) + SaveLocked(); + } + } + + /// Retries a save that failed earlier, then stops writing (executor disposed). + internal void Close() + { + lock (_gate) + { + if (_unsaved) + SaveLocked(); + _closed = true; + } + } + + // ===== persistence ===== + + private void SaveLocked() + { + if (_persistence == null || _closed) + { + _unsaved = false; + return; + } + + var data = new GatherRuleStateData + { + SessionId = _sessionId, + SavedAtUtc = DateTime.UtcNow, + LogPositions = _logPositions + .SelectMany(rule => rule.Value.Select(file => new { RuleId = rule.Key, Path = file.Key, Entry = file.Value })) + .OrderBy(x => x.Entry.LastTouched) + .Select(x => new GatherLogPositionData + { + RuleId = x.RuleId, + Path = x.Path, + Position = x.Entry.Position, + LineNumber = x.Entry.LineNumber, + HeadLength = x.Entry.HeadLength, + HeadHash = x.Entry.HeadHash, + }) + .ToList(), + OnChangeHashes = new Dictionary(_emittedHashes, StringComparer.Ordinal), + }; + + if (_persistence.Save(data)) + { + _unsaved = false; + _saveFailureLogged = false; + return; + } + + _unsaved = true; + if (!_saveFailureLogged) + { + _saveFailureLogged = true; + _logger.Warning($"GatherRuleSessionState: saving {_persistence.StateFilePath} failed — kept in memory, retried at the next change"); + } + } + + private string Restore() + { + if (_persistence == null) + return "persistence off (no state directory) — positions and on_change values live in memory only"; + + string discardReason; + var data = _persistence.Load(_sessionId, out discardReason); + if (data == null) + { + return discardReason == null + ? "no saved state — first agent run of this session" + : $"saved state discarded ({discardReason}) — starting fresh"; + } + + // The file is input like any other: whatever it holds, the executor must come up. A + // failure here would otherwise stop every collector host, at every restart of the session. + try + { + return RestoreFrom(data); + } + catch (Exception ex) + { + _logPositions.Clear(); + _emittedHashes.Clear(); + _restoredDetails.Clear(); + _logger.Warning($"GatherRuleSessionState: restoring {_persistence.StateFilePath} failed, starting fresh: {ex.Message}"); + return $"saved state discarded (restore failed: {ex.Message}) — starting fresh"; + } + } + + private string RestoreFrom(GatherRuleStateData data) + { + var dropped = 0; + foreach (var position in data.LogPositions ?? new List()) + { + if (!IsValid(position)) + { + dropped++; + continue; + } + + Dictionary files; + if (!_logPositions.TryGetValue(position.RuleId, out files)) + { + files = new Dictionary(StringComparer.OrdinalIgnoreCase); + _logPositions[position.RuleId] = files; + } + files[position.Path] = new Entry + { + Position = position.Position, + LineNumber = position.LineNumber, + HeadLength = position.HeadLength, + HeadHash = position.HeadLength > 0 ? position.HeadHash : null, + FromPreviousRun = true, + LastTouched = ++_touchCounter, + }; + if (files.Count > MaxLogFilesPerRule) + EvictLeastRecentlyTouched(position.RuleId, files); + } + + foreach (var hash in data.OnChangeHashes ?? new Dictionary()) + { + if (string.IsNullOrEmpty(hash.Key) || string.IsNullOrEmpty(hash.Value)) + { + dropped++; + continue; + } + _emittedHashes[hash.Key] = hash.Value; + } + + foreach (var rule in _logPositions) + { + foreach (var file in rule.Value) + { + _restoredDetails.Add(new KeyValuePair(rule.Key, + $"logparser position restored: {Path.GetFileName(file.Key)} at byte {file.Value.Position}, after line {file.Value.LineNumber}")); + } + } + foreach (var hash in _emittedHashes) + { + _restoredDetails.Add(new KeyValuePair(hash.Key, + $"on_change value restored (hash {Prefix(hash.Value)}) — an unchanged result stays silent")); + } + + var positionCount = _logPositions.Sum(rule => rule.Value.Count); + return $"restored {positionCount} logparser position(s) for {_logPositions.Count} rule(s) and {_emittedHashes.Count} on_change value(s), " + + $"saved {data.SavedAtUtc:yyyy-MM-ddTHH:mm:ssZ} by a previous agent run of this session" + + (dropped > 0 ? $"; {dropped} invalid entr{(dropped == 1 ? "y" : "ies")} dropped" : "") + + " — logparser rules continue after their saved position, on_change rules stay silent until their result changes"; + } + + private static bool IsValid(GatherLogPositionData position) + { + if (position == null || string.IsNullOrEmpty(position.RuleId) || string.IsNullOrEmpty(position.Path)) + return false; + // A real resolved path never carries these; Path APIs throw on them under .NET Framework. + if (position.Path.IndexOfAny(Path.GetInvalidPathChars()) >= 0 || position.Path.IndexOfAny(WildcardChars) >= 0) + return false; + if (position.Position < 0 || position.LineNumber < 0) + return false; + if (position.HeadLength < 0 || position.HeadLength > LogFileHead.MaxBytes) + return false; + if (position.HeadLength > 0 && string.IsNullOrEmpty(position.HeadHash)) + return false; + // Reading any byte records a non-empty head; a position without one cannot be verified. + return position.Position == 0 || position.HeadLength > 0; + } + + private void EvictLeastRecentlyTouched(string ruleId, Dictionary files) + { + var oldest = files.OrderBy(file => file.Value.LastTouched).First(); + files.Remove(oldest.Key); + _logger.Info($"GatherRuleSessionState: rule {ruleId} tracks more than {MaxLogFilesPerRule} files — forgot the position of {Path.GetFileName(oldest.Key)}"); + } + + private static string Prefix(string hash) => hash.Length > 8 ? hash.Substring(0, 8) : hash; + } + + /// Immutable copy of a recorded logparser position. + internal sealed class LogPositionSnapshot + { + public LogPositionSnapshot(long position, long lineNumber, int headLength, string headHash, + bool fromPreviousRun, long heldTailStart) + { + Position = position; + LineNumber = lineNumber; + HeadLength = headLength; + HeadHash = headHash; + FromPreviousRun = fromPreviousRun; + HeldTailStart = heldTailStart; + } + + public long Position { get; } + public long LineNumber { get; } + public int HeadLength { get; } + public string HeadHash { get; } + + /// Restored from disk and not advanced by this process yet. + public bool FromPreviousRun { get; } + + /// Start of an unfinished last line left for this run by the previous one; -1 when none. + public long HeldTailStart { get; } + } + + /// How decided where reading starts. + internal enum LogReadStart + { + FirstRead, + Continue, + Shrunk, + Replaced, + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/GatherRuleStatePersistence.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/GatherRuleStatePersistence.cs new file mode 100644 index 00000000..e5564337 --- /dev/null +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Monitoring/Telemetry/Gather/GatherRuleStatePersistence.cs @@ -0,0 +1,131 @@ +using System; +using System.Collections.Generic; +using System.IO; +using AutopilotMonitor.Agent.V2.Core.Logging; +using Newtonsoft.Json; + +namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather +{ + /// + /// Persists in the agent's state directory so one + /// enrollment's logparser positions and on_change values survive agent restarts, reboots and + /// the WhiteGlove Part 2 resume. + /// + /// The state is only resumed for the SAME session id: the state directory is one fixed folder + /// that a session rotation does not clear, and the new session starts empty in the backend, so + /// it must read its logs from the beginning. + /// + /// Fail-soft like BandwidthStatePersistence: I/O errors never throw; a missing, corrupt + /// or foreign file loads as null (fresh start — the pre-persistence behavior). + /// + public class GatherRuleStatePersistence + { + public const string FileName = "gather-rule-state.json"; + + private const string SourceName = "GatherRuleStatePersistence"; + + private readonly string _stateDirectory; + private readonly string _stateFilePath; + private readonly AgentLogger _logger; + + public GatherRuleStatePersistence(string stateDirectory, AgentLogger logger) + { + _stateDirectory = Environment.ExpandEnvironmentVariables(stateDirectory); + _stateFilePath = Path.Combine(_stateDirectory, FileName); + _logger = logger; + } + + public string StateFilePath => _stateFilePath; + + /// + /// Loads the persisted state, or null when absent, unreadable, or written for a different + /// session. is null when there simply was no file. + /// + public GatherRuleStateData Load(string expectedSessionId, out string discardReason) + { + discardReason = null; + if (!File.Exists(_stateFilePath)) return null; + try + { + var json = File.ReadAllText(_stateFilePath); + var state = JsonConvert.DeserializeObject(json); + if (state == null || string.IsNullOrEmpty(state.SessionId)) + { + discardReason = "state file empty or invalid"; + _logger.Warning($"[{SourceName}] persisted state file was empty or invalid — starting fresh"); + return null; + } + if (!string.Equals(state.SessionId, expectedSessionId, StringComparison.OrdinalIgnoreCase)) + { + discardReason = $"saved for session {state.SessionId}, current session is {expectedSessionId}"; + _logger.Warning($"[{SourceName}] persisted state belongs to session {state.SessionId}, not {expectedSessionId} — starting fresh"); + return null; + } + return state; + } + catch (Exception ex) + { + discardReason = $"state file unreadable: {ex.Message}"; + _logger.Warning($"[{SourceName}] failed to load persisted state, starting fresh: {ex.Message}"); + return null; + } + } + + /// Persists the state (atomic write via temp file). Never throws; false when the write failed. + public bool Save(GatherRuleStateData state) + { + try + { + Directory.CreateDirectory(_stateDirectory); + var tempPath = _stateFilePath + ".tmp"; + File.WriteAllText(tempPath, JsonConvert.SerializeObject(state)); + if (File.Exists(_stateFilePath)) + { + File.Replace(tempPath, _stateFilePath, destinationBackupFileName: null, ignoreMetadataErrors: true); + } + else + { + File.Move(tempPath, _stateFilePath); + } + return true; + } + catch (Exception ex) + { + _logger.Debug($"[{SourceName}] failed to save state: {ex.Message}"); + return false; + } + } + } + + // ---- DTOs for state serialization ---- + + public class GatherRuleStateData + { + public string SessionId { get; set; } + public DateTime SavedAtUtc { get; set; } + + /// Least recently used first, so a restore keeps the eviction order. + public List LogPositions { get; set; } + + /// Rule id → canonical hash of the result the rule last emitted (on_change). + public Dictionary OnChangeHashes { get; set; } + } + + public class GatherLogPositionData + { + public string RuleId { get; set; } + public string Path { get; set; } + + /// File offset right behind the last line the rule processed. + public long Position { get; set; } + + /// Lines before — the base for logLineNumber. + public long LineNumber { get; set; } + + /// Bytes from the start of the file the fingerprint covers (0..4096). + public int HeadLength { get; set; } + + /// SHA-256 of those bytes, Base64; null when is 0. + public string HeadHash { get; set; } + } +} diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DefaultComponentFactory.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DefaultComponentFactory.cs index 26772a80..c0e41f09 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DefaultComponentFactory.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/DefaultComponentFactory.cs @@ -736,7 +736,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration imeLogPathOverride: _agentConfig.ImeLogPathOverride, unrestrictedMode: _agentConfig.UnrestrictedMode, gatherDebugLogPath: _agentConfig.GatherRuleDebugLogPath, - timelineEvents: timelineEvents)); + timelineEvents: timelineEvents, + stateDirectory: _stateDirectory)); } else if (!string.IsNullOrEmpty(_agentConfig.GatherRuleDebugLogPath)) { diff --git a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/Hosts/GatherRuleExecutorHost.cs b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/Hosts/GatherRuleExecutorHost.cs index 6c25049b..7a87b04d 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/Hosts/GatherRuleExecutorHost.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2.Core/Orchestration/Hosts/GatherRuleExecutorHost.cs @@ -47,7 +47,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration string? imeLogPathOverride, bool unrestrictedMode = false, string? gatherDebugLogPath = null, - TimelineEventStream? timelineEvents = null) + TimelineEventStream? timelineEvents = null, + string? stateDirectory = null) { if (ingress == null) throw new ArgumentNullException(nameof(ingress)); if (clock == null) throw new ArgumentNullException(nameof(clock)); @@ -66,7 +67,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Orchestration var post = new InformationalEventPost(ingress, clock, logger); _executor = new Monitoring.Telemetry.Gather.GatherRuleExecutor( sessionId, tenantId, evt => post.Emit(evt), logger, imeLogPathOverride, - debugLogPath: gatherDebugLogPath); + debugLogPath: gatherDebugLogPath, stateDirectory: stateDirectory); _timelineEvents = timelineEvents; } diff --git a/src/Agent/AutopilotMonitor.Agent.V2/Program.GatherRulesMode.cs b/src/Agent/AutopilotMonitor.Agent.V2/Program.GatherRulesMode.cs index 10e4bfdc..f97f1f68 100644 --- a/src/Agent/AutopilotMonitor.Agent.V2/Program.GatherRulesMode.cs +++ b/src/Agent/AutopilotMonitor.Agent.V2/Program.GatherRulesMode.cs @@ -230,6 +230,9 @@ namespace AutopilotMonitor.Agent.V2 logger.Info($"--run-gather-rules: debug trace -> {gatherDebugLogPath}"); if (consoleMode) Console.WriteLine($"Debug trace: {gatherDebugLogPath}"); + // No stateDirectory on purpose: this mode runs beside the live agent under a + // throwaway session id and must neither resume nor overwrite the session's + // gather state (logparser positions, on_change values). using (var executor = new GatherRuleExecutor(sessionId, tenantId, emitEvent, logger, config.ImeLogPathOverride, debugLogPath: gatherDebugLogPath, debugEcho: consoleMode ? (Action)(line => Console.WriteLine($" {line}")) : null)) diff --git a/src/McpServer/autopilot-monitor-mcp/src/__tests__/auto-exhaust.test.ts b/src/McpServer/autopilot-monitor-mcp/src/__tests__/auto-exhaust.test.ts index ea276728..dc2420c3 100644 --- a/src/McpServer/autopilot-monitor-mcp/src/__tests__/auto-exhaust.test.ts +++ b/src/McpServer/autopilot-monitor-mcp/src/__tests__/auto-exhaust.test.ts @@ -10,9 +10,7 @@ import { describe, it, expect } from 'vitest'; import { scanUntilMatch } from '../client.js'; import { isKnownEventType, - isGatherOutputEventType, - isFilterableEventType, - assertKnownEventType, + withEventTypeNote, assertKnownDevicePropertyKeys, eventTypePrefixOf, ALL_DEVICE_PROPERTY_KEYS, @@ -146,50 +144,59 @@ describe('scanUntilMatch', () => { }); }); -describe('event-type validation', () => { - it('accepts known event types', () => { +describe('event-type filter', () => { + it('keeps isKnownEventType strict: built-in types only', () => { expect(isKnownEventType('app_install_failed')).toBe(true); - expect(isKnownEventType('enrollment_failed')).toBe(true); - expect(() => assertKnownEventType('app_install_failed')).not.toThrow(); - }); - - it('rejects an unknown event type with a helpful, suggestion-bearing error', () => { - expect(isKnownEventType('made_up_type')).toBe(false); - expect(() => assertKnownEventType('app_install_fialed')).toThrow(/event_types catalog/); - // "app_install_fialed" shares the "app" token → at least one suggestion surfaces. - expect(() => assertKnownEventType('app_install_fialed')).toThrow(/Did you mean/); - }); - - it('accepts the hybrid user-affinity types added to the catalog', () => { + expect(isKnownEventType('gather_result')).toBe(true); for (const t of ['desktop_real_user_detected', 'desktop_excluded_user', 'ime_user_token_acquired', 'entra_user_affinity_pending', 'device_registration_event']) { expect(isKnownEventType(t), t).toBe(true); - expect(() => assertKnownEventType(t)).not.toThrow(); } - }); - - it('lets gather-rule output types through the filter gate without cataloguing them', () => { - // Built-in and tenant-authored outputEventType values are rule data, never catalogued. - for (const t of ['gather_dsregcmd_status', 'gather_custom_anything']) { - expect(isGatherOutputEventType(t), t).toBe(true); - expect(isFilterableEventType(t), t).toBe(true); - expect(() => assertKnownEventType(t)).not.toThrow(); - // The strict predicate stays strict: rule-validation uses it as "is a built-in type". + // Gather-rule output types are rule data, never catalogued — rule-validation relies on this. + for (const t of ['gather_dsregcmd_status', 'HPiA-UpdateStatus', 'made_up_type']) { expect(isKnownEventType(t), t).toBe(false); } - // Catalogued code-emitted gather_* lifecycle types are known AND filterable. - expect(isKnownEventType('gather_result')).toBe(true); - expect(isFilterableEventType('gather_result')).toBe(true); }); - it('requires a non-empty suffix after the gather_ prefix', () => { - expect(isGatherOutputEventType('gather_')).toBe(false); - expect(isFilterableEventType('gather_')).toBe(false); - expect(() => assertKnownEventType('gather_')).toThrow(/event_types catalog/); + it('passes a result for a built-in type through unchanged, even when empty', () => { + const empty = { success: true, count: 0, events: [] }; + expect(withEventTypeNote(empty, 'app_install_failed', true)).toBe(empty); }); - it('still rejects a typo that only resembles the gather_ prefix', () => { - expect(isFilterableEventType('gathre_dsregcmd_status')).toBe(false); - expect(() => assertKnownEventType('gathre_dsregcmd_status')).toThrow(/Unknown eventType "gathre_dsregcmd_status" — it is not in the event_types catalog/); + it('passes events of a gather rule\'s own type through unchanged, whatever its spelling', () => { + // Regression anchor: "HPiA-UpdateStatus" was rejected because it is neither catalogued nor gather_-prefixed. + for (const t of ['HPiA-UpdateStatus', 'CustomNetworkStatus', 'gather_dsregcmd_status']) { + const page = { success: true, count: 1, events: [{ eventType: t }] }; + expect(withEventTypeNote(page, t, true), t).toBe(page); + } + }); + + it('marks an exhausted empty result for an uncatalogued type and names close built-in types', () => { + const res = withEventTypeNote({ success: true, count: 0, events: [] }, 'app_install_fialed', true); + + expect(res.eventTypeNote).toMatch(/"app_install_fialed" is not a built-in event type and no event of it was found/); + // "app_install_fialed" shares the "app" token → at least one suggestion surfaces. + expect(res.eventTypeNote).toMatch(/Closest built-in types: .*app_/); + }); + + it('leaves a page alone that is not the final answer yet', () => { + expect(withEventTypeNote({ count: 0, events: [], nextLink: '/x?continuation=t' }, 'custom', true).eventTypeNote).toBeUndefined(); + expect(withEventTypeNote({ count: 0, events: [], moreToScan: true }, 'custom', true).eventTypeNote).toBeUndefined(); + }); + + it('stays silent when the type was not the only filter of a first call', () => { + // Another filter (severity, source, time) or an earlier page of the sweep may explain the + // emptiness — blaming the type would steer the reader towards a typo that is not there. + expect(withEventTypeNote({ count: 0, events: [] }, 'HPiA-UpdateStatus', false).eventTypeNote).toBeUndefined(); + }); + + it('recognises the sessions[] shape of search_sessions_by_event', () => { + expect(withEventTypeNote({ success: true, count: 0, sessions: [] }, 'HPiA-UpdateStatus', true).eventTypeNote) + .toMatch(/not a built-in event type/); + }); + + it('does nothing without an eventType filter', () => { + const empty = { count: 0, events: [] }; + expect(withEventTypeNote(empty, undefined, true)).toBe(empty); }); }); diff --git a/src/McpServer/autopilot-monitor-mcp/src/__tests__/event-type-filter.test.ts b/src/McpServer/autopilot-monitor-mcp/src/__tests__/event-type-filter.test.ts new file mode 100644 index 00000000..9cb27164 --- /dev/null +++ b/src/McpServer/autopilot-monitor-mcp/src/__tests__/event-type-filter.test.ts @@ -0,0 +1,101 @@ +/** + * eventType filters accept a gather rule's own output type. Regression anchor: get_session_events + * rejected "HPiA-UpdateStatus" — a tenant rule's outputEventType — because it is neither catalogued + * nor gather_-prefixed, so the rule's events could not be filtered at all. An empty result for an + * uncatalogued type is marked instead of rejected up front. + * + * Pure unit tests: fetch is stubbed and the registered tool handlers are invoked directly. + */ +import { describe, it, expect, vi, afterEach } from 'vitest'; +import { McpServer } from '@modelcontextprotocol/server'; +import { registerTools } from '../tools.js'; +import { runWithCaller } from '../client.js'; + +type ToolHandler = (args: Record, extra: unknown) => Promise<{ + content?: Array<{ type: string; text?: string }>; + isError?: boolean; +}>; + +const SESSION = 'e259c121-1234-4abc-9def-0123456789ab'; +const GA = { token: 'ga', isGlobalAdmin: true }; +const extra = { signal: new AbortController().signal }; + +function handlerFor(name: string): ToolHandler { + const server = new McpServer({ name: 'test', version: '0.0.0' }); + registerTools(server, undefined, undefined, undefined, true, true, false); + const registry = (server as unknown as { _registeredTools: Record })._registeredTools; + const tool = registry[name]; + if (!tool) throw new Error(`tool ${name} not registered`); + return tool.handler; +} + +function stubFetch(body: Record): { urls: string[] } { + const urls: string[] = []; + vi.stubGlobal('fetch', vi.fn(async (url: string) => { + urls.push(decodeURIComponent(String(url))); + return { ok: true, status: 200, json: async () => body, text: async () => JSON.stringify(body) } as unknown as Response; + })); + return { urls }; +} + +function resultJson(r: { content?: Array<{ text?: string }> }): Record { + return JSON.parse((r.content ?? []).map((c) => c.text ?? '').join('')) as Record; +} + +afterEach(() => vi.unstubAllGlobals()); + +describe('eventType filter with a gather rule\'s own output type', () => { + it('get_session_events filters by the custom type and returns its events unmarked', async () => { + const { urls } = stubFetch({ success: true, count: 1, events: [{ eventType: 'HPiA-UpdateStatus', sequence: 69 }] }); + + const result = await runWithCaller(GA, () => handlerFor('get_session_events')({ sessionId: SESSION, eventType: 'HPiA-UpdateStatus' }, extra)); + + expect(result.isError).toBeFalsy(); + expect(urls[0]).toContain('eventType=HPiA-UpdateStatus'); + const body = resultJson(result); + expect(body.count).toBe(1); + expect(body).not.toHaveProperty('eventTypeNote'); + }); + + it('an empty result for an uncatalogued type carries the note instead of passing as "nothing happened"', async () => { + stubFetch({ success: true, count: 0, events: [] }); + + const result = await runWithCaller(GA, () => handlerFor('get_session_events')({ sessionId: SESSION, eventType: 'app_install_fialed' }, extra)); + + expect(result.isError).toBeFalsy(); + expect(String(resultJson(result).eventTypeNote)).toMatch(/"app_install_fialed" is not a built-in event type.*Closest built-in types/); + }); + + it('no note when severity, not the type, may have emptied the result, or on a continued sweep', async () => { + stubFetch({ success: true, count: 0, events: [] }); + + const filtered = await runWithCaller(GA, () => handlerFor('get_session_events')( + { sessionId: SESSION, eventType: 'HPiA-UpdateStatus', severity: 'Error' }, extra)); + expect(resultJson(filtered)).not.toHaveProperty('eventTypeNote'); + + const continued = await runWithCaller(GA, () => handlerFor('search_sessions_by_event')( + { eventType: 'HPiA-UpdateStatus', continuation: '/api/global/search/sessions-by-event?eventType=HPiA-UpdateStatus&continuation=tok' }, extra)); + expect(resultJson(continued)).not.toHaveProperty('eventTypeNote'); + }); + + it('a built-in type keeps a plain empty result', async () => { + stubFetch({ success: true, count: 0, events: [] }); + + const result = await runWithCaller(GA, () => handlerFor('get_session_events')({ sessionId: SESSION, eventType: 'app_install_failed' }, extra)); + + expect(resultJson(result)).not.toHaveProperty('eventTypeNote'); + }); + + it('search_sessions_by_event and query_raw_events accept the custom type too', async () => { + stubFetch({ success: true, count: 0, sessions: [] }); + const bySession = await runWithCaller(GA, () => handlerFor('search_sessions_by_event')({ eventType: 'CustomNetworkStatus' }, extra)); + expect(bySession.isError).toBeFalsy(); + expect(String(resultJson(bySession).eventTypeNote)).toMatch(/not a built-in event type/); + + const { urls } = stubFetch({ success: true, count: 1, events: [{ EventType: 'CustomNetworkStatus' }] }); + const raw = await runWithCaller(GA, () => handlerFor('query_raw_events')({ eventType: 'CustomNetworkStatus' }, extra)); + expect(raw.isError).toBeFalsy(); + expect(urls[0]).toContain('eventType=CustomNetworkStatus'); + expect(resultJson(raw)).not.toHaveProperty('eventTypeNote'); + }); +}); diff --git a/src/McpServer/autopilot-monitor-mcp/src/client.ts b/src/McpServer/autopilot-monitor-mcp/src/client.ts index 9c4b0880..80c0c1f7 100644 --- a/src/McpServer/autopilot-monitor-mcp/src/client.ts +++ b/src/McpServer/autopilot-monitor-mcp/src/client.ts @@ -680,7 +680,7 @@ const defaultPageFetcher: PageFetcher = (path) => apiFetch): unknown[] | undefined { +export function extractItems(page: Record): unknown[] | undefined { for (const key of ITEM_ARRAY_KEYS) { const v = page[key]; if (Array.isArray(v)) return v; diff --git a/src/McpServer/autopilot-monitor-mcp/src/resource-catalog.ts b/src/McpServer/autopilot-monitor-mcp/src/resource-catalog.ts index e099d73e..36642e7b 100644 --- a/src/McpServer/autopilot-monitor-mcp/src/resource-catalog.ts +++ b/src/McpServer/autopilot-monitor-mcp/src/resource-catalog.ts @@ -4,6 +4,7 @@ * (MCP-protocol resources) and the `get_resource` tool (workaround for * stateless-HTTP MCP clients whose resource discovery is broken). */ +import { extractItems } from './client.js'; import { DIAG_ZIP_MAP } from './diag-zip-map.js'; import { RULE_AUTHORING_GUIDE } from './rule-authoring-guide.js'; import { GATHER_RULE_SCHEMA, ANALYZE_RULE_SCHEMA, RULE_GUARDRAILS } from './rule-authoring.generated.js'; @@ -19,7 +20,7 @@ import { // (all catalog values ∪ INTERNAL_EVENT_TYPES) === the C# const values. When the // agent adds a new event type to Constants.EventTypes, add it to a group here too. // Gather-rule output types ("gather_dsregcmd_status", ...) are rule data, not code-emitted -// types, and stay OUT of the catalog on purpose — see isGatherOutputEventType below. +// types, and stay OUT of the catalog on purpose — see withEventTypeNote below. export const EVENT_TYPES_CATALOG = { phase_events: [ 'phase_transition', @@ -357,38 +358,20 @@ export const DEVICE_PROPERTIES_CATALOG = { // // Without these, an invalid eventType / deviceProperties key produces the SAME // `count: 0` as a genuine miss, so "bad filter" is indistinguishable from "no -// matches". The search tools call these to reject a typo with a clear, correctable -// error instead of a silent empty result. +// matches". The tools reject a deviceProperties typo with a clear error and mark an +// empty result for an uncatalogued eventType with a correctable note. /** Set form of ALL_EVENT_TYPES for O(1) membership checks. */ const EVENT_TYPE_SET = new Set(ALL_EVENT_TYPES); -/** True if `s` is a known event type (public catalog ∪ internal). */ +/** + * True if `s` is a built-in event type (public catalog ∪ internal). Strict on purpose: + * rule-validation relies on it for its outputEventType collision check. + */ export function isKnownEventType(s: string): boolean { return EVENT_TYPE_SET.has(s); } -/** - * Gather-rule output types (`outputEventType`, e.g. "gather_dsregcmd_status") are rule - * DATA — built-in and tenant-authored — not code-emitted types, so they are deliberately - * NOT catalogued: D-072 scopes the catalog to `Constants.EventTypes`, and the C# guard - * test excludes gather outputs for the same reason. They are real events on the wire - * nonetheless, so the filter gate lets any `gather_`-prefixed type (with a non-empty - * suffix) through — the same "a valid prefix passes even when the full key is - * uncatalogued" stance as assertKnownDevicePropertyKeys. Kept apart from - * isKnownEventType, whose strict "is a built-in type" meaning rule-validation relies - * on for its outputEventType collision check. - */ -const GATHER_OUTPUT_PREFIX = 'gather_'; -export function isGatherOutputEventType(s: string): boolean { - return s.startsWith(GATHER_OUTPUT_PREFIX) && s.length > GATHER_OUTPUT_PREFIX.length; -} - -/** True if `s` is accepted as an eventType filter: catalogued, internal, or a gather-rule output. */ -export function isFilterableEventType(s: string): boolean { - return isKnownEventType(s) || isGatherOutputEventType(s); -} - /** * Every catalogued deviceProperties key, flattened across event-type groups * (excluding the `_usage` doc block). This is a CURATED subset — not exhaustive — @@ -415,19 +398,35 @@ function suggestEventTypes(input: string): string[] { } /** - * Throws a descriptive Error if `eventType` is neither a known type nor a - * gather-rule output (see isFilterableEventType). The thrown message routes - * through the tools' `toolError` handler so the model gets an actionable - * correction instead of a misleading empty result. + * eventType filters take ANY type and pass it on as given. Gather-rule output types + * (`outputEventType`, e.g. "gather_dsregcmd_status", "HPiA-UpdateStatus") are rule DATA — built-in + * and tenant-authored — so they are deliberately NOT catalogued (D-072), yet they are real events + * on the wire, and the portal suggests free names like "CustomNetworkStatus". A gate on the catalog + * (or on a naming prefix) locked those out. What the gate protected — a typo must not read as + * "nothing happened" — this note keeps: when the uncatalogued type was the only filter of a first + * call and the answer is an exhausted empty result, it says so and names the closest built-in + * types. With other filters or on a continued sweep the emptiness proves nothing about the type, + * and non-empty or still-continuable pages pass unchanged. */ -export function assertKnownEventType(eventType: string): void { - if (isFilterableEventType(eventType)) return; +export function withEventTypeNote( + data: T, + eventType: string | undefined, + soleFilterOfFirstCall: boolean, +): T & { eventTypeNote?: string } { + const page = data as Record; + if (!eventType || !soleFilterOfFirstCall || isKnownEventType(eventType) || page.nextLink || page.moreToScan) return data; + const items = extractItems(page); + const empty = typeof page.count === 'number' ? page.count === 0 : items !== undefined && items.length === 0; + if (!empty) return data; const suggestions = suggestEventTypes(eventType); - const hint = suggestions.length > 0 ? ` Did you mean: ${suggestions.join(', ')}?` : ''; - throw new Error( - `Unknown eventType "${eventType}" — it is not in the event_types catalog.${hint} ` + - 'Call get_resource(name="event_types") for the full list of valid event types.', - ); + const hint = suggestions.length > 0 ? ` Closest built-in types: ${suggestions.join(', ')}.` : ''; + return { + ...data, + eventTypeNote: + `"${eventType}" is not a built-in event type and no event of it was found here. Gather rules emit their own ` + + `output types (outputEventType) — check the spelling against the rule.${hint} ` + + 'Built-in types: get_resource(name="event_types").', + }; } /** diff --git a/src/McpServer/autopilot-monitor-mcp/src/tools/admin.ts b/src/McpServer/autopilot-monitor-mcp/src/tools/admin.ts index 1e4302fc..d26af8af 100644 --- a/src/McpServer/autopilot-monitor-mcp/src/tools/admin.ts +++ b/src/McpServer/autopilot-monitor-mcp/src/tools/admin.ts @@ -2,7 +2,7 @@ import { McpServer } from '@modelcontextprotocol/server'; import { z } from 'zod'; import { apiFetch, buildQuery, DEFAULT_FIRST_PAGE_SIZE, effectivePageSize, enforceDelegatedTenant, enforceDelegatedTenantForPage, followNextLink, getCallerUpnDomain, getDelegatedTenantIds, getHomeTenantId, pageSizeForCall, pickGlobalOrTenantPath, scanUntilMatch, scanWithTimeoutFallback, jsonBody } from '../client.js'; import { withToolTelemetry } from '../telemetry.js'; -import { getResourceContent, assertKnownEventType, RESOURCE_NAMES } from '../resource-catalog.js'; +import { getResourceContent, withEventTypeNote, RESOURCE_NAMES } from '../resource-catalog.js'; import { READ_ONLY, READ_ONLY_OPEN, MUTATING, MAX_RESULT_SIZE_CHARS, LEAN_RAW_EVENT_FIELDS, LEAN_RAW_EVENT_OMISSION, leanFieldSelection, toolResultText, SessionIdSchema, TenantGuidSchema, tenantIdDescription, pageSizeDescription, CONTINUATION_DESCRIPTION, daysDescription } from './shared.js'; import { toolError } from './error-handler.js'; import { API_BASE_URL } from '../config.js'; @@ -1473,8 +1473,9 @@ export function registerAdminTools(server: McpServer, ga: boolean, strictGa: boo 'PartitionKey/RowKey/Timestamp. Deliberately unenriched: DataJson is the raw stored string (not parsed), ' + 'Severity/Phase are raw ints, no decoded error meanings; for the enriched stream use get_session_events or ' + 'search_events. Use it when search_events does not cover the time range or session scope, or for exact ' + - 'event-type filtering across many sessions. eventType is validated against the event_types catalog (a typo ' + - 'is rejected, not an empty result). With a filter the tool auto-scans past empty pages: "count": 0 without ' + + 'event-type filtering across many sessions. eventType may also be a gather rule\'s own outputEventType; an empty ' + + 'first-page result for a type outside the event_types catalog carries eventTypeNote with the closest built-in ' + + 'types. With a filter the tool auto-scans past empty pages: "count": 0 without ' + 'nextLink means no matches; "moreToScan": true means the per-call scan budget was hit — continue with nextLink. ' + 'pageSize is the index-scan cadence (one indexed session can contribute several events, so a page may hold ' + 'more than pageSize). A page that ends early carries "partial": true — nothing is missing up to its nextLink. ' + @@ -1513,7 +1514,6 @@ export function registerAdminTools(server: McpServer, ga: boolean, strictGa: boo // Default projection follows intent — see leanFieldSelection (DataJson is the raw payload column). const { fields, leanDefaultApplied } = leanFieldSelection(explicitFields, continuation, targeted, LEAN_RAW_EVENT_FIELDS); const tenantId = enforceDelegatedTenantForPage(rawTenantId, continuation); - if (eventType) assertKnownEventType(eventType); const basePath = pickGlobalOrTenantPath('/api/global/raw/events', '/api/raw/events', tenantId); const path = followNextLink( basePath, @@ -1526,7 +1526,11 @@ export function registerAdminTools(server: McpServer, ga: boolean, strictGa: boo // forward so the model isn't misled by an empty-but-continuable page. The server // bounds each page by its own scan budget; should a call still time out, retry once // with a halved pageSize on the same cursor instead of failing identically twice. - const data = await scanWithTimeoutFallback(path, basePath, effectivePageSize(pageSize, continuation)); + const data = withEventTypeNote( + await scanWithTimeoutFallback(path, basePath, effectivePageSize(pageSize, continuation)), + eventType, + !continuation && !severity && !source && !startedAfter && !startedBefore, + ); // Announce the omission in-band so a reader of the result knows DataJson exists and how to get it. return toolResultText(leanDefaultApplied ? { ...data, ...LEAN_RAW_EVENT_OMISSION } : data, MAX_RESULT_SIZE_CHARS.events); } catch (error: unknown) { diff --git a/src/McpServer/autopilot-monitor-mcp/src/tools/sessions.ts b/src/McpServer/autopilot-monitor-mcp/src/tools/sessions.ts index 91fb39a4..7dd12797 100644 --- a/src/McpServer/autopilot-monitor-mcp/src/tools/sessions.ts +++ b/src/McpServer/autopilot-monitor-mcp/src/tools/sessions.ts @@ -5,7 +5,7 @@ import { withToolTelemetry } from '../telemetry.js'; import { READ_ONLY, MAX_RESULT_SIZE_CHARS, LEAN_EVENT_FIELDS, LEAN_EVENT_OMISSION, leanFieldSelection, SUMMARY_EVENT_FIELDS, toolResultText, SessionIdSchema, isBenignHealthDetectionReport, tenantIdDescription, pageSizeDescription, CONTINUATION_DESCRIPTION, daysDescription } from './shared.js'; import { toolError } from './error-handler.js'; import { lookupErrorCode } from '../error-code-catalog.js'; -import { assertKnownEventType, assertKnownDevicePropertyKeys } from '../resource-catalog.js'; +import { withEventTypeNote, assertKnownDevicePropertyKeys } from '../resource-catalog.js'; import { interpolateAnalysisResults } from '../interpolate-rule-template.js'; import { buildSessionCoverage } from '../session-coverage.js'; import { API_BASE_URL } from '../config.js'; @@ -220,10 +220,10 @@ export function registerSessionTools(server: McpServer, ga: boolean, delegated: title: 'Search Sessions by Event', description: 'Find sessions that contain a given event type, read from the event-type index (e.g. every session with ' + - 'app_install_failed or enrollment_failed). eventType is validated against the event_types catalog; an unknown ' + - 'type is rejected, not an empty result.', + 'app_install_failed or enrollment_failed). eventType may also be a gather rule\'s own outputEventType; an empty ' + + 'first-page result for a type outside the event_types catalog carries eventTypeNote with the closest built-in types.', inputSchema: { - eventType: z.string().describe('Event type string from get_resource(name="event_types") (e.g. "app_install_failed", "enrollment_failed")'), + eventType: z.string().describe('Event type from get_resource(name="event_types") (e.g. "app_install_failed", "enrollment_failed"), or a gather rule\'s own outputEventType'), tenantId: z.string().optional().describe(tenantIdDescription(ga, delegated)), pageSize: z.coerce.number().int().min(1).max(1000).optional() .describe(pageSizeDescription(DEFAULT_FIRST_PAGE_SIZE, 1000, 'Raise it for full sweeps.')), @@ -237,9 +237,8 @@ export function registerSessionTools(server: McpServer, ga: boolean, delegated: const pageSize = pageSizeForCall(args.pageSize, continuation, DEFAULT_FIRST_PAGE_SIZE); const tenantId = enforceDelegatedTenantForPage(rawTenantId, continuation); // eventType is the sole filter and is applied server-side (EventTypeIndex OData), - // so an empty page never carries a nextLink — no auto-exhaust needed. But validate - // the type so a typo is a clear error rather than a silent empty result. - assertKnownEventType(eventType); + // so an empty page never carries a nextLink — no auto-exhaust needed. An empty result + // for an uncatalogued type says so, so a typo does not read as "no such sessions". const basePath = pickGlobalOrTenantPath('/api/global/search/sessions-by-event', '/api/search/sessions-by-event', tenantId); const path = followNextLink( basePath, @@ -248,7 +247,7 @@ export function registerSessionTools(server: McpServer, ga: boolean, delegated: { pageSize }, ); const data = await apiFetch(path); - return toolResultText(data, MAX_RESULT_SIZE_CHARS.indexSessions); + return toolResultText(withEventTypeNote(data, eventType, !continuation), MAX_RESULT_SIZE_CHARS.indexSessions); } catch (error: unknown) { return toolError('search_sessions_by_event', args, error); } @@ -393,8 +392,8 @@ export function registerSessionTools(server: McpServer, ga: boolean, delegated: title: 'Get Session Events', description: 'RAW EVENT RETRIEVAL (fallback when ranked search_events misses, or when every event of one session is needed ' + - 'in chronological sequence). Filter by eventType (validated against the event_types catalog; an unknown type is ' + - 'rejected, not an empty result), severity or source (app name). An unfiltered read omits the per-event "data" ' + + 'in chronological sequence). Filter by eventType (built-in, or a gather rule\'s own outputEventType; an empty first-page ' + + 'result for a type outside the event_types catalog carries eventTypeNote), severity or source (app name). An unfiltered read omits the per-event "data" ' + 'payload by default and says so in omittedFields; a filtered read includes it (see fields). ' + 'Filters are applied after the partition read, so the tool scans forward past empty pages: "count": 0 without ' + 'nextLink means no matching events; "moreToScan": true means the per-call scan budget was hit — pass the ' + @@ -402,7 +401,7 @@ export function registerSessionTools(server: McpServer, ga: boolean, delegated: inputSchema: { sessionId: SessionIdSchema.describe('Session UUID'), tenantId: z.string().optional().describe(sessionTenantIdDescription), - eventType: z.string().optional().describe('Only events of this type; values from get_resource(name="event_types")'), + eventType: z.string().optional().describe('Only events of this type: a value from get_resource(name="event_types"), or a gather rule\'s own outputEventType'), severity: z.enum(EVENT_SEVERITIES).optional(), source: z.string().optional().describe('Filter by event source/app name (e.g. "MicrosoftTeams")'), fields: z.string().optional() @@ -420,7 +419,6 @@ export function registerSessionTools(server: McpServer, ga: boolean, delegated: // Default projection follows intent — see leanFieldSelection. const { fields, leanDefaultApplied } = leanFieldSelection(explicitFields, continuation, Boolean(eventType || severity || source), LEAN_EVENT_FIELDS); const tenantId = enforceDelegatedTenantForPage(rawTenantId, continuation); - if (eventType) assertKnownEventType(eventType); const basePath = `/api/sessions/${sessionId}/events`; const path = followNextLink( basePath, @@ -432,7 +430,11 @@ export function registerSessionTools(server: McpServer, ga: boolean, delegated: // event partition, so a page can be empty while matches sit on a later page. // Auto-exhaust forward so the model isn't misled by an empty-but-continuable page; // a timeout is retried once with a halved pageSize on the same cursor. - const data = await scanWithTimeoutFallback(path, basePath, effectivePageSize(pageSize, continuation)); + const data = withEventTypeNote( + await scanWithTimeoutFallback(path, basePath, effectivePageSize(pageSize, continuation)), + eventType, + !continuation && !severity && !source, + ); // Announce the omission in-band so a reader of the result knows the payload exists and how to get it. return toolResultText(leanDefaultApplied ? { ...data, ...LEAN_EVENT_OMISSION } : data, MAX_RESULT_SIZE_CHARS.events); } catch (error: unknown) {