fix: serialize desktop commands across CLI processes

2026-09-05 23:24:56 +08:00
parent e0d1ea1022
commit 786932e892
8 changed files with 244 additions and 85 deletions
@@ -31,6 +31,29 @@

 脱敏摘要见 `evidence/M6-20260905/`；完整本地证据位于 `artifacts/m6-final-validation/m6-validation-20260905T143832/`。历史失败证据不作为通过依据。

+## 已撤销的 72 小时候选运行（不得作为验收）
+
+报告可靠性修复 `e0d1ea1`：Core 56/56 通过，Windows 8 项验证全部 exit 0。阶段报告与最终报告共用完整性/心跳/异常检查；最终采样失败仍保留失败报告，报告使用临时文件原子替换。
+
+- 固定产物：`releases/e0d1ea1/WxAgent.Host.exe`。
+- SHA-256：`AB02B373EDCF76F8A4D392B55F3CF0464D37474ED4214D1C1E3F5FA0E67CED65`。
+- Windows 任务：`WxAgentM6Endurance-e0d1ea1`，仅手动启动，75 小时上限，无定时触发器。
+- 实际开始：2026-09-05 14:59:02 UTC；预期结束：2026-09-08 14:59:02 UTC。
+- 输出：`C:\Users\Rogee\wx-agent\m6-e0d1ea1-endurance72`。
+- 已检查到 15:03:05 UTC 的第 5 个样本；最后有效快照为 15:03:03 UTC，窗口可用、Errors 为空。这只证明启动成功和当时心跳正常。
+- 2026-09-05 15:18:00 UTC 主动停止该候选，确认旧进程数为 0；`aborted.json` 明确记录 `Accepted=false`、未完成请求时长。旧的 8/24/72 小时唤醒已删除。
+- 原因：审计发现原队列仅在单进程内互斥，旧 `ListenAsync` 也未复用新监听入口。修复与新候选验证完成前，不能继续用旧产物做最终长测。
+- Windows 11 环境尚未提供，仍保留未验收状态。
+
+## 跨进程队列补强
+
+- 同一 Windows 用户会话的 CLI 实例通过同一独占文件句柄串行执行 UI 操作；句柄由操作系统在进程退出时释放，不留下永久锁死的命名信号量计数。
+- 会话读取/定位（可能隐式恢复窗口）和关闭子窗口也纳入队列。
+- 旧 `ListenAsync` 复用 `ListenEventsAsync` 的会话校验与去重；等待队列时仍遵守监听时长。
+- 托盘激活和等待真实主控件的恢复过程遵守调用方取消令牌。
+- Linux 59/59 通过；完整 Release 构建 0 警告、0 错误。新增独立队列实例互斥、取消后可再次使用、非争用 I/O 异常不盲重试等回归。
+- Windows 跨进程门禁验证脚本：`scripts/windows/Test-CommandGate.ps1`，由 M6 验证脚本统一调用。其结果必须单独检查，不能仅凭 Linux 通过认定 Windows 通过。
+
 ## 尚待完成的验收

 - [ ] 8/24/72 小时真实监听及长时资源趋势。
@@ -27,6 +27,10 @@ foreach ($case in $cases) {
    [IO.File]::WriteAllText("$directory\$($case.Name).json", ($lines -join [Environment]::NewLine), $utf8)
    $results += @{ Name=$case.Name; ExitCode=$code; StartedAt=$started; FinishedAt=[DateTimeOffset]::UtcNow }
 }
