fix(agent): the IME log tracker verifies and rewinds when a concurrent IME process overwrote bytes behind its bookmark; executor lines belong to the invocation at their file position (D-246)

IME's trace listener opens every log once per process with FileMode.Append and FileShare.ReadWrite
and writes at its own stream position afterwards. AgentExecutor.exe (two processes per user-context
script, one per detection script) has such a listener on AgentExecutor.log and on
IntuneManagementExtension.log, so a process that resumes logging after another one appended
overwrites those bytes: the final file holds the last writer, a byte-position tailer has read the
first version. Session 46749560 lost 6 of 16 platform-script results and 4 of 16 executor end
blocks with every tracker health counter at zero — results fell back to the exit code (one Success
where IME said Failed), one finished script stayed Incomplete. The agentexecutor_fallback results
in the fleet are this race, not late IME logging.

The tracker now keeps a fingerprint ledger of every entry it processed from the two multi-writer
files (FNV-1a 64 over the decoded bytes, one CmTraceEntryAssembler shared by the read loop and the
verification so entry boundaries are identical by construction), verifies it on every script end
signal, on a byte-budgeted cadence while an invocation is in flight (256 KB/s, at least 1 s
apart), in a 4 KB guard zone on every growth pass and on a fragment at the bookmark, and rewinds
to the first changed entry; entries whose (offset, hash) still read unchanged are skipped,
everything else is processed as backlog (never fresh, never a calibration anchor). The
policyId-less executor lines are owned by the nearest preceding invocation marker in their file
(PS-AGENT-ARG opens, PS-AGENT-SCRIPT-START upgrades to the platform script, PS-AGENT-INVOCATION and
PS-AGENT-COMPLETED close) instead of a last-started pointer, so an end block a rewind uncovers
reaches its own script. Restart seeds the ledger from the 64 KB behind the persisted bookmark;
a rollover clears it. Four restart-safe counters (ime_overwrite_rewinds,
ime_overwrite_bytes_reprocessed, ime_verify_passes, ime_verified_bytes) ride agent_metrics_snapshot
and ime_pattern_hits — recovered data, not skipped work, no new event type.

Tests: ImeLogTrackerOverwriteRewindTests (15) drive the positional-overwrite write pattern against
both files, including the two field scenarios with the shipped patterns; agent suite 2921 green,
backend known-actions list extended for the two new patterns.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
2026-09-10 23:40:20 +02:00
parent 18d4f08f2b
commit 197de1d718
16 changed files with 1570 additions and 297 deletions
+16
View File
@@ -646,6 +646,22 @@
"description": "Detects WinGet app installation progress with Operation Phase = Installing (format variant 2).",
"enabled": true
},
{
"patternId": "PS-AGENT-ARG",
"category": "always",
"pattern": "^Adding argument (?<kind>[A-Za-z0-9]+) with value ",
"action": "agentInvocationArgument",
"description": "The first argument line of every AgentExecutor.exe invocation (powershell, remediationScript, detectionScript, requirementScript, ...). Opens the invocation that owns the policyId-less exit/output lines following it in AgentExecutor.log; PS-AGENT-SCRIPT-START upgrades a powershell invocation to its platform script. Also keeps the tracker's overwrite check active while the invocation runs.",
"enabled": true
},
{
"patternId": "PS-AGENT-COMPLETED",
"category": "always",
"pattern": "^Agent executor completed\\.",
"action": "agentExecutorCompleted",
"description": "Last line of an AgentExecutor.exe invocation. Closes the invocation's ownership of the exit/output lines in AgentExecutor.log and triggers the tracker's overwrite check: the executor wrote its end block at its own stream position, which may lie behind the tracker's bookmark when another process appended meanwhile.",
"enabled": true
},
{
"patternId": "PS-AGENT-ERROR",
"category": "always",
+9
View File
@@ -0,0 +1,9 @@
{
"$schema": "../schema/ime-log-pattern.schema.json",
"patternId": "PS-AGENT-ARG",
"category": "always",
"pattern": "^Adding argument (?<kind>[A-Za-z0-9]+) with value ",
"action": "agentInvocationArgument",
"description": "The first argument line of every AgentExecutor.exe invocation (powershell, remediationScript, detectionScript, requirementScript, ...). Opens the invocation that owns the policyId-less exit/output lines following it in AgentExecutor.log; PS-AGENT-SCRIPT-START upgrades a powershell invocation to its platform script. Also keeps the tracker's overwrite check active while the invocation runs.",
"enabled": true
}
@@ -0,0 +1,9 @@
{
"$schema": "../schema/ime-log-pattern.schema.json",
"patternId": "PS-AGENT-COMPLETED",
"category": "always",
"pattern": "^Agent executor completed\\.",
"action": "agentExecutorCompleted",
"description": "Last line of an AgentExecutor.exe invocation. Closes the invocation's ownership of the exit/output lines in AgentExecutor.log and triggers the tracker's overwrite check: the executor wrote its end block at its own stream position, which may lie behind the tracker's bookmark when another process appended meanwhile.",
"enabled": true
}
+2
View File
@@ -71,6 +71,8 @@
"scriptOutput",
"scriptCompleted",
"resetPlatformScriptContext",
"agentInvocationArgument",
"agentExecutorCompleted",
"healthScriptResult",
"healthScriptDetectionResult",
"captureExitCode",
@@ -0,0 +1,527 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Text.RegularExpressions;
using System.Threading;
using System.Threading.Tasks;
using AutopilotMonitor.Agent.V2.Core.Logging;
using AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime;
using AutopilotMonitor.Agent.V2.Core.Tests.Harness;
using AutopilotMonitor.Shared.Models;
using Newtonsoft.Json;
using Xunit;
namespace AutopilotMonitor.Agent.V2.Core.Tests.Monitoring.Ime
{
/// <summary>
/// Session 46749560: IME's trace listener writes every log through one FileMode.Append
/// stream per process, so a process that resumes logging after another one appended
/// overwrites those bytes at its own stale position. The tracker had read the first version
/// and never saw the second — 6 of 16 script results and 4 of 16 executor end blocks were
/// lost while the final files looked intact. These tests drive that exact write pattern
/// (positional overwrite behind the bookmark) against the two multi-writer logs and pin the
/// contract: unchanged bytes never rewind, changed bytes are processed exactly once, and the
/// executor lines a rewind uncovers belong to the invocation that owns their file position.
/// </summary>
public sealed class ImeLogTrackerOverwriteRewindTests
{
private static readonly DateTime T0 = new DateTime(2026, 9, 10, 19, 30, 0, DateTimeKind.Utc);
private static List<ImeLogPattern> MarkerPatterns() => new List<ImeLogPattern>
{
new ImeLogPattern
{
PatternId = "T-MARK", Category = "always", Enabled = true,
Pattern = @"^marker (?<n>\d+)", Action = "noop",
Parameters = new Dictionary<string, string>(),
},
};
private static readonly string[] ScriptPatternIds =
{
"PS-AGENT-INVOCATION", "PS-AGENT-ARG", "PS-AGENT-SCRIPT-START", "PS-AGENT-EXITCODE",
"PS-AGENT-OUTPUT", "PS-AGENT-COMPLETED", "PS-SCRIPT-GENERATED", "PS-SCRIPT-CONTEXT", "PS-SCRIPT-RESULT",
};
/// <summary>The shipped pattern JSON — the same source combine.js embeds — so the contract is the real one.</summary>
private static List<ImeLogPattern> ScriptPatterns()
{
DirectoryInfo? dir = new DirectoryInfo(AppContext.BaseDirectory);
string? patternDir = null;
while (dir != null)
{
var candidate = Path.Combine(dir.FullName, "rules", "ime-log-patterns");
if (Directory.Exists(candidate)) { patternDir = candidate; break; }
dir = dir.Parent;
}
Assert.NotNull(patternDir);
var byId = new Dictionary<string, ImeLogPattern>(StringComparer.OrdinalIgnoreCase);
foreach (var file in Directory.GetFiles(patternDir, "*.json"))
{
var pattern = JsonConvert.DeserializeObject<ImeLogPattern>(File.ReadAllText(file));
if (pattern?.PatternId != null) byId[pattern.PatternId] = pattern;
}
return ScriptPatternIds.Select(id =>
{
Assert.True(byId.TryGetValue(id, out var p), $"Shipped IME pattern '{id}' not found under {patternDir}.");
return p!;
}).ToList();
}
private static string Entry(string message, string thread = "1")
=> $"<![LOG[{message}]LOG]!><time=\"12:00:00.0000000\" date=\"9-10-2026\" component=\"X\" context=\"\" type=\"1\" thread=\"{thread}\" file=\"\">\r\n";
private const string PlatformId = "d39eeef3-e88e-41a0-82d7-5e28ee7b5acf";
private const string PlatformStartLine =
@"Adding argument powershell with value C:\Program Files (x86)\Microsoft Intune Management Extension\Policies\Scripts\73d664e4-0886-4a73-b745-c694da45ddb4_d39eeef3-e88e-41a0-82d7-5e28ee7b5acf.ps1 to the named argument list.";
private const string DetectionStartLine =
@"Adding argument detectionScript with value C:\Windows\IMECache\DetectionScripts\App_cc46aa62-e93a-41a1-bdd0-44f57e92a66d.ps1 to the named argument list.";
private const string GeneratedLine =
@"Script file C:\Program Files (x86)\Microsoft Intune Management Extension\Policies\Scripts\73d664e4-0886-4a73-b745-c694da45ddb4_d39eeef3-e88e-41a0-82d7-5e28ee7b5acf.ps1 is generated.";
private const string ResultLine =
"[PowerShell] User Id = 73d664e4-0886-4a73-b745-c694da45ddb4, Policy id = " + PlatformId + ", policy result = Success";
private sealed class Harness : IDisposable
{
private readonly TempDirectory _tmp = new TempDirectory();
private readonly string _matchLog;
private readonly List<ImeLogPattern> _patterns;
public ImeLogTracker Tracker { get; private set; }
public DateTime Now { get; set; } = T0;
public string LogName { get; }
public string LogPath => Path.Combine(_tmp.Path, LogName);
public string StateDir { get; }
public List<ScriptExecutionState> Completed { get; } = new List<ScriptExecutionState>();
public Harness(string logName, List<ImeLogPattern> patterns)
{
LogName = logName;
_patterns = patterns;
var logDir = Path.Combine(_tmp.Path, "agent");
Directory.CreateDirectory(logDir);
_matchLog = Path.Combine(logDir, "ime-pattern-matches.log");
StateDir = Path.Combine(_tmp.Path, "state");
Tracker = Build();
}
private ImeLogTracker Build()
{
var t = new ImeLogTracker(_tmp.Path, _patterns, new AgentLogger(Path.Combine(_tmp.Path, "agent"), AgentLogLevel.Debug),
matchLogPath: _matchLog, stateDirectory: StateDir);
t.UtcNowProvider = () => Now;
t.OnScriptCompleted = s => Completed.Add(s);
return t;
}
/// <summary>Agent restart: dispose, rebuild over the same state directory, restore the bookmark.</summary>
public void Restart()
{
Tracker.SaveStateForTest();
Tracker.Dispose();
Tracker = Build();
Tracker.LoadStateForTest();
}
public void Append(string text) => File.AppendAllText(LogPath, text, new UTF8Encoding(false));
/// <summary>What IME's stale writer does: write at its own position, over whatever is there.</summary>
public void Overwrite(long offset, string text)
{
using (var fs = new FileStream(LogPath, FileMode.Open, FileAccess.Write, FileShare.ReadWrite))
{
fs.Seek(offset, SeekOrigin.Begin);
var bytes = new UTF8Encoding(false).GetBytes(text);
fs.Write(bytes, 0, bytes.Length);
}
}
public long Length => new FileInfo(LogPath).Length;
public Task Pass(CancellationToken token = default) => Tracker.CheckLogFilesAsync(token);
/// <summary>Marker numbers matched so far, in match order.</summary>
public List<int> Matched()
{
if (!File.Exists(_matchLog)) return new List<int>();
return File.ReadAllLines(_matchLog)
.Select(l => Regex.Match(l, @"\[T-MARK\] <!\[LOG\[marker (\d+)"))
.Where(m => m.Success).Select(m => int.Parse(m.Groups[1].Value)).ToList();
}
public ImeTrackerHealth Health => Tracker.GetHealthSnapshot();
public void Dispose()
{
Tracker.Dispose();
_tmp.Dispose();
}
}
// -----------------------------------------------------------------------
// Marker-level contract on AgentExecutor.log
// -----------------------------------------------------------------------
[Fact]
public async Task Unchanged_bytes_never_rewind_however_often_they_are_verified()
{
using var h = new Harness("AgentExecutor.log", MarkerPatterns());
h.Append(Entry("marker 1") + "<![LOG[marker 2\nsecond line\n]LOG]!><time=\"12:00:00.0000000\" date=\"9-10-2026\" component=\"X\" context=\"\" type=\"1\" thread=\"1\" file=\"\">\r\n" + Entry("marker 3"));
await h.Pass();
Assert.Equal(new[] { 1, 2, 3 }, h.Matched());
for (var i = 0; i < 5; i++)
{
h.Tracker.RequestOverwriteCheckForTest();
h.Now = h.Now.AddSeconds(2);
await h.Pass();
}
Assert.Equal(new[] { 1, 2, 3 }, h.Matched());
Assert.Equal(0, h.Health.OverwriteRewinds);
Assert.Equal(5, h.Health.VerifyPasses);
Assert.Equal(3, h.Tracker.LedgerEntryCountForTest("AgentExecutor.log"));
}
[Fact]
public async Task Non_multi_writer_files_keep_no_ledger_and_are_never_verified()
{
using var h = new Harness("AppWorkload.log", MarkerPatterns());
h.Append(Entry("marker 1") + Entry("marker 2"));
await h.Pass();
h.Tracker.RequestOverwriteCheckForTest();
h.Now = h.Now.AddSeconds(2);
await h.Pass();
Assert.Equal(new[] { 1, 2 }, h.Matched());
Assert.Equal(0, h.Tracker.LedgerEntryCountForTest("AppWorkload.log"));
Assert.Equal(0, h.Health.VerifyPasses);
Assert.Equal(0, h.Health.VerifiedBytes);
}
[Fact]
public async Task Block_written_over_read_bytes_is_processed_once_and_the_surviving_tail_is_not_repeated()
{
using var h = new Harness("AgentExecutor.log", MarkerPatterns());
var one = Entry("marker 1");
h.Append(one + Entry("marker 2") + Entry("marker 3"));
await h.Pass();
Assert.Equal(new[] { 1, 2, 3 }, h.Matched());
// A stale writer lays two entries over the first two (same length: marker 3 survives
// intact at its old offset). No growth, no fragment — only the end signal finds it.
h.Overwrite(0, Entry("marker 7") + Entry("marker 8"));
Assert.Equal(3 * one.Length, h.Length);
h.Tracker.RequestOverwriteCheckForTest();
h.Now = h.Now.AddMilliseconds(100);
await h.Pass();
Assert.Equal(new[] { 1, 2, 3, 7, 8 }, h.Matched());
Assert.Equal(1, h.Health.OverwriteRewinds);
Assert.Equal(3 * one.Length, h.Health.OverwriteBytesReprocessed);
Assert.Equal(3, h.Tracker.LedgerEntryCountForTest("AgentExecutor.log"));
// Nothing left to find.
h.Tracker.RequestOverwriteCheckForTest();
h.Now = h.Now.AddSeconds(2);
await h.Pass();
Assert.Equal(new[] { 1, 2, 3, 7, 8 }, h.Matched());
Assert.Equal(1, h.Health.OverwriteRewinds);
}
[Fact]
public async Task Block_that_cuts_an_old_entry_leaves_a_fragment_that_is_never_matched()
{
using var h = new Harness("AgentExecutor.log", MarkerPatterns());
h.Append(Entry("marker 1") + Entry("marker 2") + Entry("marker 3"));
await h.Pass();
// Longer than two old entries, shorter than three: the tail of marker 3's line survives
// as a fragment behind the new block.
h.Overwrite(0, Entry("marker 7") + Entry("marker 8 is a little longer"));
h.Tracker.RequestOverwriteCheckForTest();
h.Now = h.Now.AddMilliseconds(100);
await h.Pass();
Assert.Equal(new[] { 1, 2, 3, 7, 8 }, h.Matched());
Assert.Equal(1, h.Health.OverwriteRewinds);
}
[Fact]
public async Task Guard_zone_catches_a_short_block_when_the_file_grows_without_any_signal()
{
using var h = new Harness("AgentExecutor.log", MarkerPatterns());
h.Append(Entry("marker 1") + Entry("marker 2") + Entry("marker 3"));
await h.Pass();
// Same-length replacement of marker 1 (no fragment), then an ordinary append by
// another writer: the growth pass re-checks the last 4 KB and finds the change.
h.Overwrite(0, Entry("marker 7"));
h.Append(Entry("marker 4"));
h.Now = h.Now.AddMilliseconds(100);
await h.Pass();
Assert.Equal(new[] { 1, 2, 3, 7, 4 }, h.Matched());
Assert.Equal(1, h.Health.OverwriteRewinds);
Assert.Equal(1, h.Health.VerifyPasses); // the full check that follows a guard hit
Assert.True(h.Health.VerifiedBytes > 0);
}
[Fact]
public async Task Guard_hit_rewinds_to_the_true_start_of_the_block_not_the_guard_zone()
{
using var h = new Harness("AgentExecutor.log", MarkerPatterns());
// ~6 KB of entries so the guard zone covers only the last few.
var sb = new StringBuilder();
for (var i = 1; i <= 60; i++) sb.Append(Entry($"marker {i}"));
h.Append(sb.ToString());
await h.Pass();
Assert.Equal(60, h.Matched().Count);
// A block over everything plus one more entry: the guard sees a change at its
// zone start, the full check must place the rewind at offset 0.
var block = new StringBuilder();
for (var i = 101; i <= 161; i++) block.Append(Entry($"marker {i}"));
h.Overwrite(0, block.ToString());
h.Now = h.Now.AddMilliseconds(100);
await h.Pass();
var matched = h.Matched();
Assert.Equal(60 + 61, matched.Count);
Assert.Equal(Enumerable.Range(101, 61), matched.Skip(60));
Assert.Equal(1, h.Health.OverwriteRewinds);
}
[Fact]
public async Task Fragment_at_the_bookmark_triggers_the_check_when_no_ledger_entry_starts_in_the_guard_zone()
{
using var h = new Harness("AgentExecutor.log", MarkerPatterns());
var one = Entry("marker 1");
var big = Entry("marker 2 " + new string('x', 6000));
h.Append(one + big);
await h.Pass();
Assert.Equal(new[] { 1, 2 }, h.Matched());
// The 6 KB entry is the last ledger entry and starts before the 4 KB guard zone, so
// the guard has nothing to check. The stale writer's block is longer than that entry:
// the pass after it starts mid-line — a fragment — and must not match it raw.
h.Overwrite(one.Length, Entry("marker 7 " + new string('y', 3000)) + Entry("marker 8 " + new string('z', 3100)));
h.Now = h.Now.AddMilliseconds(100);
await h.Pass();
Assert.Equal(new[] { 1, 2 }, h.Matched());
h.Now = h.Now.AddMilliseconds(100);
await h.Pass();
Assert.Equal(new[] { 1, 2, 7, 8 }, h.Matched());
Assert.Equal(1, h.Health.OverwriteRewinds);
}
[Fact]
public async Task Hidden_block_is_found_by_the_cadence_while_a_script_is_in_flight_and_the_cadence_is_throttled()
{
using var h = new Harness("AgentExecutor.log", MarkerPatterns());
h.Append(Entry("marker 1") + Entry("marker 2") + Entry("marker 3"));
await h.Pass();
// A platform script is pending → contention. First pass in contention verifies at once.
h.Tracker.SeedPendingPlatformScriptForTesting("policyX", exitCode: null, exitObservedAtUtc: null, startedAtUtc: T0);
h.Now = T0.AddMilliseconds(100);
await h.Pass();
Assert.Equal(1, h.Health.VerifyPasses);
// Same-length replacement, no growth, no signal: only the cadence can find it —
// and not before a second has passed since the last verification.
h.Overwrite(0, Entry("marker 7"));
h.Now = T0.AddMilliseconds(500);
await h.Pass();
Assert.Equal(new[] { 1, 2, 3 }, h.Matched());
Assert.Equal(1, h.Health.VerifyPasses);
h.Now = T0.AddMilliseconds(1200);
await h.Pass();
Assert.Equal(new[] { 1, 2, 3, 7 }, h.Matched());
Assert.Equal(2, h.Health.VerifyPasses);
Assert.Equal(1, h.Health.OverwriteRewinds);
}
[Fact]
public async Task Without_contention_or_signal_nothing_is_verified()
{
using var h = new Harness("AgentExecutor.log", MarkerPatterns());
h.Append(Entry("marker 1") + Entry("marker 2"));
await h.Pass();
for (var i = 0; i < 20; i++)
{
h.Now = h.Now.AddSeconds(1);
await h.Pass();
}
Assert.Equal(0, h.Health.VerifyPasses);
Assert.Equal(0, h.Health.VerifiedBytes);
}
[Fact]
public async Task Restart_seeds_the_ledger_so_an_overwrite_after_the_restart_is_still_found()
{
using var h = new Harness("AgentExecutor.log", MarkerPatterns());
h.Append(Entry("marker 1") + Entry("marker 2") + Entry("marker 3"));
await h.Pass();
h.Restart();
Assert.Equal(0, h.Tracker.LedgerEntryCountForTest("AgentExecutor.log"));
// First pass after the restart: nothing new to read, but the ledger is seeded from
// the bytes behind the persisted bookmark.
h.Now = h.Now.AddMilliseconds(100);
await h.Pass();
Assert.Equal(3, h.Tracker.LedgerEntryCountForTest("AgentExecutor.log"));
h.Overwrite(0, Entry("marker 7"));
h.Tracker.RequestOverwriteCheckForTest();
h.Now = h.Now.AddMilliseconds(100);
await h.Pass();
Assert.Equal(new[] { 1, 2, 3, 7 }, h.Matched());
Assert.Equal(1, h.Health.OverwriteRewinds);
// The counter is part of the persisted health.
h.Restart();
Assert.Equal(1, h.Health.OverwriteRewinds);
}
[Fact]
public async Task Rollover_clears_the_ledger_and_the_new_file_is_read_without_a_false_rewind()
{
using var h = new Harness("AgentExecutor.log", MarkerPatterns());
h.Append(Entry("marker 1") + Entry("marker 2") + Entry("marker 3"));
await h.Pass();
File.WriteAllText(h.LogPath, Entry("marker 7") + Entry("marker 8"), new UTF8Encoding(false));
h.Now = h.Now.AddMilliseconds(100);
await h.Pass();
Assert.Equal(new[] { 1, 2, 3, 7, 8 }, h.Matched());
h.Tracker.RequestOverwriteCheckForTest();
h.Now = h.Now.AddSeconds(2);
await h.Pass();
Assert.Equal(new[] { 1, 2, 3, 7, 8 }, h.Matched());
Assert.Equal(0, h.Health.OverwriteRewinds);
Assert.Equal(2, h.Tracker.LedgerEntryCountForTest("AgentExecutor.log"));
}
[Fact]
public async Task Cancelled_pass_does_not_lose_a_pending_check()
{
using var h = new Harness("AgentExecutor.log", MarkerPatterns());
h.Append(Entry("marker 1") + Entry("marker 2") + Entry("marker 3"));
await h.Pass();
h.Overwrite(0, Entry("marker 7") + Entry("marker 8"));
h.Tracker.RequestOverwriteCheckForTest();
h.Now = h.Now.AddMilliseconds(100);
using (var cts = new CancellationTokenSource())
{
cts.Cancel();
await h.Pass(cts.Token);
}
// The cancelled pass touched nothing after the rewind; the next one reads it all.
h.Now = h.Now.AddMilliseconds(100);
await h.Pass();
Assert.Equal(new[] { 1, 2, 3, 7, 8 }, h.Matched());
}
// -----------------------------------------------------------------------
// The field scenario with the shipped patterns
// -----------------------------------------------------------------------
[Fact]
public async Task Executor_end_block_written_over_a_detection_invocation_belongs_to_its_own_script()
{
using var h = new Harness("AgentExecutor.log", ScriptPatterns());
// Executor E starts its platform script and waits for PowerShell.
h.Append(Entry("ExecutorLog AgentExecutor gets invoked") + Entry(PlatformStartLine) + Entry("[Executor] created powershell with process id 12708"));
await h.Pass();
var executorPosition = h.Length;
// Detection executor D appends its whole run while E waits — the tracker reads it.
var detectionBlock = Entry("ExecutorLog AgentExecutor gets invoked") + Entry(DetectionStartLine)
+ Entry("Powershell exit code is 3") + Entry("write output done. output = detected nothing, error = ")
+ Entry("Agent executor completed.") + Entry(@"Writing result: False to resultFilePath: C:\x\App_cc46aa62.txt");
h.Append(detectionBlock);
h.Now = h.Now.AddMilliseconds(100);
await h.Pass();
// E ends: its end block lands at its own stale position, over D's lines.
var endBlock = Entry("Powershell exit code is 0") + Entry("write output done. output = Hello from E, error = ") + Entry("Agent executor completed.");
Assert.True(endBlock.Length < detectionBlock.Length);
h.Overwrite(executorPosition, endBlock);
// Nothing visible signalled the end; the cadence (E is pending) finds the block.
h.Now = h.Now.AddSeconds(1.5);
await h.Pass();
Assert.Equal(1, h.Health.OverwriteRewinds);
// IME's result line (in the other file) completes the script with E's own data.
h.Tracker.ProcessLogMessageForTest(ResultLine);
var script = Assert.Single(h.Completed);
Assert.Equal(PlatformId, script.PolicyId);
Assert.Equal("Success", script.Result);
Assert.Equal("ime_policy_result", script.ResultSource);
Assert.Equal(0, script.ExitCode);
Assert.Equal("Hello from E", script.Stdout);
}
[Fact]
public async Task Detection_exit_lines_read_before_the_overwrite_never_reach_the_platform_script()
{
using var h = new Harness("AgentExecutor.log", ScriptPatterns());
h.Append(Entry("ExecutorLog AgentExecutor gets invoked") + Entry(PlatformStartLine) + Entry("[Executor] created powershell with process id 1"));
await h.Pass();
h.Append(Entry("ExecutorLog AgentExecutor gets invoked") + Entry(DetectionStartLine)
+ Entry("Powershell exit code is 3") + Entry("write output done. output = detected nothing, error = ") + Entry("Agent executor completed."));
h.Now = h.Now.AddMilliseconds(100);
await h.Pass();
// The platform result arrives without any end block ever becoming visible.
h.Tracker.ProcessLogMessageForTest(ResultLine);
var script = Assert.Single(h.Completed);
Assert.Equal("Success", script.Result);
Assert.Null(script.ExitCode);
Assert.Null(script.Stdout);
}
[Fact]
public async Task Ime_result_line_hidden_by_an_executor_start_line_is_recovered_from_the_ime_log()
{
using var h = new Harness("IntuneManagementExtension.log", ScriptPatterns());
h.Append(Entry(GeneratedLine, "17") + Entry("Launch powershell executor in user session", "17") + Entry("process id = 10608", "17"));
await h.Pass();
var servicePosition = h.Length;
// The executor process appends its telemetry start line at the true end of file.
h.Append(Entry("[Telemetry] Telemetry is enabled. Starting mananger...", "1"));
h.Now = h.Now.AddMilliseconds(100);
await h.Pass();
Assert.Empty(h.Completed);
// The service's first write after the script wait lands at ITS position — over the
// executor's line; the block is longer, so the pass sees a fragment at the bookmark.
h.Overwrite(servicePosition, Entry("Execution is done, collecting result", "17") + Entry(ResultLine, "17") + Entry("[PowerShell] Success, Result details length: 315", "17"));
h.Now = h.Now.AddMilliseconds(100);
await h.Pass();
h.Now = h.Now.AddMilliseconds(100);
await h.Pass();
var script = Assert.Single(h.Completed);
Assert.Equal(PlatformId, script.PolicyId);
Assert.Equal("Success", script.Result);
Assert.Equal("ime_policy_result", script.ResultSource);
Assert.Equal("User", script.RunContext);
Assert.Equal(1, h.Health.OverwriteRewinds);
}
}
}
@@ -33,6 +33,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
private readonly Stream _stream;
private readonly int _maxLineBytes;
private readonly bool _hashLines;
private readonly byte[] _readBuffer = new byte[ReadBufferSize];
private int _readPos;
private int _readLen;
@@ -41,21 +42,32 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
private byte[] _line = new byte[4096];
private int _lineLen;
private ulong _lineHash;
/// <param name="stream">Seekable stream positioned at the first byte to read.</param>
/// <param name="maxLineBytes">Upper bound of bytes kept per line; the remainder is discarded.</param>
public BoundedLineReader(Stream stream, int maxLineBytes)
/// <param name="hashLines">Fingerprint the raw bytes of every line (<see cref="LastLineHash"/>) — the
/// tracker's byte ledger needs it for multi-writer files only.</param>
public BoundedLineReader(Stream stream, int maxLineBytes, bool hashLines = false)
{
if (stream == null) throw new ArgumentNullException(nameof(stream));
if (maxLineBytes <= 0) throw new ArgumentOutOfRangeException(nameof(maxLineBytes));
_stream = stream;
_maxLineBytes = maxLineBytes;
_hashLines = hashLines;
_readBufferFileOffset = stream.Position;
}
/// <summary>Cap in bytes applied to every line.</summary>
public int MaxLineBytes => _maxLineBytes;
/// <summary>
/// FNV-1a 64 over the bytes the most recently returned line was decoded from (BOM and
/// line terminator excluded; the capped prefix of a truncated line). Zero unless the
/// reader was created with hashing on.
/// </summary>
public ulong LastLineHash => _lineHash;
/// <summary>Absolute file offset of the next unread byte (exact — no decoder read-ahead).</summary>
public long Position => _readBufferFileOffset + _readPos;
@@ -78,6 +90,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
LastLineTerminated = false;
LastLineTruncated = false;
_lineLen = 0;
_lineHash = _hashLines ? Fnv1a64.OffsetBasis : 0UL;
var sawAnyByte = false;
while (true)
@@ -154,6 +167,16 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
// tail (and thus any CR) discarded, so the check is skipped there.
if (LastLineTerminated && !LastLineTruncated && len > 0 && _line[start + len - 1] == (byte)'\r')
len--;
if (_hashLines)
{
// Fingerprint exactly the bytes the string is decoded from, so a tail that was
// first read unterminated and later gained its CRLF hashes the same.
var h = Fnv1a64.OffsetBasis;
var end = start + len;
for (var i = start; i < end; i++)
h = (h ^ _line[i]) * Fnv1a64.Prime;
_lineHash = h;
}
return len == 0 ? string.Empty : Encoding.UTF8.GetString(_line, start, len);
}
}
@@ -0,0 +1,171 @@
using System.Text;
namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
{
/// <summary>FNV-1a 64-bit — the entry fingerprint used by the tracker's byte ledger.</summary>
internal static class Fnv1a64
{
public const ulong OffsetBasis = 14695981039346656037UL;
public const ulong Prime = 1099511628211UL;
public static ulong Fold(ulong hash, ulong value) => (hash ^ value) * Prime;
}
/// <summary>One assembled CMTrace entry: its byte range in the file, the fingerprint of those bytes and the text.</summary>
internal struct AssembledEntry
{
public long Offset;
public int Length;
public ulong Hash;
public string Text;
}
/// <summary>
/// Turns physical lines into CMTrace entries: a line is one entry, unless it opens
/// <c>&lt;![LOG[</c> without its <c>]LOG]!&gt;</c> close — then the continuation lines up to the
/// closing one belong to it. Oversized lines and capped entries are dropped together with
/// their remaining continuation lines, so no fragment is ever matched as an entry.
/// <para>
/// The read loop and the ledger verification (<c>ImeLogTracker.Overwrite.cs</c>) both feed
/// this class, which is what makes their entry boundaries — and therefore the fingerprints
/// they compare — identical by construction.
/// </para>
/// </summary>
internal sealed class CmTraceEntryAssembler
{
public enum Outcome
{
/// <summary>The line was absorbed into an open multiline entry.</summary>
Buffered,
/// <summary>An entry is complete.</summary>
Entry,
/// <summary>A line over the byte cap was dropped (its entry with it).</summary>
OversizedDropped,
/// <summary>An open multiline entry hit the line/char cap and was dropped.</summary>
CapDropped,
/// <summary>A continuation line of a dropped entry was skipped.</summary>
Skipped,
}
private const string LogOpen = "<![LOG[";
private const string LogClose = "]LOG]!>";
private readonly int _maxLines;
private readonly int _maxChars;
private StringBuilder _buffer;
private int _lineCount;
private long _entryStart;
private ulong _entryHash;
private bool _skippingDroppedEntry;
public CmTraceEntryAssembler(int maxMultiLineBufferLines, int maxEntryBytes)
{
_maxLines = maxMultiLineBufferLines;
_maxChars = maxEntryBytes;
}
/// <summary>True while a multiline entry is being assembled.</summary>
public bool HasOpenEntry => _buffer != null;
/// <summary>File offset of the open multiline entry (valid while <see cref="HasOpenEntry"/>).</summary>
public long OpenEntryStart => _entryStart;
/// <summary>Line count of the entry dropped by the last <see cref="Outcome.CapDropped"/>.</summary>
public int DroppedLines { get; private set; }
/// <summary>Char count of the entry dropped by the last <see cref="Outcome.CapDropped"/>.</summary>
public int DroppedChars { get; private set; }
/// <summary>
/// Feeds one physical line. <paramref name="lineStart"/> / <paramref name="positionAfter"/>
/// are the line's file offsets (start, next unread byte), <paramref name="lineHash"/> the
/// reader's raw-byte fingerprint of the line.
/// </summary>
public Outcome Feed(string line, long lineStart, long positionAfter, bool truncated, ulong lineHash, out AssembledEntry entry)
{
entry = default(AssembledEntry);
if (truncated)
{
// The capped prefix only tells whether the line opened an entry whose remaining
// lines must be skipped rather than matched as raw text.
if (_buffer != null)
{
_buffer = null;
_lineCount = 0;
_skippingDroppedEntry = true;
}
else if (line.StartsWith(LogOpen) && !line.Contains(LogClose))
{
_skippingDroppedEntry = true;
}
return Outcome.OversizedDropped;
}
if (_skippingDroppedEntry)
{
// A line opening a new entry ends the skip and is processed itself — checked
// BEFORE the close-tag test, because a complete single-line entry contains both.
if (line.StartsWith(LogOpen))
_skippingDroppedEntry = false;
else
{
if (line.Contains(LogClose)) _skippingDroppedEntry = false;
return Outcome.Skipped;
}
}
if (_buffer != null)
{
_buffer.Append('\n').Append(line);
_lineCount++;
_entryHash = Fnv1a64.Fold(_entryHash, lineHash);
if (line.Contains(LogClose))
{
entry = new AssembledEntry
{
Offset = _entryStart,
Length = (int)(positionAfter - _entryStart),
Hash = _entryHash,
Text = _buffer.ToString(),
};
_buffer = null;
_lineCount = 0;
return Outcome.Entry;
}
if (_lineCount >= _maxLines || _buffer.Length >= _maxChars)
{
DroppedLines = _lineCount;
DroppedChars = _buffer.Length;
_buffer = null;
_lineCount = 0;
_skippingDroppedEntry = true;
return Outcome.CapDropped;
}
return Outcome.Buffered;
}
if (line.StartsWith(LogOpen) && !line.Contains(LogClose))
{
_buffer = new StringBuilder(line);
_lineCount = 1;
_entryStart = lineStart;
_entryHash = lineHash;
return Outcome.Buffered;
}
entry = new AssembledEntry
{
Offset = lineStart,
Length = (int)(positionAfter - lineStart),
Hash = lineHash,
Text = line,
};
return Outcome.Entry;
}
}
}
@@ -222,19 +222,14 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
// PowerShell script tracking handlers
// -----------------------------------------------------------------------
// Returns the current platform-script accumulator, or null when no platform script is active.
// Returns the platform-script accumulator that owns the entry being processed, or null.
// Health-script (remediation) state is no longer tracked line-by-line — the HS-NEW-RESULT
// pattern delivers the full pre-detection / remediation / post-detection JSON in one shot
// via HandleHealthScriptResult. The platform branch keeps its line-by-line accumulator
// because PS-* patterns (PS-SCRIPT-CONTEXT / PS-SCRIPT-EXITCODE / PS-AGENT-OUTPUT / …) still
// arrive across multiple log lines.
private ScriptExecutionState GetCurrentPlatformScript()
{
if (!string.IsNullOrEmpty(_lastPlatformScriptPolicyId) &&
_pendingPlatformScripts.TryGetValue(_lastPlatformScriptPolicyId, out var state))
return state;
return null;
}
// arrive across multiple log lines; ownership is by file position (see
// ImeLogTracker.Overwrite.cs), never by which script happened to start last.
private ScriptExecutionState GetCurrentPlatformScript() => ResolvePlatformScriptForCurrentEntry();
private void HandleScriptStarted(Match match, Dictionary<string, string> parameters)
{
@@ -320,7 +315,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
try { OnScriptStarted?.Invoke(new ScriptStartedInfo { PolicyId = id, ScriptType = "platform" }); }
catch (Exception ex) { _logger.Warning($"ImeLogTracker: OnScriptStarted handler threw: {ex.Message}"); }
}
_lastPlatformScriptPolicyId = id;
RecordInvocationMarker(id, isClose: false);
NoteInvocationActivity();
// Started lines fire twice per script (agentexecutor + ime source) and carry no
// outcome — the matching `platform script completed` line below carries result+exit
// and stays on Info. Keep starts on Debug so Info reflects script outcomes only.
@@ -338,33 +334,34 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
: GetCurrentPlatformScript();
}
// Session 6b4993e5 fix — platform-script stdout/exit cross-contamination.
//
// AgentExecutor.exe hosts BOTH platform scripts (…\Policies\Scripts\<id>.ps1) and
// proactive-remediation scripts (…\IMECache\HealthScripts\<id>\detect.ps1), and writes
// their lines interleaved into the shared AgentExecutor.log. The policyId-less
// PS-AGENT-EXITCODE ("Powershell exit code is N") and PS-AGENT-OUTPUT ("write output
// done. output = …") lines are routed to whichever platform script
// _lastPlatformScriptPolicyId points at. A remediation invocation's start
// ("Adding argument remediationScript …") does NOT match PS-AGENT-SCRIPT-START, so it
// never moves the pointer — its exit/output then bleed into the last-started PLATFORM
// script's slot (observed: platform c3e0124c emitted result=Failed but with exit 0 +
// stdout "[Compliant] No Classic Teams found" — the Teams remediation's output).
//
// Each AgentExecutor process logs "ExecutorLog AgentExecutor gets invoked" first, so a
// new banner means the previous invocation's line-capture context is over. Clearing the
// pointer here scopes line-capture to a single invocation: the immediately following
// "Adding argument powershell …\Policies\Scripts\<id>.ps1" (PS-AGENT-SCRIPT-START)
// re-establishes it for a platform invocation, while a remediation invocation leaves it
// null so its exit/output are dropped (the remediation's authoritative data still
// arrives via HS-NEW-RESULT). The platform script's final result is unaffected — it is
// keyed by policyId via PS-SCRIPT-RESULT, not by this pointer.
private void HandleAgentExecutorInvocationBoundary()
// AgentExecutor.exe hosts platform scripts (…\Policies\Scripts\<id>.ps1) as well as
// detection, requirement and proactive-remediation scripts, and its policyId-less
// PS-AGENT-EXITCODE / PS-AGENT-OUTPUT lines belong to whichever invocation owns that
// file position. The banner closes the previous invocation; the argument line that
// follows opens the next one — as a platform script when PS-AGENT-SCRIPT-START matched
// it, otherwise as an invocation whose exit/output are not captured (a remediation's
// authoritative data arrives via HS-NEW-RESULT; session 6b4993e5 saw its output bleed
// into a platform slot). "Agent executor completed" closes the invocation again.
private void HandleAgentExecutorInvocation()
{
if (_lastPlatformScriptPolicyId == null) return;
_logger.Debug(
$"ImeLogTracker: AgentExecutor invocation boundary — clearing platform-script line-capture pointer (was {_lastPlatformScriptPolicyId}).");
_lastPlatformScriptPolicyId = null;
RecordInvocationMarker(null, isClose: true);
NoteInvocationActivity();
}
private void HandleAgentInvocationArgument(string kind)
{
RecordInvocationMarker(null, isClose: false);
NoteInvocationActivity();
if (!string.IsNullOrEmpty(kind) && !string.Equals(kind, "powershell", StringComparison.OrdinalIgnoreCase))
_logger.Debug($"ImeLogTracker: AgentExecutor invocation of kind '{kind}' — exit/output lines not captured as a platform script");
}
private void HandleAgentExecutorCompleted()
{
RecordInvocationMarker(null, isClose: true);
// The end block was written at the executor's own stream position — if that lay
// behind the bookmark, the ledger check on the next pass finds it.
RequestOverwriteCheck();
}
private void HandleScriptContext(Match match, Dictionary<string, string> parameters)
@@ -372,6 +369,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
var context = match.Groups["context"]?.Value;
if (string.IsNullOrEmpty(context)) return;
NoteInvocationActivity();
var runContext = string.Equals(context, "machine", StringComparison.OrdinalIgnoreCase) ? "System" : "User";
var script = GetCurrentScriptForLineUpdate(parameters);
if (script != null)
@@ -398,6 +396,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
script.ExitObservedAtUtc = LastMatchedLogTimestamp ?? DateTime.UtcNow;
_logger.Debug($"ImeLogTracker: script exit code {exitCode} for {script.PolicyId}");
}
// An exit code means the executor just wrote its end block and IME is about to
// write the result — both land at a stale stream position when another process
// appended meanwhile. Check the ledgers on the next pass.
RequestOverwriteCheck();
}
private void HandleScriptOutput(Match match, Dictionary<string, string> parameters)
@@ -455,8 +457,6 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
{
_logger.Debug($"ImeLogTracker: platform script {id} PS-SCRIPT-RESULT arrived after fallback emit — skipping duplicate");
_pendingPlatformScripts.Remove(id);
if (string.Equals(_lastPlatformScriptPolicyId, id, StringComparison.OrdinalIgnoreCase))
_lastPlatformScriptPolicyId = null;
return;
}
@@ -496,17 +496,18 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
_platformScriptResultEmitted.Add(id);
_pendingPlatformScripts.Remove(id);
_stateDirty = true;
if (string.Equals(_lastPlatformScriptPolicyId, id, StringComparison.OrdinalIgnoreCase))
_lastPlatformScriptPolicyId = null;
// The result block is IME's first write after the script wait — the executor's end
// block in the other file may have landed behind the bookmark at the same time.
RequestOverwriteCheck();
}
/// <summary>
/// Grace period after a platform script's AgentExecutor exit code is observed before
/// <see cref="FlushStalePlatformScriptResults"/> emits a completion from that exit code.
/// IME logs its authoritative <c>PS-SCRIPT-RESULT</c> line within ~1 s of exit on healthy
/// reporting cycles, so 15 s lets the normal path win virtually always — the fallback is a
/// true safety net for the case where IME has not flushed its batch-send to the Microsoft
/// service before the (often short) Autopilot enrollment ends and the agent terminates.
/// IME logs its authoritative <c>PS-SCRIPT-RESULT</c> line within ~1 s of exit; when it
/// is missing after 15 s it was hidden by a concurrent writer (the ledger check in
/// <c>ImeLogTracker.Overwrite.cs</c> recovers that line) or the enrollment ended before
/// IME wrote it — this fallback is the safety net for the remaining cases.
/// </summary>
private static readonly TimeSpan PlatformScriptResultGrace = TimeSpan.FromSeconds(15);
@@ -578,11 +579,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
// the state file before a restart or the restarted tracker re-emits (H1).
_stateDirty = true;
foreach (var key in toRemove)
{
_pendingPlatformScripts.Remove(key);
if (string.Equals(_lastPlatformScriptPolicyId, key, StringComparison.OrdinalIgnoreCase))
_lastPlatformScriptPolicyId = null;
}
RequestOverwriteCheck();
}
}
@@ -602,7 +600,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
Stdout = stdout,
StartedAtUtc = startedAtUtc,
};
_lastPlatformScriptPolicyId = policyId;
NextTestEntry();
RecordInvocationMarker(policyId, isClose: false);
}
/// <summary>Test seam: simulate the authoritative IME PS-SCRIPT-RESULT path for a platform
@@ -616,8 +615,6 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
if (_platformScriptResultEmitted.Contains(policyId))
{
_pendingPlatformScripts.Remove(policyId);
if (string.Equals(_lastPlatformScriptPolicyId, policyId, StringComparison.OrdinalIgnoreCase))
_lastPlatformScriptPolicyId = null;
return;
}
@@ -642,8 +639,6 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
EmitScriptEvent(script);
_platformScriptResultEmitted.Add(policyId);
_pendingPlatformScripts.Remove(policyId);
if (string.Equals(_lastPlatformScriptPolicyId, policyId, StringComparison.OrdinalIgnoreCase))
_lastPlatformScriptPolicyId = null;
_stateDirty = true;
}
@@ -794,6 +789,8 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
return;
}
HandleHealthScriptResultJson(json);
// A remediation executor just ended too — its end block may sit behind the bookmark.
RequestOverwriteCheck();
}
/// <summary>
@@ -78,243 +78,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
try
{
var fileInfo = new FileInfo(filePath);
if (!fileInfo.Exists) continue;
// Captured BEFORE MarkChecked below stamps this pass. Growth measured
// against a previously observed state is what makes this pass's lines valid
// "written now" anchors — for the per-file measurement AND per-line anchoring.
var hadPreviousObservation = _positionTracker.HasSeen(filePath);
var lastCheckedUtc = _positionTracker.GetLastCheckedUtc(filePath);
var passNowUtc = UtcNowProvider();
var startPos = _positionTracker.GetSafePosition(filePath, fileInfo.Length);
// Every look counts, including "no new data" and an empty first sight: the
// NEXT pass's freshness window is measured from here. Restored bookmarks
// deliberately carry no LastCheckedUtc — the first pass after a restart reads
// downtime backlog and must never count as fresh.
_positionTracker.MarkChecked(filePath, passNowUtc);
if (startPos >= fileInfo.Length)
{
// M2: guard the interpolated string so it isn't built every 100 ms tick
// (per file) when Trace is off — which is the production default (Info).
if (_logger.LogLevel >= AgentLogLevel.Trace)
_logger.Trace($"ImeLogTracker: {Path.GetFileName(filePath)} — no new data (pos={startPos}, size={fileInfo.Length})");
continue;
}
if (_logger.LogLevel >= AgentLogLevel.Trace)
_logger.Trace($"ImeLogTracker: reading {Path.GetFileName(filePath)} from pos {startPos} (size={fileInfo.Length}, delta={fileInfo.Length - startPos})");
_currentSourceFileName = Path.GetFileName(filePath);
_currentPassLinesAreFresh = hadPreviousObservation
&& lastCheckedUtc > DateTime.MinValue
&& (passNowUtc - lastCheckedUtc) <= FreshLineMaxAge;
// Newest bias-less line of this pass — the calibration anchor. Bias-carrying
// lines are skipped: they already state the writer's offset, so they need no
// measurement and must not overwrite one.
CmTraceLogEntry calibrationAnchor = null;
using (var stream = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
{
stream.Seek(startPos, SeekOrigin.Begin);
var reader = new BoundedLineReader(stream, MaxEntryBytes);
var stats = new PassStats();
// Buffer for multiline CMTrace entries (e.g. AgentExecutor.log
// "write output done. output = ..." spans many lines)
StringBuilder multiLineBuffer = null;
int multiLineCount = 0;
// Set after a capped entry is dropped: its remaining physical lines
// are skipped instead of being matched as raw text, until the entry
// closes or a new entry begins (so no real entry is lost).
bool skippingDroppedEntry = false;
// File offset where the entry currently being assembled/processed
// begins — the bookmark to fall back to when that entry is NOT
// consumed (cancellation, or a held-back unterminated tail).
long entryStart = startPos;
// Where the next pass resumes. EOF unless an entry is left unconsumed.
long resumePos;
while (true)
{
// CancellationToken.None: cancellation is honoured by the explicit
// check below (exact bookmark) rather than by an exception out of
// the read, which the poll loop would log as an error.
var line = await reader.ReadLineAsync(CancellationToken.None);
if (line == null)
{
// EOF. An in-flight multiline entry here is an entry the writer
// has not finished — hold it back (see HoldBackTail) instead of
// dropping it and raw-matching its remaining lines next pass.
if (multiLineBuffer != null && HoldBackTail(filePath, entryStart, fileInfo.Length, passNowUtc))
{
_heldTailCount++;
resumePos = entryStart;
break;
}
resumePos = reader.Position;
break;
}
if (token.IsCancellationRequested)
{
// Exact bookmark: nothing of this line (or of the multiline
// entry it belongs to) has been handled.
resumePos = multiLineBuffer != null ? entryStart : reader.LastLineStart;
break;
}
if (multiLineBuffer == null)
entryStart = reader.LastLineStart;
_linesRead++;
// A physical line over the cap is never matched: its captures would be
// cut and the regexes would run over attacker-sized input. The capped
// prefix is enough to tell whether it opened an entry whose remaining
// lines must be skipped rather than raw-matched.
if (reader.LastLineTruncated)
{
stats.OversizedLines++;
_oversizedLines++;
if (multiLineBuffer != null)
{
multiLineBuffer = null;
multiLineCount = 0;
skippingDroppedEntry = true;
}
else if (line.StartsWith("<![LOG[") && !line.Contains("]LOG]!>"))
{
skippingDroppedEntry = true;
}
continue;
}
if (!reader.LastLineTerminated && multiLineBuffer == null
&& HoldBackTail(filePath, reader.LastLineStart, fileInfo.Length, passNowUtc))
{
// Unterminated single line at EOF: the writer is mid-line.
_heldTailCount++;
resumePos = reader.LastLineStart;
break;
}
if (skippingDroppedEntry)
{
// A line opening a new entry ends the skip and is processed
// itself — checked BEFORE the close-tag test, because a complete
// single-line entry contains both and must not be consumed as
// the dropped entry's close.
if (line.StartsWith("<![LOG["))
skippingDroppedEntry = false;
else
{
if (line.Contains("]LOG]!>")) skippingDroppedEntry = false;
continue;
}
}
// --- Multiline CMTrace buffering ---
// CMTrace entries: <![LOG[message]LOG]!><time=...>
// When message contains newlines, the entry spans multiple lines.
// We buffer until we find the closing ]LOG]!> tag.
if (multiLineBuffer != null)
{
// Continuing a multiline entry
multiLineBuffer.Append('\n').Append(line);
multiLineCount++;
if (line.Contains("]LOG]!>"))
{
// Entry complete — use the assembled line
line = multiLineBuffer.ToString();
multiLineBuffer = null;
multiLineCount = 0;
}
else if (multiLineCount >= MaxMultiLineBufferLines || multiLineBuffer.Length >= MaxEntryBytes)
{
// Safety limit — discard to bound memory and parser work. Warning
// (not Debug) so a capped entry is visible in the client log at the
// default level: a real IME entry this large would be news.
_logger.Warning($"ImeLogTracker: discarding multiline CMTrace buffer in {Path.GetFileName(filePath)} after {multiLineCount} lines / {multiLineBuffer.Length} chars (cap {MaxMultiLineBufferLines} lines / {MaxEntryBytes} chars) — entry dropped");
multiLineBuffer = null;
multiLineCount = 0;
skippingDroppedEntry = true;
continue;
}
else
{
// Still accumulating — read next line
continue;
}
}
else if (line.StartsWith("<![LOG[") && !line.Contains("]LOG]!>"))
{
// Start of a multiline CMTrace entry
multiLineBuffer = new StringBuilder(line);
multiLineCount = 1;
continue;
}
// --- Normal processing (single-line or completed multiline) ---
CmTraceLogEntry entry;
string messageToMatch;
_currentEntryOffset = entryStart;
if (CmTraceLogParser.TryParseLine(line, out entry))
{
messageToMatch = entry.Message;
if (entry.HasTimestamp && !entry.BiasMinutes.HasValue)
calibrationAnchor = entry;
}
else
{
// Non-CMTrace line - match raw
messageToMatch = line;
entry = null;
}
if (string.IsNullOrEmpty(messageToMatch)) continue;
// Simulation mode delay
if (SimulationMode && entry != null)
{
await ApplySimulationDelay(ResolveEntryUtc(entry), token);
}
MatchLine(filePath, line, messageToMatch, entry, stats);
}
ClearHeldTailIfConsumed(filePath, resumePos);
_positionTracker.SetPosition(filePath, resumePos);
_stateDirty = true;
backlogBytes += Math.Max(0, fileInfo.Length - resumePos);
// One line per pass and file, never per hostile line: the counters are
// the operator's only trace that matching was skipped or cut short.
if (stats.RegexTimeouts > 0 || stats.BudgetBreaks > 0 || stats.OversizedLines > 0)
{
_logger.Warning(
$"ImeLogTracker: {Path.GetFileName(filePath)} pass skipped work — " +
$"oversizedLines={stats.OversizedLines} (cap {MaxEntryBytes} bytes), " +
$"regexTimeouts={stats.RegexTimeouts}, lineBudgetBreaks={stats.BudgetBreaks}" +
(stats.FirstSkippedPatternId != null ? $", firstSkippedPattern={stats.FirstSkippedPatternId}" : string.Empty));
RaiseTrackerDegradedOnce(Path.GetFileName(filePath), stats.FirstSkippedPatternId);
}
// Calibrate AFTER the pass: this pass's lines were resolved with the
// offset established previously, at most one poll (100 ms) old. Buffering
// the pass to calibrate first is not an option — the first pass of
// AppWorkload.log can be hundreds of MB. The cost is a warm-up of one
// growing pass, during which lines fall back to the reader zone and are
// flagged as such.
if (hadPreviousObservation && calibrationAnchor != null)
CalibrateFrom(_currentSourceFileName, Path.GetFileName(filePath), calibrationAnchor);
_currentEntryOffset = -1;
}
backlogBytes += await ProcessFileAsync(filePath, token);
}
catch (FileNotFoundException) { }
catch (IOException ex)
@@ -330,6 +94,246 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
}
}
/// <summary>
/// One pass over one file: the bytes behind the bookmark, on a multi-writer file preceded
/// by the overwrite check that may move the bookmark back first (see
/// <c>ImeLogTracker.Overwrite.cs</c>). Returns the file's remaining backlog in bytes.
/// </summary>
private async Task<long> ProcessFileAsync(string filePath, CancellationToken token)
{
var fileInfo = new FileInfo(filePath);
if (!fileInfo.Exists) return 0;
// Captured BEFORE MarkChecked below stamps this pass. Growth measured
// against a previously observed state is what makes this pass's lines valid
// "written now" anchors — for the per-file measurement AND per-line anchoring.
var hadPreviousObservation = _positionTracker.HasSeen(filePath);
var lastCheckedUtc = _positionTracker.GetLastCheckedUtc(filePath);
var passNowUtc = UtcNowProvider();
var startPos = _positionTracker.GetSafePosition(filePath, fileInfo.Length);
// Every look counts, including "no new data" and an empty first sight: the
// NEXT pass's freshness window is measured from here. Restored bookmarks
// deliberately carry no LastCheckedUtc — the first pass after a restart reads
// downtime backlog and must never count as fresh.
_positionTracker.MarkChecked(filePath, passNowUtc);
var fileName = Path.GetFileName(filePath);
var ledger = IsMultiWriterLogFile(fileName) ? GetLedger(fileName) : null;
// Set by a rewind: entries that still read unchanged behind the old bookmark are
// skipped, and nothing below the old bookmark is fresh or a calibration anchor.
HashSet<LedgerEntry> rewoundEntries = null;
long oldBookmark = -1;
if (ledger != null)
{
var rewindTo = await PrepareLedgerAsync(filePath, fileName, ledger, startPos, passNowUtc, token);
if (rewindTo >= 0)
{
rewoundEntries = ApplyRewind(filePath, fileName, ledger, rewindTo, startPos);
oldBookmark = startPos;
startPos = rewindTo;
}
}
if (startPos >= fileInfo.Length)
{
// M2: guard the interpolated string so it isn't built every 100 ms tick
// (per file) when Trace is off — which is the production default (Info).
if (_logger.LogLevel >= AgentLogLevel.Trace)
_logger.Trace($"ImeLogTracker: {fileName} — no new data (pos={startPos}, size={fileInfo.Length})");
return 0;
}
if (_logger.LogLevel >= AgentLogLevel.Trace)
_logger.Trace($"ImeLogTracker: reading {fileName} from pos {startPos} (size={fileInfo.Length}, delta={fileInfo.Length - startPos})");
_currentSourceFileName = fileName;
var passLinesAreFresh = hadPreviousObservation
&& lastCheckedUtc > DateTime.MinValue
&& (passNowUtc - lastCheckedUtc) <= FreshLineMaxAge;
_currentPassLinesAreFresh = passLinesAreFresh;
// Newest bias-less line of this pass — the calibration anchor. Bias-carrying
// lines are skipped: they already state the writer's offset, so they need no
// measurement and must not overwrite one.
CmTraceLogEntry calibrationAnchor = null;
using (var stream = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
{
// Guard zone: the file grew and is open anyway — re-check the entries just behind
// the bookmark, where a stale writer's block lands when it is short.
if (ledger != null && rewoundEntries == null && ledger.Entries.Count > 0)
{
var guardFrom = ledger.OffsetOfEntryAtOrAfter(startPos - OverwriteGuardBytes);
if (guardFrom >= 0 && guardFrom < startPos)
{
var divergence = await FindOverwriteAsync(stream, ledger, guardFrom, startPos, token);
_verifiedBytes += startPos - guardFrom;
if (divergence >= 0)
{
// The guard saw only the tail of the ledger; the block may start earlier.
var earliest = await FindOverwriteAsync(stream, ledger, ledger.StartOffset, startPos, token);
_verifiedBytes += startPos - ledger.StartOffset;
_verifyPasses++;
if (earliest >= 0) divergence = earliest;
rewoundEntries = ApplyRewind(filePath, fileName, ledger, divergence, startPos);
oldBookmark = startPos;
startPos = divergence;
}
}
}
stream.Seek(startPos, SeekOrigin.Begin);
var reader = new BoundedLineReader(stream, MaxEntryBytes, hashLines: ledger != null);
var assembler = new CmTraceEntryAssembler(MaxMultiLineBufferLines, MaxEntryBytes);
var stats = new PassStats();
var firstEntry = true;
// Where the next pass resumes. EOF unless an entry is left unconsumed.
long resumePos;
while (true)
{
// CancellationToken.None: cancellation is honoured by the explicit
// check below (exact bookmark) rather than by an exception out of
// the read, which the poll loop would log as an error.
var line = await reader.ReadLineAsync(CancellationToken.None);
if (line == null)
{
// EOF. An in-flight multiline entry here is an entry the writer
// has not finished — hold it back (see HoldBackTail) instead of
// dropping it and raw-matching its remaining lines next pass.
if (assembler.HasOpenEntry && HoldBackTail(filePath, assembler.OpenEntryStart, fileInfo.Length, passNowUtc))
{
_heldTailCount++;
resumePos = assembler.OpenEntryStart;
break;
}
resumePos = reader.Position;
break;
}
if (token.IsCancellationRequested)
{
// Exact bookmark: nothing of this line (or of the multiline
// entry it belongs to) has been handled.
resumePos = assembler.HasOpenEntry ? assembler.OpenEntryStart : reader.LastLineStart;
break;
}
_linesRead++;
if (!reader.LastLineTruncated && !reader.LastLineTerminated && !assembler.HasOpenEntry
&& HoldBackTail(filePath, reader.LastLineStart, fileInfo.Length, passNowUtc))
{
// Unterminated single line at EOF: the writer is mid-line.
_heldTailCount++;
resumePos = reader.LastLineStart;
break;
}
AssembledEntry entry;
var outcome = assembler.Feed(line, reader.LastLineStart, reader.Position, reader.LastLineTruncated, reader.LastLineHash, out entry);
if (outcome == CmTraceEntryAssembler.Outcome.OversizedDropped)
{
// A physical line over the cap is never matched: its captures would be
// cut and the regexes would run over attacker-sized input.
stats.OversizedLines++;
_oversizedLines++;
continue;
}
if (outcome == CmTraceEntryAssembler.Outcome.CapDropped)
{
// Warning (not Debug) so a capped entry is visible in the client log at the
// default level: a real IME entry this large would be news.
_logger.Warning($"ImeLogTracker: discarding multiline CMTrace buffer in {fileName} after {assembler.DroppedLines} lines / {assembler.DroppedChars} chars (cap {MaxMultiLineBufferLines} lines / {MaxEntryBytes} chars) — entry dropped");
continue;
}
if (outcome != CmTraceEntryAssembler.Outcome.Entry) continue;
if (firstEntry)
{
firstEntry = false;
// A fragment right at the bookmark of a multi-writer file (the previous
// entry ended exactly here, this one does not open an entry) is the tail of
// a block a stale writer laid over already-read bytes: verify before reading on.
if (ledger != null && rewoundEntries == null && entry.Offset == ledger.EndOffset
&& ledger.LastFragmentOffset != entry.Offset && !entry.Text.StartsWith("<![LOG["))
{
ledger.VerifyRequested = true;
ledger.LastFragmentOffset = entry.Offset;
resumePos = entry.Offset;
break;
}
}
if (ledger != null) ledger.Record(entry);
if (rewoundEntries != null && rewoundEntries.Contains(new LedgerEntry { Offset = entry.Offset, Hash = entry.Hash }))
continue; // unchanged behind the old bookmark — processed before the rewind
_currentPassLinesAreFresh = passLinesAreFresh && entry.Offset >= oldBookmark;
// --- Normal processing (single-line or completed multiline) ---
CmTraceLogEntry parsed;
string messageToMatch;
_currentEntryOffset = entry.Offset;
if (CmTraceLogParser.TryParseLine(entry.Text, out parsed))
{
messageToMatch = parsed.Message;
if (parsed.HasTimestamp && !parsed.BiasMinutes.HasValue && entry.Offset >= oldBookmark)
calibrationAnchor = parsed;
}
else
{
// Non-CMTrace line - match raw
messageToMatch = entry.Text;
parsed = null;
}
if (string.IsNullOrEmpty(messageToMatch)) continue;
// Simulation mode delay
if (SimulationMode && parsed != null)
{
await ApplySimulationDelay(ResolveEntryUtc(parsed), token);
}
MatchLine(filePath, entry.Text, messageToMatch, parsed, stats);
}
ClearHeldTailIfConsumed(filePath, resumePos);
_positionTracker.SetPosition(filePath, resumePos);
_stateDirty = true;
if (ledger != null) ledger.TrimBelow(RetentionFloor(fileName, resumePos));
// One line per pass and file, never per hostile line: the counters are
// the operator's only trace that matching was skipped or cut short.
if (stats.RegexTimeouts > 0 || stats.BudgetBreaks > 0 || stats.OversizedLines > 0)
{
_logger.Warning(
$"ImeLogTracker: {fileName} pass skipped work — " +
$"oversizedLines={stats.OversizedLines} (cap {MaxEntryBytes} bytes), " +
$"regexTimeouts={stats.RegexTimeouts}, lineBudgetBreaks={stats.BudgetBreaks}" +
(stats.FirstSkippedPatternId != null ? $", firstSkippedPattern={stats.FirstSkippedPatternId}" : string.Empty));
RaiseTrackerDegradedOnce(fileName, stats.FirstSkippedPatternId);
}
// Calibrate AFTER the pass: this pass's lines were resolved with the
// offset established previously, at most one poll (100 ms) old. Buffering
// the pass to calibrate first is not an option — the first pass of
// AppWorkload.log can be hundreds of MB. The cost is a warm-up of one
// growing pass, during which lines fall back to the reader zone and are
// flagged as such.
if (hadPreviousObservation && calibrationAnchor != null)
CalibrateFrom(_currentSourceFileName, fileName, calibrationAnchor);
_currentEntryOffset = -1;
return Math.Max(0, fileInfo.Length - resumePos);
}
}
private sealed class PassStats
{
public int RegexTimeouts;
@@ -828,10 +832,18 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
break;
case "resetplatformscriptcontext":
// Session 6b4993e5 fix: a fresh AgentExecutor invocation banner
// ("ExecutorLog AgentExecutor gets invoked") ends the previous
// invocation's line-capture context. See HandleAgentExecutorInvocationBoundary.
HandleAgentExecutorInvocationBoundary();
// "ExecutorLog AgentExecutor gets invoked": a new process — the executor
// lines that follow belong to its own argument line, never to the
// previous invocation (session 6b4993e5).
HandleAgentExecutorInvocation();
break;
case "agentinvocationargument":
HandleAgentInvocationArgument(match.Groups["kind"]?.Value);
break;
case "agentexecutorcompleted":
HandleAgentExecutorCompleted();
break;
case "healthscriptresult":
@@ -864,6 +876,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
internal void ProcessLogMessageForTest(string message, DateTime? sourceTimestampUtc = null)
{
if (string.IsNullOrEmpty(message)) return;
NextTestEntry();
// The test seam hands in an already-resolved UTC instant, so it takes the same route a
// writer-declared bias does: TimestampUtc set, no zone left to guess.
var entry = sourceTimestampUtc.HasValue
@@ -0,0 +1,465 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
{
/// <summary>
/// Partial: overwrite-aware tailing of the two IME logs that several processes append to.
/// <para>
/// IME's trace listener opens each log once per process with FileMode.Append and
/// FileShare.ReadWrite and writes at its own stream position afterwards. AgentExecutor.exe
/// (two processes per user-context script, one per detection or remediation script) has such
/// a listener on AgentExecutor.log AND on IntuneManagementExtension.log. Whenever one process
/// resumes writing after another one appended, it overwrites those bytes: the final file
/// holds the last writer, while a tailer that trusts its bookmark read the first version and
/// never sees the second (session 46749560: 6 of 16 script results, 4 of 16 exit blocks).
/// </para>
/// <para>
/// So the bytes behind the bookmark are not immutable. The tracker keeps a fingerprint ledger
/// of every entry it read from these files, re-verifies it while script invocations are in
/// flight or right after a signal that a stale writer just wrote, and on the first changed
/// entry rewinds the bookmark — processing only entries it has not processed before.
/// </para>
/// </summary>
public partial class ImeLogTracker
{
/// <summary>Live logs with more than one writing process; archived rotations are closed files.</summary>
internal static readonly string[] MultiWriterLogFiles = { "AgentExecutor.log", "IntuneManagementExtension.log" };
/// <summary>Ledger depth behind the bookmark while no script is in flight.</summary>
internal const int LedgerRetentionBytes = 1024 * 1024;
/// <summary>Hard cap on the ledger depth, whatever is in flight.</summary>
internal const int LedgerMaxBytes = 4 * 1024 * 1024;
/// <summary>Entries this close to the bookmark are re-checked on every pass that found growth (the file is open anyway).</summary>
internal const int OverwriteGuardBytes = 4 * 1024;
/// <summary>After a restart the ledger is seeded from this many bytes before the persisted bookmark.</summary>
internal const int RestartSeedBytes = 64 * 1024;
/// <summary>Verification work per second while in contention — a deep ledger is checked less often, never skipped.</summary>
internal const long VerifyBudgetBytesPerSecond = 256 * 1024;
internal static readonly TimeSpan VerifyMinInterval = TimeSpan.FromSeconds(1);
/// <summary>How long an observed invocation keeps the files in contention on its own (detection executors have no pending slot).</summary>
internal static readonly TimeSpan InvocationContentionWindow = TimeSpan.FromSeconds(120);
/// <summary>A pending script older than this no longer holds the files in contention.</summary>
internal static readonly TimeSpan MaxPendingScriptContention = TimeSpan.FromHours(2);
private const int MaxInvocationMarkersPerFile = 512;
private const string TestSourceFileName = "TEST.log";
private const string LogOpenTag = "<![LOG[";
/// <summary>One processed entry: where it starts, how far it reaches, what its bytes hashed to.</summary>
internal struct LedgerEntry : IEquatable<LedgerEntry>
{
public long Offset;
public int Length;
public ulong Hash;
// Identity is (offset, hash): a tail first read unterminated and later completed
// keeps its hash but grows by its terminator.
public bool Equals(LedgerEntry other) => Offset == other.Offset && Hash == other.Hash;
public override bool Equals(object obj) => obj is LedgerEntry other && Equals(other);
public override int GetHashCode() => unchecked(((int)Offset * 397) ^ (int)Hash ^ (int)(Hash >> 32));
}
internal sealed class FileLedger
{
public readonly List<LedgerEntry> Entries = new List<LedgerEntry>();
public bool VerifyRequested;
public DateTime NextCadenceVerifyUtc = DateTime.MinValue;
public long LastFragmentOffset = -1;
public bool Seeded;
public bool RewindLogged;
public long StartOffset => Entries.Count > 0 ? Entries[0].Offset : -1;
public long EndOffset => Entries.Count > 0 ? Entries[Entries.Count - 1].Offset + Entries[Entries.Count - 1].Length : -1;
public int IndexOfFirstAtOrAfter(long offset)
{
int lo = 0, hi = Entries.Count;
while (lo < hi)
{
var mid = (lo + hi) >> 1;
if (Entries[mid].Offset < offset) lo = mid + 1; else hi = mid;
}
return lo;
}
/// <summary>Entry offset at or after <paramref name="offset"/>, or -1.</summary>
public long OffsetOfEntryAtOrAfter(long offset)
{
var idx = IndexOfFirstAtOrAfter(offset);
return idx < Entries.Count ? Entries[idx].Offset : -1;
}
public void Record(AssembledEntry e)
{
var idx = IndexOfFirstAtOrAfter(e.Offset);
if (idx < Entries.Count) Entries.RemoveRange(idx, Entries.Count - idx);
Entries.Add(new LedgerEntry { Offset = e.Offset, Length = e.Length, Hash = e.Hash });
}
/// <summary>Drops everything from <paramref name="offset"/> on and returns it — the entries a rewind must not process twice.</summary>
public HashSet<LedgerEntry> TruncateFrom(long offset)
{
var idx = IndexOfFirstAtOrAfter(offset);
var removed = new HashSet<LedgerEntry>();
for (var i = idx; i < Entries.Count; i++) removed.Add(Entries[i]);
if (idx < Entries.Count) Entries.RemoveRange(idx, Entries.Count - idx);
return removed;
}
public void TrimBelow(long floor)
{
var idx = IndexOfFirstAtOrAfter(floor);
if (idx > 0) Entries.RemoveRange(0, idx);
}
public void Clear()
{
Entries.Clear();
LastFragmentOffset = -1;
}
}
// Keyed by file name (unique within the log folder). Poll thread only.
private readonly Dictionary<string, FileLedger> _ledgers = new Dictionary<string, FileLedger>(StringComparer.OrdinalIgnoreCase);
private DateTime _contentionUntilUtc = DateTime.MinValue;
// Cumulative, restart-safe like the other health counters.
private long _overwriteRewinds;
private long _overwriteBytesReprocessed;
private long _verifyPasses;
private long _verifiedBytes;
internal static bool IsMultiWriterLogFile(string fileName)
{
foreach (var f in MultiWriterLogFiles)
if (string.Equals(f, fileName, StringComparison.OrdinalIgnoreCase)) return true;
return false;
}
private FileLedger GetLedger(string fileName)
{
FileLedger ledger;
if (!_ledgers.TryGetValue(fileName, out ledger))
{
ledger = new FileLedger();
_ledgers[fileName] = ledger;
}
return ledger;
}
/// <summary>A stale writer writes exactly when its script ends: every end signal asks both files for a check on the next pass.</summary>
private void RequestOverwriteCheck()
{
foreach (var ledger in _ledgers.Values) ledger.VerifyRequested = true;
}
private void NoteInvocationActivity()
{
_contentionUntilUtc = UtcNowProvider() + InvocationContentionWindow;
}
private bool InContention(DateTime nowUtc)
{
if (nowUtc < _contentionUntilUtc || _pendingHealthScript != null) return true;
foreach (var script in _pendingPlatformScripts.Values)
{
if (script.Result != null) continue;
if (!script.StartedAtUtc.HasValue || nowUtc - script.StartedAtUtc.Value < MaxPendingScriptContention) return true;
}
return false;
}
/// <summary>
/// Runs before a multi-writer file is read: rollover reset, restart seeding, then the
/// overwrite check when one is due. Returns the offset to rewind to, or -1.
/// </summary>
private async Task<long> PrepareLedgerAsync(string filePath, string fileName, FileLedger ledger, long bookmark, DateTime nowUtc, CancellationToken token)
{
if (ledger.Entries.Count > 0 && bookmark < ledger.EndOffset)
{
// The bookmark moved below what was read: the file was replaced (rollover).
ledger.Clear();
InvalidateInvocationMarkersFrom(fileName, 0);
}
if (!ledger.Seeded)
{
ledger.Seeded = true;
if (ledger.Entries.Count == 0 && bookmark > 0)
await SeedLedgerAsync(filePath, ledger, bookmark, token).ConfigureAwait(false);
}
var due = ledger.VerifyRequested || (InContention(nowUtc) && nowUtc >= ledger.NextCadenceVerifyUtc);
if (!due || ledger.Entries.Count == 0) return -1;
var from = ledger.StartOffset;
var bytes = bookmark - from;
if (bytes <= 0)
{
ledger.VerifyRequested = false;
return -1;
}
long divergence;
using (var stream = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
divergence = await FindOverwriteAsync(stream, ledger, from, bookmark, token).ConfigureAwait(false);
if (token.IsCancellationRequested) return -1;
ledger.VerifyRequested = false;
var paceSeconds = Math.Max(VerifyMinInterval.TotalSeconds, (double)bytes / VerifyBudgetBytesPerSecond);
ledger.NextCadenceVerifyUtc = nowUtc.AddSeconds(paceSeconds);
_verifyPasses++;
_verifiedBytes += bytes;
return divergence;
}
/// <summary>
/// Re-reads [fromOffset, toOffset) with the read loop's assembler and compares against the
/// ledger. Returns the offset of the first entry whose bytes changed — a ledger entry not
/// reproduced, or an entry the ledger never had — or -1 when everything reads as before.
/// </summary>
private async Task<long> FindOverwriteAsync(FileStream stream, FileLedger ledger, long fromOffset, long toOffset, CancellationToken token)
{
var entries = ledger.Entries;
var idx = ledger.IndexOfFirstAtOrAfter(fromOffset);
if (idx >= entries.Count) return -1;
stream.Seek(entries[idx].Offset, SeekOrigin.Begin);
var reader = new BoundedLineReader(stream, MaxEntryBytes, hashLines: true);
var assembler = new CmTraceEntryAssembler(MaxMultiLineBufferLines, MaxEntryBytes);
while (reader.Position < toOffset)
{
var line = await reader.ReadLineAsync(CancellationToken.None).ConfigureAwait(false);
if (line == null) break;
if (token.IsCancellationRequested) return -1;
AssembledEntry e;
if (assembler.Feed(line, reader.LastLineStart, reader.Position, reader.LastLineTruncated, reader.LastLineHash, out e)
!= CmTraceEntryAssembler.Outcome.Entry)
continue;
if (e.Offset >= toOffset) break;
if (idx < entries.Count && entries[idx].Offset < e.Offset) return entries[idx].Offset;
if (idx < entries.Count && entries[idx].Offset == e.Offset)
{
if (entries[idx].Hash != e.Hash) return e.Offset;
idx++;
continue;
}
return e.Offset; // an entry the ledger never had: the bytes changed here
}
return idx < entries.Count && entries[idx].Offset < toOffset ? entries[idx].Offset : -1;
}
/// <summary>After a restart the ledger is empty: fingerprint the entries just behind the persisted bookmark without processing them.</summary>
private async Task SeedLedgerAsync(string filePath, FileLedger ledger, long bookmark, CancellationToken token)
{
var from = Math.Max(0, bookmark - RestartSeedBytes);
using (var stream = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
{
stream.Seek(from, SeekOrigin.Begin);
var reader = new BoundedLineReader(stream, MaxEntryBytes, hashLines: true);
var assembler = new CmTraceEntryAssembler(MaxMultiLineBufferLines, MaxEntryBytes);
var aligned = from == 0;
while (reader.Position < bookmark)
{
var line = await reader.ReadLineAsync(CancellationToken.None).ConfigureAwait(false);
if (line == null || token.IsCancellationRequested) break;
if (!aligned)
{
// The seek may have landed mid-entry: wait for an entry boundary.
if (!line.StartsWith(LogOpenTag)) continue;
aligned = true;
}
AssembledEntry e;
if (assembler.Feed(line, reader.LastLineStart, reader.Position, reader.LastLineTruncated, reader.LastLineHash, out e)
== CmTraceEntryAssembler.Outcome.Entry
&& e.Offset + e.Length <= bookmark)
ledger.Record(e);
}
}
}
/// <summary>Moves the bookmark back to the first changed entry; returns the entries to skip if they still read unchanged behind it.</summary>
private HashSet<LedgerEntry> ApplyRewind(string filePath, string fileName, FileLedger ledger, long rewindTo, long bookmark)
{
var removed = ledger.TruncateFrom(rewindTo);
_positionTracker.SetPosition(filePath, rewindTo);
_heldTails.Remove(filePath);
InvalidateInvocationMarkersFrom(fileName, rewindTo);
_overwriteRewinds++;
_overwriteBytesReprocessed += bookmark - rewindTo;
var message = $"ImeLogTracker: {fileName} changed behind the bookmark at offset {rewindTo} " +
$"(a concurrent writer overwrote already-read bytes) — rewinding {bookmark - rewindTo} bytes";
if (!ledger.RewindLogged)
{
ledger.RewindLogged = true;
_logger.Info(message);
}
else
{
_logger.Debug(message);
}
return removed;
}
private long RetentionFloor(string fileName, long bookmark)
{
var floor = bookmark - LedgerRetentionBytes;
var oldest = OldestPendingScriptMarkerOffset(fileName);
if (oldest >= 0 && oldest < floor) floor = oldest;
var hardFloor = bookmark - LedgerMaxBytes;
return floor < hardFloor ? hardFloor : floor;
}
// -----------------------------------------------------------------------
// Invocation markers: who owns the policyId-less executor lines at a file position
// -----------------------------------------------------------------------
//
// AgentExecutor.log carries "Powershell exit code is N" / "write output done" without a
// policy id. Their owner is decided by FILE POSITION, not by the order lines were read:
// an executor writes its end block at its own stream position, i.e. directly after its
// own start block in the final bytes, so the nearest preceding invocation marker names
// it — and stays right after a rewind re-reads that region. A marker is a platform start
// (policy id), some other invocation (detection, remediation, requirement: null) or a
// close ("gets invoked" banner, "Agent executor completed").
private sealed class InvocationMarker
{
public long Offset;
public string PolicyId;
public bool IsClose;
}
private readonly Dictionary<string, List<InvocationMarker>> _invocationMarkers =
new Dictionary<string, List<InvocationMarker>>(StringComparer.OrdinalIgnoreCase);
private InvocationMarker _lastPlatformMarker;
private long _testEntryOrdinal;
private void RecordInvocationMarker(string policyId, bool isClose)
{
if (_currentEntryOffset < 0) return;
var file = _currentSourceFileName ?? TestSourceFileName;
List<InvocationMarker> list;
if (!_invocationMarkers.TryGetValue(file, out list))
{
list = new List<InvocationMarker>();
_invocationMarkers[file] = list;
}
if (list.Count > 0 && list[list.Count - 1].Offset == _currentEntryOffset)
{
// Two patterns on one line: the platform start outranks the generic argument line.
var last = list[list.Count - 1];
if (!isClose && policyId != null)
{
last.PolicyId = policyId;
last.IsClose = false;
_lastPlatformMarker = last;
}
return;
}
if (list.Count >= MaxInvocationMarkersPerFile) list.RemoveAt(0);
var marker = new InvocationMarker { Offset = _currentEntryOffset, PolicyId = policyId, IsClose = isClose };
list.Add(marker);
if (!isClose && policyId != null) _lastPlatformMarker = marker;
}
/// <summary>The pending platform script that owns the entry being processed, or null.</summary>
private ScriptExecutionState ResolvePlatformScriptForCurrentEntry()
{
InvocationMarker marker = null;
List<InvocationMarker> list = null;
if (_currentSourceFileName != null && _currentEntryOffset >= 0 && _invocationMarkers.TryGetValue(_currentSourceFileName, out list))
{
for (var i = list.Count - 1; i >= 0; i--)
{
if (list[i].Offset <= _currentEntryOffset)
{
marker = list[i];
break;
}
}
}
// A file without any marker (a context line after a rotation): the last platform start wins.
if (marker == null && (list == null || list.Count == 0)) marker = _lastPlatformMarker;
if (marker == null || marker.IsClose || marker.PolicyId == null) return null;
ScriptExecutionState state;
return _pendingPlatformScripts.TryGetValue(marker.PolicyId, out state) ? state : null;
}
private void InvalidateInvocationMarkersFrom(string fileName, long offset)
{
List<InvocationMarker> list;
if (!_invocationMarkers.TryGetValue(fileName, out list) || list.Count == 0) return;
var removedLast = false;
for (var i = list.Count - 1; i >= 0 && list[i].Offset >= offset; i--)
{
if (ReferenceEquals(list[i], _lastPlatformMarker)) removedLast = true;
list.RemoveAt(i);
}
if (removedLast)
{
_lastPlatformMarker = null;
for (var i = list.Count - 1; i >= 0; i--)
{
if (!list[i].IsClose && list[i].PolicyId != null)
{
_lastPlatformMarker = list[i];
break;
}
}
}
}
private long OldestPendingScriptMarkerOffset(string fileName)
{
List<InvocationMarker> list;
if (!_invocationMarkers.TryGetValue(fileName, out list)) return -1;
long oldest = -1;
foreach (var marker in list)
{
if (marker.IsClose || marker.PolicyId == null || !_pendingPlatformScripts.ContainsKey(marker.PolicyId)) continue;
if (oldest < 0 || marker.Offset < oldest) oldest = marker.Offset;
}
return oldest;
}
// -----------------------------------------------------------------------
// Test seams
// -----------------------------------------------------------------------
/// <summary>Same effect as an end signal: both multi-writer ledgers are verified on the next pass.</summary>
internal void RequestOverwriteCheckForTest() => RequestOverwriteCheck();
internal int LedgerEntryCountForTest(string fileName)
{
FileLedger ledger;
return _ledgers.TryGetValue(fileName, out ledger) ? ledger.Entries.Count : 0;
}
/// <summary>Gives a seam-fed message a file position of its own so positional attribution behaves like in a real pass.</summary>
internal void NextTestEntry()
{
if (_currentSourceFileName == null) _currentSourceFileName = TestSourceFileName;
_currentEntryOffset = ++_testEntryOrdinal * 4096;
}
}
}
@@ -181,7 +181,6 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
// arrive across multiple log entries and IME may interleave platform scripts).
private readonly Dictionary<string, ScriptExecutionState> _pendingPlatformScripts =
new Dictionary<string, ScriptExecutionState>(StringComparer.OrdinalIgnoreCase);
private string _lastPlatformScriptPolicyId;
// Health-script (remediation) line-by-line accumulator — single slot because IME
// executes health scripts SEQUENTIALLY within a session (verified across multiple
@@ -367,6 +366,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
FilesTailed = _filesTailed,
BacklogBytes = _backlogBytes,
ImeAgentVersion = _imeAgentVersionSeen,
OverwriteRewinds = _overwriteRewinds,
OverwriteBytesReprocessed = _overwriteBytesReprocessed,
VerifyPasses = _verifyPasses,
VerifiedBytes = _verifiedBytes,
PatternHits = new Dictionary<string, int>(_patternHits, StringComparer.OrdinalIgnoreCase),
};
}
@@ -1140,6 +1143,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
HealthScriptResultParseFailures = state.Health.HealthScriptResultParseFailures;
_imeAgentVersionSeen = state.Health.ImeAgentVersion;
_trackerDegradedFired = state.Health.TrackerDegradedFired;
_overwriteRewinds = state.Health.OverwriteRewinds;
_overwriteBytesReprocessed = state.Health.OverwriteBytesReprocessed;
_verifyPasses = state.Health.VerifyPasses;
_verifiedBytes = state.Health.VerifiedBytes;
}
if (state.PatternHits != null)
{
@@ -1209,6 +1216,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
HealthScriptResultParseFailures = HealthScriptResultParseFailures,
ImeAgentVersion = _imeAgentVersionSeen,
TrackerDegradedFired = _trackerDegradedFired,
OverwriteRewinds = _overwriteRewinds,
OverwriteBytesReprocessed = _overwriteBytesReprocessed,
VerifyPasses = _verifyPasses,
VerifiedBytes = _verifiedBytes,
};
state.PatternHits = new Dictionary<string, int>(_patternHits, StringComparer.OrdinalIgnoreCase);
}
@@ -46,6 +46,23 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
/// <summary>IME version as logged by IME itself ("Agent version is: …"), null until seen.</summary>
public string ImeAgentVersion { get; set; }
/// <summary>
/// Times the bookmark of a multi-writer log (AgentExecutor.log, IntuneManagementExtension.log)
/// was moved back because bytes behind it had been overwritten by a concurrent IME
/// process. Expected nonzero whenever user-context scripts or detection scripts ran —
/// recovered data, not skipped work.
/// </summary>
public long OverwriteRewinds { get; set; }
/// <summary>Bytes re-read by those rewinds (only entries that changed are processed again).</summary>
public long OverwriteBytesReprocessed { get; set; }
/// <summary>Full ledger verifications run (cadence while scripts are in flight, end signals, fragments).</summary>
public long VerifyPasses { get; set; }
/// <summary>Bytes re-read for verification, guard-zone checks included — the cost of the overwrite detection.</summary>
public long VerifiedBytes { get; set; }
/// <summary>Match count per enabled pattern ID — every enabled pattern is present, zeros included.</summary>
public IReadOnlyDictionary<string, int> PatternHits { get; set; }
@@ -165,6 +165,11 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Enrollment.Ime
public int HealthScriptResultParseFailures { get; set; }
public string ImeAgentVersion { get; set; }
public bool TrackerDegradedFired { get; set; }
// Overwrite detection counters (null-defaulting on older state files).
public long OverwriteRewinds { get; set; }
public long OverwriteBytesReprocessed { get; set; }
public long VerifyPasses { get; set; }
public long VerifiedBytes { get; set; }
}
public class PackageStateData
@@ -209,6 +209,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic
data["ime_line_budget_breaks"] = health.BudgetBreaks;
data["ime_held_tails"] = health.HeldTails;
data["ime_unanchored_patterns"] = health.UnanchoredPatterns;
data["ime_overwrite_rewinds"] = health.OverwriteRewinds;
data["ime_overwrite_bytes_reprocessed"] = health.OverwriteBytesReprocessed;
data["ime_verify_passes"] = health.VerifyPasses;
data["ime_verified_bytes"] = health.VerifiedBytes;
}
}
catch (Exception ex)
@@ -584,6 +584,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Termination
["heldTails"] = health.HeldTails,
["unanchoredPatterns"] = health.UnanchoredPatterns,
["hsParseFailures"] = health.HealthScriptResultParseFailures,
["overwriteRewinds"] = health.OverwriteRewinds,
["overwriteBytesReprocessed"] = health.OverwriteBytesReprocessed,
["verifyPasses"] = health.VerifyPasses,
["verifiedBytes"] = health.VerifiedBytes,
["agentVersion"] = _agentVersion ?? string.Empty,
};
if (!string.IsNullOrEmpty(health.ImeAgentVersion))
@@ -175,7 +175,7 @@ public class BuiltInRulesTests
"cancelStuckAndSetCurrent", "imeSessionChange", "imeImpersonation", "imeTokenFailure",
"enrollmentCompleted", "userSessionZeroApps", "updateDoTelemetry", "scriptStarted",
"scriptContext", "scriptExitCode", "scriptOutput", "scriptCompleted",
"resetPlatformScriptContext",
"resetPlatformScriptContext", "agentInvocationArgument", "agentExecutorCompleted",
"healthScriptResult", "healthScriptDetectionResult",
"captureExitCode", "captureHResult",
"captureAppVersion", "captureAppTypeWinGet", "captureAppTypeMsi",