+$started = [DateTimeOffset]::UtcNow
+& (Join-Path $PSScriptRoot 'Test-CommandGate.ps1') -Executable $exe -OutputPath "$directory\command-gate.json"
+$gateCode = $LASTEXITCODE
+$results += @{ Name='command-gate'; ExitCode=$gateCode; StartedAt=$started; FinishedAt=[DateTimeOffset]::UtcNow }
 $summary = @{
    BinarySha256=(Get-FileHash $exe -Algorithm SHA256).Hash
    SessionId=[Diagnostics.Process]::GetCurrentProcess().SessionId
@@ -0,0 +1,40 @@
+param(
+    [Parameter(Mandatory=$true)][string]$Executable,
+    [Parameter(Mandatory=$true)][string]$OutputPath
+)
+$ErrorActionPreference = 'Stop'
+if (Test-Path $OutputPath) { throw 'Do not overwrite earlier gate evidence.' }
+$exe = (Resolve-Path $Executable).Path
+$session = [Diagnostics.Process]::GetCurrentProcess().SessionId
+if ($session -eq 0) { throw 'Run in the interactive WeChat session, not Session 0.' }
+$lockPath = Join-Path ([Environment]::GetFolderPath('LocalApplicationData')) "WxAgent\ui-session-$session.lock"
+$before = & $exe recover-ui --timeout 30
+$beforeCode = $LASTEXITCODE
+if ($beforeCode -ne 0) { throw 'Baseline recovery failed; a contention test would not be meaningful.' }
+$lease = [IO.File]::Open($lockPath, [IO.FileMode]::OpenOrCreate, [IO.FileAccess]::ReadWrite, [IO.FileShare]::None)
+try {
+    $blocked = & $exe recover-ui --timeout 3
+    $blockedCode = $LASTEXITCODE
+    $errorCode = (($blocked -join [Environment]::NewLine) | ConvertFrom-Json).error
+    # The legacy listener must honor its own duration while another process holds the gate.
+    $listenOutput = @(& $exe chat listen --seconds 1 --timeout 10)
+    $listenCode = $LASTEXITCODE
+    $listenResult = ($listenOutput -join [Environment]::NewLine) | ConvertFrom-Json
+    $messageCount = @($listenResult.messages).Count
+} finally { $lease.Dispose() }
+$after = & $exe recover-ui --timeout 30
+$afterCode = $LASTEXITCODE
+$result = @{
+    SessionId=$session
+    BaselineExit=$beforeCode
+    ContendedExit=$blockedCode
+    ContendedError=$errorCode
+    ListenerExit=$listenCode
+    UnexpectedListenerMessageCount=$messageCount
+    RecoveryAfterReleaseExit=$afterCode
+    Passed=($beforeCode -eq 0 -and $blockedCode -eq 124 -and $errorCode -eq 'Timeout' -and
+        $listenCode -eq 0 -and $messageCount -eq 0 -and $afterCode -eq 0)
+}
+[IO.File]::WriteAllText([IO.Path]::GetFullPath($OutputPath), ($result | ConvertTo-Json), (New-Object Text.UTF8Encoding($false)))
+if (!$result.Passed) { exit 2 }
+exit 0
@@ -0,0 +1,58 @@
+namespace WxAgent.Core;
+
+/// <summary>Serializes commands locally and across processes; the OS releases the file handle after a crash.</summary>
+public sealed class InterprocessCommandGate : IDisposable
+{
+    private readonly SemaphoreSlim _local = new(1, 1);
+    private readonly string _path;
+    private FileStream? _lease;
+
+    public InterprocessCommandGate(string path)
+    {
+        _path = Path.GetFullPath(path);
+        Directory.CreateDirectory(Path.GetDirectoryName(_path)!);
+    }
+
+    public async Task WaitAsync(CancellationToken cancellationToken)
+    {
+        await _local.WaitAsync(cancellationToken).ConfigureAwait(false);
+        try
+        {
+            while (true)
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                try
+                {
+                    _lease = new FileStream(_path, FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.None);
+                    return;
+                }
+                catch (IOException exception) when ((exception.HResult & 0xffff) is 11 or 32 or 33)
+                {
+                    // EAGAIN on Unix; sharing/lock violation on Windows. Other I/O errors must surface.
+                    await Task.Delay(50, cancellationToken).ConfigureAwait(false);
+                }
+            }
+        }
+        catch
+        {
+            _local.Release();
+            throw;
+        }
+    }
+
+    public void Release()
+    {
+        try { _lease?.Dispose(); }
+        finally
+        {
+            _lease = null;
+            _local.Release();
+        }
+    }
+
+    public void Dispose()
+    {
+        _lease?.Dispose();
+        _local.Dispose();
+    }
+}
@@ -38,18 +38,21 @@ public static partial class WechatChatClient
        (await GetSubWindowsAsync(cancellationToken).ConfigureAwait(false))
            .FirstOrDefault(window => string.Equals(window.Title, title, StringComparison.Ordinal));

-    public static Task CloseSubWindowAsync(string title, CancellationToken cancellationToken = default)
+    public static async Task CloseSubWindowAsync(string title, CancellationToken cancellationToken = default)
    {
-        cancellationToken.ThrowIfCancellationRequested();
-        var processIds = Process.GetProcessesByName("Weixin").Select(process => process.Id).ToHashSet();
-        using var automation = new UIA3Automation();
-        var window = automation.GetDesktop().FindAllChildren(cf => cf.ByControlType(ControlType.Window))
-            .FirstOrDefault(candidate => processIds.Contains(candidate.Properties.ProcessId.ValueOrDefault)
-                && !candidate.Properties.IsOffscreen.ValueOrDefault
-                && !candidate.BoundingRectangle.IsEmpty
-                && string.Equals(SafeName(candidate), title, StringComparison.Ordinal));
-        window?.AsWindow().Close();
-        return Task.CompletedTask;
+        await CommandQueue.WaitAsync(cancellationToken).ConfigureAwait(false);
+        try
+        {
+            var processIds = Process.GetProcessesByName("Weixin").Select(process => process.Id).ToHashSet();
+            using var automation = new UIA3Automation();
+            var window = automation.GetDesktop().FindAllChildren(cf => cf.ByControlType(ControlType.Window))
+                .FirstOrDefault(candidate => processIds.Contains(candidate.Properties.ProcessId.ValueOrDefault)
+                    && !candidate.Properties.IsOffscreen.ValueOrDefault
+                    && !candidate.BoundingRectangle.IsEmpty
+                    && string.Equals(SafeName(candidate), title, StringComparison.Ordinal));
+            window?.AsWindow().Close();
+        }
+        finally { CommandQueue.Release(); }
    }

    public static Task SwitchToContactsAsync(CancellationToken cancellationToken = default) =>
@@ -355,7 +358,7 @@ public static partial class WechatChatClient
        try
        {
            using var automation = new UIA3Automation();
-            var main = AttachWindow(automation);
+            var main = AttachWindow(automation, cancellationToken);
            return await action(main, automation).ConfigureAwait(false);
        }
        catch (OperationCanceledException)
@@ -13,13 +13,19 @@ namespace WxAgent.Windows;

 public static partial class WechatChatClient
 {
-    private static readonly SemaphoreSlim CommandQueue = new(1, 1);
+    private static readonly InterprocessCommandGate CommandQueue = CreateCommandGate();

-    public static Task<IReadOnlyList<WechatSessionSnapshot>> ListVisibleSessionsAsync(CancellationToken cancellationToken)
+    private static InterprocessCommandGate CreateCommandGate()
+    {
+        using var process = System.Diagnostics.Process.GetCurrentProcess();
+        return new InterprocessCommandGate(Path.Combine(
+            Environment.GetFolderPath(Environment.SpecialFolder.LocalApplicationData),
+            "WxAgent", $"ui-session-{process.SessionId}.lock"));
+    }
+
+    public static Task<IReadOnlyList<WechatSessionSnapshot>> ListVisibleSessionsAsync(CancellationToken cancellationToken) =>
+        WithMainWindowAsync((window, _) =>
    {
-        cancellationToken.ThrowIfCancellationRequested();
-        using var automation = new UIA3Automation();
-        var window = AttachWindow(automation);
        var list = FindByAutomationId(window, WechatLocators.SessionList)
            ?? throw new WxAgentException(WxAgentErrorCode.ControlNotFound, $"Control {WechatLocators.SessionList} was not found.");
        var current = SafeName(FindByAutomationId(window, WechatLocators.CurrentChatName));
@@ -29,7 +35,7 @@ public static partial class WechatChatClient
                .Select(element => (SafeAutomationId(element), SafeName(element))),
            current);
        return Task.FromResult<IReadOnlyList<WechatSessionSnapshot>>(sessions);
-    }
+    }, cancellationToken);

    public static async Task<IReadOnlyList<WechatSessionSearchResult>> SearchSessionsAsync(
        string query,
@@ -41,7 +47,7 @@ public static partial class WechatChatClient
        try
        {
            using var automation = new UIA3Automation();
-            var window = AttachWindow(automation);
+            var window = AttachWindow(automation, cancellationToken);
            var search = window.FindAllDescendants().FirstOrDefault(element =>
                SafeControlType(element) == ControlType.Edit && SafeName(element) == WechatLocators.Search)
                ?? throw new WxAgentException(WxAgentErrorCode.ControlNotFound, "WeChat search box was not found.");
@@ -74,17 +80,15 @@ public static partial class WechatChatClient
        }
    }

-    public static Task<WechatSessionSnapshot?> GetCurrentSessionAsync(CancellationToken cancellationToken)
+    public static Task<WechatSessionSnapshot?> GetCurrentSessionAsync(CancellationToken cancellationToken) =>
+        WithMainWindowAsync((window, _) =>
    {
-        cancellationToken.ThrowIfCancellationRequested();
-        using var automation = new UIA3Automation();
-        var window = AttachWindow(automation);
        var element = FindByAutomationId(window, WechatLocators.CurrentChatName);
        var name = SafeName(element);
        return Task.FromResult(string.IsNullOrWhiteSpace(name)
            ? null
            : new WechatSessionSnapshot(name, SafeAutomationId(element!), true));
-    }
+    }, cancellationToken);

    public static async Task<WechatSessionSnapshot> OpenSessionAsync(string name, CancellationToken cancellationToken)
    {
@@ -93,7 +97,7 @@ public static partial class WechatChatClient
        try
        {
            using var automation = new UIA3Automation();
-            var window = AttachWindow(automation);
+            var window = AttachWindow(automation, cancellationToken);
            if (SafeName(FindByAutomationId(window, WechatLocators.CurrentChatName)) == name)
            {
                return new WechatSessionSnapshot(name, SafeAutomationId(FindByAutomationId(window, WechatLocators.CurrentChatName)!), true);
@@ -147,7 +151,7 @@ public static partial class WechatChatClient
        try
        {
            using var automation = new UIA3Automation();
-            var window = AttachWindow(automation);
+            var window = AttachWindow(automation, cancellationToken);
            await OpenFileTransferAssistantCoreAsync(window, cancellationToken);
            await Task.Delay(500, cancellationToken);

@@ -188,7 +192,7 @@ public static partial class WechatChatClient
        try
        {
            using var automation = new UIA3Automation();
-            var window = AttachWindow(automation);
+            var window = AttachWindow(automation, cancellationToken);
            await OpenFileTransferAssistantCoreAsync(window, cancellationToken);
            var before = ReadVisible(window).Select(message => message.Fingerprint).ToHashSet(StringComparer.Ordinal);
            var list = FindByAutomationId(window, WechatLocators.MessageList)
@@ -254,7 +258,7 @@ public static partial class WechatChatClient
        try
        {
            using var automation = new UIA3Automation();
-            var window = AttachWindow(automation);
+            var window = AttachWindow(automation, cancellationToken);
            await OpenFileTransferAssistantCoreAsync(window, cancellationToken);
            var before = ReadVisible(window).Select(message => message.Fingerprint).ToHashSet(StringComparer.Ordinal);
            var input = FindByAutomationId(window, WechatLocators.ChatInput)
@@ -303,7 +307,7 @@ public static partial class WechatChatClient
        try
        {
            using var automation = new UIA3Automation();
-            var window = AttachWindow(automation);
+            var window = AttachWindow(automation, cancellationToken);
            await OpenFileTransferAssistantCoreAsync(window, cancellationToken);
            var messages = new List<ChatMessageSnapshot>();
            var deduper = new BoundedMessageDeduper(Math.Max(2048, maxMessages * 4));
@@ -346,7 +350,7 @@ public static partial class WechatChatClient
        try
        {
            using var automation = new UIA3Automation();
-            var window = AttachWindow(automation);
+            var window = AttachWindow(automation, cancellationToken);
            await OpenFileTransferAssistantCoreAsync(window, cancellationToken);
            return ReadVisible(window);
        }
@@ -360,55 +364,10 @@ public static partial class WechatChatClient
        TimeSpan duration,
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
-        if (duration <= TimeSpan.Zero)
+        await foreach (var item in ListenEventsAsync(duration, checkpointPath: null, cancellationToken))
        {
-            throw new WxAgentException(WxAgentErrorCode.InvalidArgument, "Listen duration must be positive.");
-        }
-
-        using var automation = new UIA3Automation();
-        AutomationElement window;
-        await CommandQueue.WaitAsync(cancellationToken);
-        try
-        {
-            window = AttachWindow(automation);
-            await OpenFileTransferAssistantCoreAsync(window, cancellationToken);
-        }
-        finally
-        {
-            CommandQueue.Release();
-        }
-
-        var list = FindByAutomationId(window, WechatLocators.MessageList)
-            ?? throw new WxAgentException(WxAgentErrorCode.ControlNotFound, $"Control {WechatLocators.MessageList} was not found.");
-        var signals = Channel.CreateBounded<bool>(new BoundedChannelOptions(1) { FullMode = BoundedChannelFullMode.DropWrite });
-        using var subscription = list.RegisterStructureChangedEvent(
-            TreeScope.Subtree,
-            (_, _, _) => signals.Writer.TryWrite(true));
-        var deduper = new BoundedMessageDeduper();
-        foreach (var message in ReadVisible(window))
-        {
-            deduper.TryAdd(message.Fingerprint);
-        }
-
-        var deadline = DateTimeOffset.UtcNow + duration;
-        while (DateTimeOffset.UtcNow < deadline)
-        {
-            cancellationToken.ThrowIfCancellationRequested();
-            var remaining = deadline - DateTimeOffset.UtcNow;
-            var delay = Task.Delay(remaining < TimeSpan.FromSeconds(2) ? remaining : TimeSpan.FromSeconds(2), cancellationToken);
-            var eventReady = signals.Reader.WaitToReadAsync(cancellationToken).AsTask();
-            await Task.WhenAny(delay, eventReady);
-            while (signals.Reader.TryRead(out _))
-            {
-            }
-
-            foreach (var message in ReadVisible(window))
-            {
-                if (deduper.TryAdd(message.Fingerprint))
-                {
-                    yield return message;
-                }
-            }
+            if (item.Kind == MessageEventKind.MessageReceived && item.Message is { } message)
+                yield return message;
        }
    }

@@ -440,6 +399,8 @@ public static partial class WechatChatClient
        var connected = false;
        var announceReconnect = recoveredScan;
        var elapsed = System.Diagnostics.Stopwatch.StartNew();
+        using var readDeadline = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+        readDeadline.CancelAfter(duration);

        try
        {
@@ -449,14 +410,14 @@ public static partial class WechatChatClient
                IReadOnlyList<ChatMessageSnapshot>? visible = null;
                try
                {
-                    await CommandQueue.WaitAsync(cancellationToken);
+                    await CommandQueue.WaitAsync(readDeadline.Token);
                    try
                    {
-                        window = AttachWindow(automation);
+                        window = AttachWindow(automation, readDeadline.Token);
                        if (!IsFileTransferAssistantOpen(window)) connected = false;
                        if (!connected)
                        {
-                            await OpenFileTransferAssistantCoreAsync(window, cancellationToken);
+                            await OpenFileTransferAssistantCoreAsync(window, readDeadline.Token);
                            subscription?.Dispose();
                            var list = FindByAutomationId(window, WechatLocators.MessageList)
                                ?? throw new WxAgentException(WxAgentErrorCode.ControlNotFound, "Listener message list was not found.");
@@ -476,6 +437,10 @@ public static partial class WechatChatClient
                        CommandQueue.Release();
                    }
                }
+                catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested && readDeadline.IsCancellationRequested)
+                {
+                    break;
+                }
                catch (OperationCanceledException)
                {
                    throw;
@@ -577,21 +542,24 @@ public static partial class WechatChatClient
        }
    }

-    private static AutomationElement AttachWindow(UIA3Automation automation)
+    private static AutomationElement AttachWindow(UIA3Automation automation, CancellationToken cancellationToken)
    {
+        cancellationToken.ThrowIfCancellationRequested();
        var window = WechatDoctor.FindWechatWindow(automation);
        if (window is null || FindByAutomationId(window, WechatLocators.MainView) is null)
        {
-            if (!WechatTray.TryActivateUia(automation)) WechatTray.Inspect(true, CancellationToken.None);
+            cancellationToken.ThrowIfCancellationRequested();
+            if (!WechatTray.TryActivateUia(automation)) WechatTray.Inspect(true, cancellationToken);
            for (var attempt = 0; attempt < 50; attempt++)
            {
-                Thread.Sleep(100);
+                if (cancellationToken.WaitHandle.WaitOne(100)) cancellationToken.ThrowIfCancellationRequested();
                window = WechatDoctor.FindWechatWindow(automation);
                if (window is not null && FindByAutomationId(window, WechatLocators.MainView) is not null) break;
            }
        }
        if (window is null || FindByAutomationId(window, WechatLocators.MainView) is null)
            throw new WxAgentException(WxAgentErrorCode.ControlNotFound, "WeChat tray activation did not restore an accessible main view; no new process was started.");
+        cancellationToken.ThrowIfCancellationRequested();
        window.Focus();
        return window;
    }
@@ -151,7 +151,7 @@ public static partial class WechatChatClient
        {
            using var automation = new UIA3Automation();
            cancellationToken.ThrowIfCancellationRequested();
-            var main = AttachWindow(automation);
+            var main = AttachWindow(automation, cancellationToken);
            if (FindByAutomationId(main, WechatLocators.SessionList) is not null)
                return WechatOperationResult.Ok("WeChat UI is already on chats");

@@ -0,0 +1,63 @@
+using WxAgent.Core;
+using Xunit;
+
+namespace WxAgent.Core.Tests;
+
+public sealed class InterprocessCommandGateTests
+{
+    [Fact]
+    public async Task SeparateInstancesShareAnExclusiveLease()
+    {
+        var path = Path.Combine(Path.GetTempPath(), "wx-gate-" + Guid.NewGuid(), "commands.lock");
+        try
+        {
+            using var first = new InterprocessCommandGate(path);
+            using var second = new InterprocessCommandGate(path);
+            using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
+            await first.WaitAsync(timeout.Token);
+            var waiting = second.WaitAsync(timeout.Token);
+            Assert.False(waiting.IsCompleted);
+            first.Release();
+            await waiting;
+            second.Release();
+        }
+        finally { Directory.Delete(Path.GetDirectoryName(path)!, recursive: true); }
+    }
+
+    [Fact]
+    public async Task CancelledContentionDoesNotPoisonTheLocalQueue()
+    {
+        var path = Path.Combine(Path.GetTempPath(), "wx-gate-" + Guid.NewGuid(), "commands.lock");
+        try
+        {
+            using var first = new InterprocessCommandGate(path);
+            using var second = new InterprocessCommandGate(path);
+            using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
+            await first.WaitAsync(timeout.Token);
+            using var cancelled = new CancellationTokenSource(TimeSpan.FromMilliseconds(100));
+            await Assert.ThrowsAnyAsync<OperationCanceledException>(() => second.WaitAsync(cancelled.Token));
+            first.Release();
+            await second.WaitAsync(timeout.Token);
+            second.Release();
+        }
+        finally { Directory.Delete(Path.GetDirectoryName(path)!, recursive: true); }
+    }
+
+    [Fact]
+    public async Task NonContentionIoErrorsAreNotRetried()
+    {
+        var path = Path.Combine(Path.GetTempPath(), "wx-gate-" + Guid.NewGuid(), "commands.lock");
+        Directory.CreateDirectory(path); // A directory cannot be opened as the lock file.
+        try
+        {
+            using var gate = new InterprocessCommandGate(path);
+            using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
+            var error = await Record.ExceptionAsync(() => gate.WaitAsync(timeout.Token));
+            Assert.True(error is IOException or UnauthorizedAccessException);
+            Directory.Delete(path);
+            await gate.WaitAsync(timeout.Token);
+            gate.Release();
+        }
+        finally { Directory.Delete(Path.GetDirectoryName(path)!, recursive: true); }
+    }
+}