Implement worker MXAccess event queue

Issue #28: implement Advise, UnAdvise, AdviseSupervisory

docs/implementation-plan-mxaccess-worker.md

@@ -277,6 +277,8 @@ Live tests:
 
 Labels: `area:worker`, `type:feature`, `priority:p0`
 
+Status: implemented.
+
 Deliverables:
 
 - handlers for `OnDataChange`,

docs/mxaccess-worker-instance-design.md

@@ -348,9 +348,28 @@ Event handling rules:
 - Enqueue to the outbound event queue.
 - Return quickly to preserve message pumping.
 
-If event conversion throws, catch it inside the event handler, enqueue a
-structured `WorkerFault` or diagnostic event, and keep the worker alive only if
-the fault policy allows it.
+`MxAccessBaseEventSink` implements the COM connection-point handlers and keeps
+the handlers limited to event argument conversion plus enqueue. It uses
+`MxAccessEventMapper` to create `MxEvent` DTOs for `OnDataChange`,
+`OnWriteComplete`, `OperationComplete`, and `OnBufferedDataChange`. The mapper
+converts scalar and array values through `VariantConverter`, converts
+`MXSTATUS_PROXY[]` through `MxStatusProxyConverter`, and maps installed
+`MxDataType` values to the public protobuf enum while preserving the raw data
+type on buffered events. `OperationComplete` is only emitted from the native
+`OperationComplete` handler; write completion does not synthesize it.
+
+`MxAccessEventQueue` is the bounded outbound event queue for one worker
+session. It assigns the monotonic `WorkerSequence` and `WorkerTimestamp` when an
+event is accepted, preserving the order in which MXAccess handlers enqueue
+events. The default capacity is `10000`. When the queue reaches capacity it
+records a `WorkerFaultCategory.QueueOverflow` fault and rejects further events.
+The event handler catches conversion and enqueue failures, records the first
+fault on the queue, and returns to the STA message pump instead of writing to
+the pipe.
+
+If event conversion throws, catch it inside the event handler, record a
+structured `WorkerFault`, and keep the worker alive only if the fault policy
+allows it.
 
 ## Command Queue
 
@@ -451,6 +470,26 @@ cross-server handle behavior remains owned by MXAccess. COM exceptions continue
 through `StaCommandDispatcher`, which preserves the HRESULT and leaves
 diagnostic registry state unchanged for failed cleanup calls.
 
+`MxAccessCommandExecutor` implements advice lifecycle commands on the same STA
+path:
+
+- `Advise` calls `LMXProxyServerClass.Advise` with the requested server handle
+  and item handle.
+- `AdviseSupervisory` calls `LMXProxyServerClass.AdviseSupervisory` with the
+  requested server handle and item handle. This remains a distinct command from
+  plain `Advise` even though observed scalar captures share the same lower-level
+  subscription body.
+- `UnAdvise` calls `LMXProxyServerClass.UnAdvise` with the requested server
+  handle and item handle.
+
+The worker records plain and supervisory advice separately only after the COM
+call returns normally. Successful `UnAdvise` removes all tracked advice for the
+server and item pair because the public MXAccess cleanup method has no plain
+versus supervisory selector. Successful `RemoveItem` and `Unregister` also clear
+related advice state from the worker registry. Failed advice and cleanup calls
+leave registry state unchanged so diagnostics continue to reflect the last
+successful MXAccess-owned state transition.
+
 ## Handle Registry
 
 The worker should track MXAccess state for diagnostics and cleanup, while still
@@ -475,6 +514,9 @@ Rules:
 - Remove server handles only after `Unregister` succeeds.
 - Record item handles only after `AddItem` or `AddItem2` succeeds.
 - Remove item handles only after `RemoveItem` succeeds.
+- Record advice state only after `Advise` or `AdviseSupervisory` succeeds.
+- Remove advice state only after `UnAdvise`, `RemoveItem`, or `Unregister`
+  succeeds.
 - Preserve invalid-handle behavior from MXAccess.
 - Preserve cross-server handle behavior from MXAccess.
 - Use registry state for cleanup and diagnostics, not semantic correction.

src/MxGateway.Worker.Tests/MxAccess/MxAccessCommandExecutorTests.cs

@@ -177,6 +177,30 @@ public sealed class MxAccessCommandExecutorTests
         Assert.Empty(await session.GetRegisteredItemHandlesAsync());
     }
 
+    [Fact]
+    public async Task DispatchAsync_RemoveItemWithAdvisedHandle_RemovesTrackedAdviceAfterMxAccessSucceeds()
+    {
+        FakeMxAccessComObject fakeComObject = new(
+            registerHandle: 148,
+            addItemHandle: 603);
+        FakeMxAccessComObjectFactory factory = new(fakeComObject);
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+        await session.StartAsync(workerProcessId: 1234);
+        await session.DispatchAsync(CreateRegisterCommand("register-before-advised-remove", "client-a"));
+        await session.DispatchAsync(CreateAddItemCommand("add-before-advised-remove", 148, "Galaxy.Tag.Value"));
+        await session.DispatchAsync(CreateAdviseCommand("advise-before-remove", 148, 603));
+
+        MxCommandReply reply = await session.DispatchAsync(CreateRemoveItemCommand(
+            "remove-advised-item",
+            148,
+            603));
+
+        Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+        Assert.Empty(await session.GetRegisteredItemHandlesAsync());
+        Assert.Empty(await session.GetRegisteredAdviceHandlesAsync());
+    }
+
     [Fact]
     public async Task DispatchAsync_RemoveItemWithCrossServerHandle_PreservesHResultAndKeepsTrackedItemHandle()
     {
@@ -238,6 +262,158 @@ public sealed class MxAccessCommandExecutorTests
         Assert.Empty(await session.GetRegisteredItemHandlesAsync());
     }
 
+    [Fact]
+    public async Task DispatchAsync_Advise_CallsMxAccessOnStaAndTracksPlainAdvice()
+    {
+        FakeMxAccessComObject fakeComObject = new(
+            registerHandle: 52,
+            addItemHandle: 505);
+        FakeMxAccessComObjectFactory factory = new(fakeComObject);
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+        await session.StartAsync(workerProcessId: 1234);
+        await session.DispatchAsync(CreateRegisterCommand("register-before-advise", "client-a"));
+        await session.DispatchAsync(CreateAddItemCommand("add-before-advise", 52, "Galaxy.Tag.Value"));
+
+        MxCommandReply reply = await session.DispatchAsync(CreateAdviseCommand(
+            "advise",
+            52,
+            505));
+
+        Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+        Assert.True(reply.HasHresult);
+        Assert.Equal(0, reply.Hresult);
+        Assert.Equal(52, fakeComObject.AdviseServerHandle);
+        Assert.Equal(505, fakeComObject.AdvisedItemHandle);
+        Assert.Equal(runtime.StaThreadId, fakeComObject.AdviseThreadId);
+
+        RegisteredAdviceHandle adviceHandle = Assert.Single(
+            await session.GetRegisteredAdviceHandlesAsync());
+        Assert.Equal(52, adviceHandle.ServerHandle);
+        Assert.Equal(505, adviceHandle.ItemHandle);
+        Assert.Equal(MxAccessAdviceKind.Plain, adviceHandle.AdviceKind);
+    }
+
+    [Fact]
+    public async Task DispatchAsync_AdviseSupervisory_CallsDistinctMxAccessMethodAndTracksSupervisoryAdvice()
+    {
+        FakeMxAccessComObject fakeComObject = new(
+            registerHandle: 53,
+            addItemHandle: 506);
+        FakeMxAccessComObjectFactory factory = new(fakeComObject);
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+        await session.StartAsync(workerProcessId: 1234);
+        await session.DispatchAsync(CreateRegisterCommand("register-before-supervisory", "client-a"));
+        await session.DispatchAsync(CreateAddItemCommand("add-before-supervisory", 53, "Galaxy.Tag.Value"));
+
+        MxCommandReply reply = await session.DispatchAsync(CreateAdviseSupervisoryCommand(
+            "advise-supervisory",
+            53,
+            506));
+
+        Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+        Assert.Equal(53, fakeComObject.AdviseSupervisoryServerHandle);
+        Assert.Equal(506, fakeComObject.AdviseSupervisoryItemHandle);
+        Assert.Equal(runtime.StaThreadId, fakeComObject.AdviseSupervisoryThreadId);
+        Assert.Null(fakeComObject.AdviseServerHandle);
+
+        RegisteredAdviceHandle adviceHandle = Assert.Single(
+            await session.GetRegisteredAdviceHandlesAsync());
+        Assert.Equal(53, adviceHandle.ServerHandle);
+        Assert.Equal(506, adviceHandle.ItemHandle);
+        Assert.Equal(MxAccessAdviceKind.Supervisory, adviceHandle.AdviceKind);
+    }
+
+    [Fact]
+    public async Task DispatchAsync_UnAdvise_CallsMxAccessOnStaAndRemovesTrackedAdvice()
+    {
+        FakeMxAccessComObject fakeComObject = new(
+            registerHandle: 54,
+            addItemHandle: 507);
+        FakeMxAccessComObjectFactory factory = new(fakeComObject);
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+        await session.StartAsync(workerProcessId: 1234);
+        await session.DispatchAsync(CreateRegisterCommand("register-before-unadvise", "client-a"));
+        await session.DispatchAsync(CreateAddItemCommand("add-before-unadvise", 54, "Galaxy.Tag.Value"));
+        await session.DispatchAsync(CreateAdviseCommand("advise-before-unadvise", 54, 507));
+        await session.DispatchAsync(CreateAdviseSupervisoryCommand("supervisory-before-unadvise", 54, 507));
+
+        MxCommandReply reply = await session.DispatchAsync(CreateUnAdviseCommand(
+            "unadvise",
+            54,
+            507));
+
+        Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+        Assert.Equal(54, fakeComObject.UnAdviseServerHandle);
+        Assert.Equal(507, fakeComObject.UnAdvisedItemHandle);
+        Assert.Equal(runtime.StaThreadId, fakeComObject.UnAdviseThreadId);
+        Assert.Empty(await session.GetRegisteredAdviceHandlesAsync());
+    }
+
+    [Fact]
+    public async Task DispatchAsync_AdviseWhenMxAccessThrows_PreservesHResultAndDoesNotTrackAdvice()
+    {
+        const int hresult = unchecked((int)0x80070057);
+        FakeMxAccessComObject fakeComObject = new(
+            registerHandle: 55,
+            addItemHandle: 508,
+            adviseException: new COMException("Invalid item handle.", hresult));
+        FakeMxAccessComObjectFactory factory = new(fakeComObject);
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+        await session.StartAsync(workerProcessId: 1234);
+        await session.DispatchAsync(CreateRegisterCommand("register-before-advise-failure", "client-a"));
+        await session.DispatchAsync(CreateAddItemCommand("add-before-advise-failure", 55, "Galaxy.Tag.Value"));
+
+        MxCommandReply reply = await session.DispatchAsync(CreateAdviseCommand(
+            "advise-failure",
+            55,
+            999));
+
+        Assert.Equal(ProtocolStatusCode.MxaccessFailure, reply.ProtocolStatus.Code);
+        Assert.True(reply.HasHresult);
+        Assert.Equal(hresult, reply.Hresult);
+        Assert.Contains("0x80070057", reply.DiagnosticMessage);
+        Assert.Equal(55, fakeComObject.AdviseServerHandle);
+        Assert.Equal(999, fakeComObject.AdvisedItemHandle);
+        Assert.Empty(await session.GetRegisteredAdviceHandlesAsync());
+    }
+
+    [Fact]
+    public async Task DispatchAsync_UnAdviseWhenMxAccessThrows_PreservesHResultAndKeepsTrackedAdvice()
+    {
+        const int hresult = unchecked((int)0x80070057);
+        FakeMxAccessComObject fakeComObject = new(
+            registerHandle: 56,
+            addItemHandle: 509,
+            unAdviseException: new COMException("Invalid item handle.", hresult));
+        FakeMxAccessComObjectFactory factory = new(fakeComObject);
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+        await session.StartAsync(workerProcessId: 1234);
+        await session.DispatchAsync(CreateRegisterCommand("register-before-unadvise-failure", "client-a"));
+        await session.DispatchAsync(CreateAddItemCommand("add-before-unadvise-failure", 56, "Galaxy.Tag.Value"));
+        await session.DispatchAsync(CreateAdviseCommand("advise-before-unadvise-failure", 56, 509));
+
+        MxCommandReply reply = await session.DispatchAsync(CreateUnAdviseCommand(
+            "unadvise-failure",
+            56,
+            509));
+
+        Assert.Equal(ProtocolStatusCode.MxaccessFailure, reply.ProtocolStatus.Code);
+        Assert.True(reply.HasHresult);
+        Assert.Equal(hresult, reply.Hresult);
+        Assert.Contains("0x80070057", reply.DiagnosticMessage);
+        Assert.Equal(56, fakeComObject.UnAdviseServerHandle);
+        Assert.Equal(509, fakeComObject.UnAdvisedItemHandle);
+
+        RegisteredAdviceHandle adviceHandle = Assert.Single(
+            await session.GetRegisteredAdviceHandlesAsync());
+        Assert.Equal(MxAccessAdviceKind.Plain, adviceHandle.AdviceKind);
+    }
+
     [Fact]
     public async Task DispatchAsync_RegisterWithoutPayload_ReturnsInvalidRequest()
     {
@@ -278,6 +454,26 @@ public sealed class MxAccessCommandExecutorTests
         Assert.Null(factory.FakeComObject.AddItemDefinition);
     }
 
+    [Fact]
+    public async Task DispatchAsync_AdviseWithoutPayload_ReturnsInvalidRequest()
+    {
+        FakeMxAccessComObjectFactory factory = new(new FakeMxAccessComObject(registerHandle: 57));
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+        await session.StartAsync(workerProcessId: 1234);
+
+        MxCommandReply reply = await session.DispatchAsync(new StaCommand(
+            "session-1",
+            "missing-advise-payload",
+            new MxCommand
+            {
+                Kind = MxCommandKind.Advise,
+            }));
+
+        Assert.Equal(ProtocolStatusCode.InvalidRequest, reply.ProtocolStatus.Code);
+        Assert.Null(factory.FakeComObject.AdviseServerHandle);
+    }
+
     private static StaCommand CreateRegisterCommand(
         string correlationId,
         string clientName)
@@ -371,6 +567,63 @@ public sealed class MxAccessCommandExecutorTests
             });
     }
 
+    private static StaCommand CreateAdviseCommand(
+        string correlationId,
+        int serverHandle,
+        int itemHandle)
+    {
+        return new StaCommand(
+            "session-1",
+            correlationId,
+            new MxCommand
+            {
+                Kind = MxCommandKind.Advise,
+                Advise = new AdviseCommand
+                {
+                    ServerHandle = serverHandle,
+                    ItemHandle = itemHandle,
+                },
+            });
+    }
+
+    private static StaCommand CreateUnAdviseCommand(
+        string correlationId,
+        int serverHandle,
+        int itemHandle)
+    {
+        return new StaCommand(
+            "session-1",
+            correlationId,
+            new MxCommand
+            {
+                Kind = MxCommandKind.UnAdvise,
+                UnAdvise = new UnAdviseCommand
+                {
+                    ServerHandle = serverHandle,
+                    ItemHandle = itemHandle,
+                },
+            });
+    }
+
+    private static StaCommand CreateAdviseSupervisoryCommand(
+        string correlationId,
+        int serverHandle,
+        int itemHandle)
+    {
+        return new StaCommand(
+            "session-1",
+            correlationId,
+            new MxCommand
+            {
+                Kind = MxCommandKind.AdviseSupervisory,
+                AdviseSupervisory = new AdviseSupervisoryCommand
+                {
+                    ServerHandle = serverHandle,
+                    ItemHandle = itemHandle,
+                },
+            });
+    }
+
     private static StaRuntime CreateRuntime()
     {
         return new StaRuntime(
@@ -388,6 +641,9 @@ public sealed class MxAccessCommandExecutorTests
         private readonly Exception? addItemException;
         private readonly Exception? addItem2Exception;
         private readonly Exception? removeItemException;
+        private readonly Exception? adviseException;
+        private readonly Exception? unAdviseException;
+        private readonly Exception? adviseSupervisoryException;
 
         public FakeMxAccessComObject(
             int registerHandle,
@@ -396,7 +652,10 @@ public sealed class MxAccessCommandExecutorTests
             Exception? unregisterException = null,
             Exception? addItemException = null,
             Exception? addItem2Exception = null,
-            Exception? removeItemException = null)
+            Exception? removeItemException = null,
+            Exception? adviseException = null,
+            Exception? unAdviseException = null,
+            Exception? adviseSupervisoryException = null)
         {
             this.registerHandle = registerHandle;
             this.addItemHandle = addItemHandle;
@@ -405,6 +664,9 @@ public sealed class MxAccessCommandExecutorTests
             this.addItemException = addItemException;
             this.addItem2Exception = addItem2Exception;
             this.removeItemException = removeItemException;
+            this.adviseException = adviseException;
+            this.unAdviseException = unAdviseException;
+            this.adviseSupervisoryException = adviseSupervisoryException;
         }
 
         public string? RegisteredClientName { get; private set; }
@@ -435,6 +697,24 @@ public sealed class MxAccessCommandExecutorTests
 
         public int? RemoveItemThreadId { get; private set; }
 
+        public int? AdviseServerHandle { get; private set; }
+
+        public int? AdvisedItemHandle { get; private set; }
+
+        public int? AdviseThreadId { get; private set; }
+
+        public int? UnAdviseServerHandle { get; private set; }
+
+        public int? UnAdvisedItemHandle { get; private set; }
+
+        public int? UnAdviseThreadId { get; private set; }
+
+        public int? AdviseSupervisoryServerHandle { get; private set; }
+
+        public int? AdviseSupervisoryItemHandle { get; private set; }
+
+        public int? AdviseSupervisoryThreadId { get; private set; }
+
         public int Register(string clientName)
         {
             RegisteredClientName = clientName;
@@ -501,6 +781,48 @@ public sealed class MxAccessCommandExecutorTests
                 throw removeItemException;
             }
         }
+
+        public void Advise(
+            int serverHandle,
+            int itemHandle)
+        {
+            AdviseServerHandle = serverHandle;
+            AdvisedItemHandle = itemHandle;
+            AdviseThreadId = Environment.CurrentManagedThreadId;
+
+            if (adviseException is not null)
+            {
+                throw adviseException;
+            }
+        }
+
+        public void UnAdvise(
+            int serverHandle,
+            int itemHandle)
+        {
+            UnAdviseServerHandle = serverHandle;
+            UnAdvisedItemHandle = itemHandle;
+            UnAdviseThreadId = Environment.CurrentManagedThreadId;
+
+            if (unAdviseException is not null)
+            {
+                throw unAdviseException;
+            }
+        }
+
+        public void AdviseSupervisory(
+            int serverHandle,
+            int itemHandle)
+        {
+            AdviseSupervisoryServerHandle = serverHandle;
+            AdviseSupervisoryItemHandle = itemHandle;
+            AdviseSupervisoryThreadId = Environment.CurrentManagedThreadId;
+
+            if (adviseSupervisoryException is not null)
+            {
+                throw adviseSupervisoryException;
+            }
+        }
     }
 
     private sealed class FakeMxAccessComObjectFactory : IMxAccessComObjectFactory
@@ -520,7 +842,9 @@ public sealed class MxAccessCommandExecutorTests
 
     private sealed class NoopEventSink : IMxAccessEventSink
     {
-        public void Attach(object mxAccessComObject)
+        public void Attach(
+            object mxAccessComObject,
+            string sessionId)
         {
         }
 

src/MxGateway.Worker.Tests/MxAccess/MxAccessEventMapperTests.cs

@@ -0,0 +1,117 @@
+using System;
+using MxGateway.Contracts.Proto;
+using MxGateway.Worker.MxAccess;
+
+namespace MxGateway.Worker.Tests.MxAccess;
+
+public sealed class MxAccessEventMapperTests
+{
+    private readonly MxAccessEventMapper mapper = new();
+
+    [Fact]
+    public void CreateOnDataChange_ConvertsValueTimestampQualityAndStatuses()
+    {
+        DateTime timestamp = new(2026, 4, 26, 12, 30, 0, DateTimeKind.Utc);
+        FakeStatus[] statuses =
+        {
+            new()
+            {
+                success = -1,
+                category = 0,
+                detectedBy = 5,
+                detail = 0,
+            },
+        };
+
+        MxEvent mxEvent = mapper.CreateOnDataChange(
+            "session-1",
+            serverHandle: 12,
+            itemHandle: 34,
+            value: 42,
+            quality: 192,
+            timestamp: timestamp,
+            statuses: statuses);
+
+        Assert.Equal(MxEventFamily.OnDataChange, mxEvent.Family);
+        Assert.Equal("session-1", mxEvent.SessionId);
+        Assert.Equal(12, mxEvent.ServerHandle);
+        Assert.Equal(34, mxEvent.ItemHandle);
+        Assert.Equal(42, mxEvent.Value.Int32Value);
+        Assert.Equal(192, mxEvent.Quality);
+        Assert.Equal(timestamp, mxEvent.SourceTimestamp.ToDateTime());
+        Assert.Equal(MxEvent.BodyOneofCase.OnDataChange, mxEvent.BodyCase);
+
+        MxStatusProxy status = Assert.Single(mxEvent.Statuses);
+        Assert.Equal(-1, status.Success);
+        Assert.Equal(MxStatusCategory.Ok, status.Category);
+        Assert.Equal(MxStatusSource.RespondingAutomationObject, status.DetectedBy);
+    }
+
+    [Fact]
+    public void CreateOnWriteCompleteAndOperationComplete_PreservesDistinctFamilies()
+    {
+        MxEvent writeComplete = mapper.CreateOnWriteComplete(
+            "session-1",
+            serverHandle: 1,
+            itemHandle: 2,
+            statuses: Array.Empty<FakeStatus>());
+        MxEvent operationComplete = mapper.CreateOperationComplete(
+            "session-1",
+            serverHandle: 1,
+            itemHandle: 2,
+            statuses: Array.Empty<FakeStatus>());
+
+        Assert.Equal(MxEventFamily.OnWriteComplete, writeComplete.Family);
+        Assert.Equal(MxEvent.BodyOneofCase.OnWriteComplete, writeComplete.BodyCase);
+        Assert.Equal(MxEventFamily.OperationComplete, operationComplete.Family);
+        Assert.Equal(MxEvent.BodyOneofCase.OperationComplete, operationComplete.BodyCase);
+    }
+
+    [Fact]
+    public void CreateOnBufferedDataChange_PreservesRawDataTypeAndArrayMetadata()
+    {
+        DateTime firstTimestamp = new(2026, 4, 26, 13, 0, 0, DateTimeKind.Utc);
+        DateTime secondTimestamp = new(2026, 4, 26, 13, 1, 0, DateTimeKind.Utc);
+
+        MxEvent mxEvent = mapper.CreateOnBufferedDataChange(
+            "session-1",
+            serverHandle: 10,
+            itemHandle: 20,
+            rawDataType: 2,
+            value: new[] { 7, 8 },
+            quality: new[] { 192, 0 },
+            timestamp: new[] { firstTimestamp, secondTimestamp },
+            statuses: null);
+
+        Assert.Equal(MxEventFamily.OnBufferedDataChange, mxEvent.Family);
+        Assert.Equal(MxDataType.Integer, mxEvent.OnBufferedDataChange.DataType);
+        Assert.Equal(2, mxEvent.OnBufferedDataChange.RawDataType);
+        Assert.Equal(MxDataType.Integer, mxEvent.Value.ArrayValue.ElementDataType);
+        Assert.Equal(new[] { 7, 8 }, mxEvent.Value.ArrayValue.Int32Values.Values);
+        Assert.Equal(new[] { 192, 0 }, mxEvent.OnBufferedDataChange.QualityValues.Int32Values.Values);
+        Assert.Equal(2, mxEvent.OnBufferedDataChange.TimestampValues.TimestampValues.Values.Count);
+    }
+
+    [Theory]
+    [InlineData(-1, MxDataType.Unknown)]
+    [InlineData(0, MxDataType.NoData)]
+    [InlineData(1, MxDataType.Boolean)]
+    [InlineData(2, MxDataType.Integer)]
+    [InlineData(6, MxDataType.Time)]
+    [InlineData(15, MxDataType.InternationalizedString)]
+    [InlineData(999, MxDataType.Unknown)]
+    public void MapMxDataType_MapsInstalledMxAccessValues(
+        int rawDataType,
+        MxDataType expectedDataType)
+    {
+        Assert.Equal(expectedDataType, MxAccessEventMapper.MapMxDataType(rawDataType));
+    }
+
+    private sealed class FakeStatus
+    {
+        public int success;
+        public int category;
+        public int detectedBy;
+        public int detail;
+    }
+}

src/MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs

@@ -0,0 +1,106 @@
+using System;
+using System.Collections.Generic;
+using MxGateway.Contracts.Proto;
+using MxGateway.Worker.MxAccess;
+
+namespace MxGateway.Worker.Tests.MxAccess;
+
+public sealed class MxAccessEventQueueTests
+{
+    [Fact]
+    public void Enqueue_AssignsMonotonicWorkerSequencesAndPreservesOrder()
+    {
+        MxAccessEventQueue queue = new(capacity: 4);
+
+        WorkerEvent first = queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 10));
+        WorkerEvent second = queue.Enqueue(CreateEvent(MxEventFamily.OnWriteComplete, itemHandle: 11));
+
+        Assert.Equal(1UL, first.Event.WorkerSequence);
+        Assert.Equal(2UL, second.Event.WorkerSequence);
+        Assert.NotNull(first.Event.WorkerTimestamp);
+        Assert.Equal(2, queue.Count);
+        Assert.Equal(2UL, queue.LastEventSequence);
+
+        Assert.True(queue.TryDequeue(out WorkerEvent? dequeuedFirst));
+        Assert.True(queue.TryDequeue(out WorkerEvent? dequeuedSecond));
+        Assert.Equal(10, dequeuedFirst?.Event.ItemHandle);
+        Assert.Equal(11, dequeuedSecond?.Event.ItemHandle);
+        Assert.False(queue.TryDequeue(out _));
+    }
+
+    [Fact]
+    public void Drain_RemovesAtMostRequestedEvents()
+    {
+        MxAccessEventQueue queue = new(capacity: 4);
+        queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 10));
+        queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 11));
+        queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 12));
+
+        IReadOnlyList<WorkerEvent> drained = queue.Drain(maxEvents: 2);
+
+        Assert.Equal(2, drained.Count);
+        Assert.Equal(10, drained[0].Event.ItemHandle);
+        Assert.Equal(11, drained[1].Event.ItemHandle);
+        Assert.Equal(1, queue.Count);
+    }
+
+    [Fact]
+    public void Enqueue_WhenCapacityIsExceeded_RecordsOverflowFaultAndRejectsNewEvents()
+    {
+        MxAccessEventQueue queue = new(capacity: 1);
+        queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 10));
+
+        MxAccessEventQueueOverflowException overflow = Assert.Throws<MxAccessEventQueueOverflowException>(
+            () => queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 11)));
+
+        Assert.Equal(1, overflow.Capacity);
+        Assert.True(queue.IsFaulted);
+        Assert.Equal(WorkerFaultCategory.QueueOverflow, queue.Fault?.Category);
+        Assert.Equal(ProtocolStatusCode.WorkerUnavailable, queue.Fault?.ProtocolStatus.Code);
+        Assert.Throws<InvalidOperationException>(
+            () => queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 12)));
+    }
+
+    [Fact]
+    public void RecordFault_KeepsFirstFault()
+    {
+        MxAccessEventQueue queue = new(capacity: 1);
+        queue.RecordFault(new WorkerFault
+        {
+            Category = WorkerFaultCategory.MxaccessEventConversionFailed,
+        });
+        queue.RecordFault(new WorkerFault
+        {
+            Category = WorkerFaultCategory.QueueOverflow,
+        });
+
+        Assert.True(queue.IsFaulted);
+        Assert.Equal(WorkerFaultCategory.MxaccessEventConversionFailed, queue.Fault?.Category);
+    }
+
+    private static MxEvent CreateEvent(
+        MxEventFamily family,
+        int itemHandle)
+    {
+        MxEvent mxEvent = new()
+        {
+            Family = family,
+            SessionId = "session-1",
+            ServerHandle = 1,
+            ItemHandle = itemHandle,
+        };
+
+        switch (family)
+        {
+            case MxEventFamily.OnWriteComplete:
+                mxEvent.OnWriteComplete = new OnWriteCompleteEvent();
+                break;
+
+            default:
+                mxEvent.OnDataChange = new OnDataChangeEvent();
+                break;
+        }
+
+        return mxEvent;
+    }
+}

src/MxGateway.Worker.Tests/MxAccess/MxAccessLiveComCreationTests.cs

@@ -215,6 +215,127 @@ public sealed class MxAccessLiveComCreationTests
         }
     }
 
+    [Fact]
+    public async Task AdviseAndUnAdvise_WhenOptedIn_RoundTripsInstalledMxAccessSubscription()
+    {
+        if (!RunLiveMxAccessTests())
+        {
+            return;
+        }
+
+        using MxAccessStaSession session = new();
+        await session.StartAsync(workerProcessId: 1234);
+
+        MxCommandReply registerReply = await RegisterLiveSessionAsync(session, "live-advise-register");
+        int serverHandle = registerReply.Register.ServerHandle;
+        int itemHandle = 0;
+        bool advised = false;
+
+        try
+        {
+            MxCommandReply addItemReply = await session.DispatchAsync(new StaCommand(
+                "session-1",
+                "live-advise-add-item",
+                new MxCommand
+                {
+                    Kind = MxCommandKind.AddItem,
+                    AddItem = new AddItemCommand
+                    {
+                        ServerHandle = serverHandle,
+                        ItemDefinition = GetLiveAddItemReference(),
+                    },
+                }));
+
+            Assert.Equal(ProtocolStatusCode.Ok, addItemReply.ProtocolStatus.Code);
+            Assert.True(addItemReply.AddItem.ItemHandle > 0);
+            itemHandle = addItemReply.AddItem.ItemHandle;
+
+            MxCommandReply adviseReply = await session.DispatchAsync(new StaCommand(
+                "session-1",
+                "live-advise",
+                new MxCommand
+                {
+                    Kind = MxCommandKind.Advise,
+                    Advise = new AdviseCommand
+                    {
+                        ServerHandle = serverHandle,
+                        ItemHandle = itemHandle,
+                    },
+                }));
+
+            Assert.Equal(ProtocolStatusCode.Ok, adviseReply.ProtocolStatus.Code);
+            advised = true;
+
+            MxCommandReply unAdviseReply = await session.DispatchAsync(new StaCommand(
+                "session-1",
+                "live-unadvise",
+                new MxCommand
+                {
+                    Kind = MxCommandKind.UnAdvise,
+                    UnAdvise = new UnAdviseCommand
+                    {
+                        ServerHandle = serverHandle,
+                        ItemHandle = itemHandle,
+                    },
+                }));
+
+            Assert.Equal(ProtocolStatusCode.Ok, unAdviseReply.ProtocolStatus.Code);
+            advised = false;
+
+            MxCommandReply removeItemReply = await session.DispatchAsync(new StaCommand(
+                "session-1",
+                "live-advise-remove-item",
+                new MxCommand
+                {
+                    Kind = MxCommandKind.RemoveItem,
+                    RemoveItem = new RemoveItemCommand
+                    {
+                        ServerHandle = serverHandle,
+                        ItemHandle = itemHandle,
+                    },
+                }));
+
+            Assert.Equal(ProtocolStatusCode.Ok, removeItemReply.ProtocolStatus.Code);
+            itemHandle = 0;
+        }
+        finally
+        {
+            if (advised && itemHandle > 0)
+            {
+                await session.DispatchAsync(new StaCommand(
+                    "session-1",
+                    "live-unadvise-cleanup",
+                    new MxCommand
+                    {
+                        Kind = MxCommandKind.UnAdvise,
+                        UnAdvise = new UnAdviseCommand
+                        {
+                            ServerHandle = serverHandle,
+                            ItemHandle = itemHandle,
+                        },
+                    }));
+            }
+
+            if (itemHandle > 0)
+            {
+                await session.DispatchAsync(new StaCommand(
+                    "session-1",
+                    "live-advise-remove-item-cleanup",
+                    new MxCommand
+                    {
+                        Kind = MxCommandKind.RemoveItem,
+                        RemoveItem = new RemoveItemCommand
+                        {
+                            ServerHandle = serverHandle,
+                            ItemHandle = itemHandle,
+                        },
+                    }));
+            }
+
+            await UnregisterLiveSessionAsync(session, serverHandle, "live-advise-unregister");
+        }
+    }
+
     private static bool RunLiveMxAccessTests()
     {
         return string.Equals(

src/MxGateway.Worker.Tests/MxAccess/MxAccessStaSessionTests.cs

@@ -18,7 +18,7 @@ public sealed class MxAccessStaSessionTests
         using StaRuntime runtime = CreateRuntime();
         using MxAccessStaSession session = new(runtime, factory, eventSink);
 
-        WorkerReady ready = await session.StartAsync(workerProcessId: 1234);
+        WorkerReady ready = await session.StartAsync("session-1", workerProcessId: 1234);
 
         Assert.Equal(1234, ready.WorkerProcessId);
         Assert.Equal(MxAccessInteropInfo.ProgId, ready.MxaccessProgid);
@@ -28,6 +28,7 @@ public sealed class MxAccessStaSessionTests
         Assert.Equal(runtime.StaThreadId, eventSink.AttachThreadId);
         Assert.Equal(ApartmentState.STA, factory.CreateApartmentState);
         Assert.Same(factory.CreatedObject, eventSink.AttachedObject);
+        Assert.Equal("session-1", eventSink.SessionId);
     }
 
     [Fact]
@@ -107,10 +108,15 @@ public sealed class MxAccessStaSessionTests
 
         public int? DetachThreadId { get; private set; }
 
-        public void Attach(object mxAccessComObject)
+        public string? SessionId { get; private set; }
+
+        public void Attach(
+            object mxAccessComObject,
+            string sessionId)
         {
             AttachedObject = mxAccessComObject;
             AttachThreadId = Thread.CurrentThread.ManagedThreadId;
+            SessionId = sessionId;
         }
 
         public void Detach()

src/MxGateway.Worker/Ipc/WorkerPipeSession.cs

@@ -235,7 +235,7 @@ public sealed class WorkerPipeSession
         try
         {
             return await _mxAccessStaSession
-                .StartAsync(_processIdProvider(), cancellationToken)
+                .StartAsync(_options.SessionId, _processIdProvider(), cancellationToken)
                 .ConfigureAwait(false);
         }
         catch

src/MxGateway.Worker/MxAccess/IMxAccessEventSink.cs

@@ -2,7 +2,9 @@ namespace MxGateway.Worker.MxAccess;
 
 public interface IMxAccessEventSink
 {
-    void Attach(object mxAccessComObject);
+    void Attach(
+        object mxAccessComObject,
+        string sessionId);
 
     void Detach();
 }

src/MxGateway.Worker/MxAccess/IMxAccessServer.cs

@@ -18,4 +18,16 @@ public interface IMxAccessServer
     void RemoveItem(
         int serverHandle,
         int itemHandle);
+
+    void Advise(
+        int serverHandle,
+        int itemHandle);
+
+    void UnAdvise(
+        int serverHandle,
+        int itemHandle);
+
+    void AdviseSupervisory(
+        int serverHandle,
+        int itemHandle);
 }

src/MxGateway.Worker/MxAccess/MxAccessAdviceKind.cs

@@ -0,0 +1,7 @@
+namespace MxGateway.Worker.MxAccess;
+
+public enum MxAccessAdviceKind
+{
+    Plain = 1,
+    Supervisory = 2,
+}

src/MxGateway.Worker/MxAccess/MxAccessBaseEventSink.cs

@@ -1,13 +1,39 @@
+using System;
 using ArchestrA.MxAccess;
+using Proto = MxGateway.Contracts.Proto;
 
 namespace MxGateway.Worker.MxAccess;
 
 public sealed class MxAccessBaseEventSink : IMxAccessEventSink
 {
+    private readonly MxAccessEventMapper eventMapper;
+    private readonly MxAccessEventQueue eventQueue;
     private LMXProxyServerClass? server;
+    private string sessionId = string.Empty;
 
-    public void Attach(object mxAccessComObject)
+    public MxAccessBaseEventSink()
+        : this(new MxAccessEventQueue())
     {
+    }
+
+    public MxAccessBaseEventSink(MxAccessEventQueue eventQueue)
+        : this(eventQueue, new MxAccessEventMapper())
+    {
+    }
+
+    public MxAccessBaseEventSink(
+        MxAccessEventQueue eventQueue,
+        MxAccessEventMapper eventMapper)
+    {
+        this.eventQueue = eventQueue ?? throw new ArgumentNullException(nameof(eventQueue));
+        this.eventMapper = eventMapper ?? throw new ArgumentNullException(nameof(eventMapper));
+    }
+
+    public void Attach(
+        object mxAccessComObject,
+        string sessionId)
+    {
+        this.sessionId = sessionId ?? string.Empty;
         server = (LMXProxyServerClass)mxAccessComObject;
         server.OnDataChange += OnDataChange;
         server.OnWriteComplete += OnWriteComplete;
@@ -27,9 +53,10 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink
         server.OperationComplete -= OperationComplete;
         server.OnBufferedDataChange -= OnBufferedDataChange;
         server = null;
+        sessionId = string.Empty;
     }
 
-    private static void OnDataChange(
+    private void OnDataChange(
         int hLMXServerHandle,
         int phItemHandle,
         object pvItemValue,
@@ -37,23 +64,44 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink
         object pftItemTimeStamp,
         ref MXSTATUS_PROXY[] pVars)
     {
+        MXSTATUS_PROXY[] statuses = pVars;
+        EnqueueEvent(() => eventMapper.CreateOnDataChange(
+            sessionId,
+            hLMXServerHandle,
+            phItemHandle,
+            pvItemValue,
+            pwItemQuality,
+            pftItemTimeStamp,
+            statuses));
     }
 
-    private static void OnWriteComplete(
+    private void OnWriteComplete(
         int hLMXServerHandle,
         int phItemHandle,
         ref MXSTATUS_PROXY[] pVars)
     {
+        MXSTATUS_PROXY[] statuses = pVars;
+        EnqueueEvent(() => eventMapper.CreateOnWriteComplete(
+            sessionId,
+            hLMXServerHandle,
+            phItemHandle,
+            statuses));
     }
 
-    private static void OperationComplete(
+    private void OperationComplete(
         int hLMXServerHandle,
         int phItemHandle,
         ref MXSTATUS_PROXY[] pVars)
     {
+        MXSTATUS_PROXY[] statuses = pVars;
+        EnqueueEvent(() => eventMapper.CreateOperationComplete(
+            sessionId,
+            hLMXServerHandle,
+            phItemHandle,
+            statuses));
     }
 
-    private static void OnBufferedDataChange(
+    private void OnBufferedDataChange(
         int hLMXServerHandle,
         int phItemHandle,
         MxDataType dtDataType,
@@ -62,5 +110,42 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink
         object pftItemTimeStamp,
         ref MXSTATUS_PROXY[] pVars)
     {
+        MXSTATUS_PROXY[] statuses = pVars;
+        EnqueueEvent(() => eventMapper.CreateOnBufferedDataChange(
+            sessionId,
+            hLMXServerHandle,
+            phItemHandle,
+            (int)dtDataType,
+            pvItemValue,
+            pwItemQuality,
+            pftItemTimeStamp,
+            statuses));
+    }
+
+    private void EnqueueEvent(Func<Proto.MxEvent> createEvent)
+    {
+        try
+        {
+            eventQueue.Enqueue(createEvent());
+        }
+        catch (Exception exception)
+        {
+            eventQueue.RecordFault(CreateEventConversionFault(exception));
+        }
+    }
+
+    private Proto.WorkerFault CreateEventConversionFault(Exception exception)
+    {
+        return new Proto.WorkerFault
+        {
+            Category = Proto.WorkerFaultCategory.MxaccessEventConversionFailed,
+            ExceptionType = exception.GetType().FullName ?? string.Empty,
+            DiagnosticMessage = $"{exception.GetType().FullName}: HRESULT 0x{unchecked((uint)exception.HResult):X8}",
+            ProtocolStatus = new Proto.ProtocolStatus
+            {
+                Code = Proto.ProtocolStatusCode.MxaccessFailure,
+                Message = "MXAccess event conversion failed.",
+            },
+        };
     }
 }

src/MxGateway.Worker/MxAccess/MxAccessComServer.cs

@@ -73,6 +73,45 @@ public sealed class MxAccessComServer : IMxAccessServer
         Invoke(nameof(RemoveItem), serverHandle, itemHandle);
     }
 
+    public void Advise(
+        int serverHandle,
+        int itemHandle)
+    {
+        if (mxAccessComObject is ILMXProxyServer mxAccessServer)
+        {
+            mxAccessServer.Advise(serverHandle, itemHandle);
+            return;
+        }
+
+        Invoke(nameof(Advise), serverHandle, itemHandle);
+    }
+
+    public void UnAdvise(
+        int serverHandle,
+        int itemHandle)
+    {
+        if (mxAccessComObject is ILMXProxyServer mxAccessServer)
+        {
+            mxAccessServer.UnAdvise(serverHandle, itemHandle);
+            return;
+        }
+
+        Invoke(nameof(UnAdvise), serverHandle, itemHandle);
+    }
+
+    public void AdviseSupervisory(
+        int serverHandle,
+        int itemHandle)
+    {
+        if (mxAccessComObject is ILMXProxyServer4 mxAccessServer)
+        {
+            mxAccessServer.AdviseSupervisory(serverHandle, itemHandle);
+            return;
+        }
+
+        Invoke(nameof(AdviseSupervisory), serverHandle, itemHandle);
+    }
+
     private object Invoke(
         string methodName,
         params object[] arguments)

src/MxGateway.Worker/MxAccess/MxAccessCommandExecutor.cs

@@ -37,6 +37,9 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor
             MxCommandKind.AddItem => ExecuteAddItem(command),
             MxCommandKind.AddItem2 => ExecuteAddItem2(command),
             MxCommandKind.RemoveItem => ExecuteRemoveItem(command),
+            MxCommandKind.Advise => ExecuteAdvise(command),
+            MxCommandKind.UnAdvise => ExecuteUnAdvise(command),
+            MxCommandKind.AdviseSupervisory => ExecuteAdviseSupervisory(command),
             _ => CreateInvalidRequestReply(command, $"Unsupported MXAccess command kind {command.Kind}."),
         };
     }
@@ -130,6 +133,51 @@ public sealed class MxAccessCommandExecutor : IStaCommandExecutor
         return CreateOkReply(command);
     }
 
+    private MxCommandReply ExecuteAdvise(StaCommand command)
+    {
+        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.Advise)
+        {
+            return CreateInvalidRequestReply(command, "Advise command payload is required.");
+        }
+
+        AdviseCommand adviseCommand = command.Command.Advise;
+        session.Advise(
+            adviseCommand.ServerHandle,
+            adviseCommand.ItemHandle);
+
+        return CreateOkReply(command);
+    }
+
+    private MxCommandReply ExecuteUnAdvise(StaCommand command)
+    {
+        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.UnAdvise)
+        {
+            return CreateInvalidRequestReply(command, "UnAdvise command payload is required.");
+        }
+
+        UnAdviseCommand unAdviseCommand = command.Command.UnAdvise;
+        session.UnAdvise(
+            unAdviseCommand.ServerHandle,
+            unAdviseCommand.ItemHandle);
+
+        return CreateOkReply(command);
+    }
+
+    private MxCommandReply ExecuteAdviseSupervisory(StaCommand command)
+    {
+        if (command.Command.PayloadCase != MxCommand.PayloadOneofCase.AdviseSupervisory)
+        {
+            return CreateInvalidRequestReply(command, "AdviseSupervisory command payload is required.");
+        }
+
+        AdviseSupervisoryCommand adviseSupervisoryCommand = command.Command.AdviseSupervisory;
+        session.AdviseSupervisory(
+            adviseSupervisoryCommand.ServerHandle,
+            adviseSupervisoryCommand.ItemHandle);
+
+        return CreateOkReply(command);
+    }
+
     private static MxCommandReply CreateOkReply(StaCommand command)
     {
         return new MxCommandReply

src/MxGateway.Worker/MxAccess/MxAccessEventMapper.cs

@@ -0,0 +1,221 @@
+using System;
+using MxGateway.Contracts.Proto;
+using MxGateway.Worker.Conversion;
+
+namespace MxGateway.Worker.MxAccess;
+
+public sealed class MxAccessEventMapper
+{
+    private readonly VariantConverter variantConverter;
+    private readonly MxStatusProxyConverter statusProxyConverter;
+
+    public MxAccessEventMapper()
+        : this(new VariantConverter(), new MxStatusProxyConverter())
+    {
+    }
+
+    public MxAccessEventMapper(
+        VariantConverter variantConverter,
+        MxStatusProxyConverter statusProxyConverter)
+    {
+        this.variantConverter = variantConverter ?? throw new ArgumentNullException(nameof(variantConverter));
+        this.statusProxyConverter = statusProxyConverter ?? throw new ArgumentNullException(nameof(statusProxyConverter));
+    }
+
+    public MxEvent CreateOnDataChange(
+        string sessionId,
+        int serverHandle,
+        int itemHandle,
+        object? value,
+        int quality,
+        object? timestamp,
+        Array? statuses)
+    {
+        MxEvent mxEvent = CreateBaseEvent(
+            MxEventFamily.OnDataChange,
+            sessionId,
+            serverHandle,
+            itemHandle,
+            statuses);
+        mxEvent.Value = variantConverter.Convert(value);
+        mxEvent.Quality = quality;
+        ApplySourceTimestamp(mxEvent, timestamp);
+        mxEvent.OnDataChange = new OnDataChangeEvent();
+
+        return mxEvent;
+    }
+
+    public MxEvent CreateOnWriteComplete(
+        string sessionId,
+        int serverHandle,
+        int itemHandle,
+        Array? statuses)
+    {
+        MxEvent mxEvent = CreateBaseEvent(
+            MxEventFamily.OnWriteComplete,
+            sessionId,
+            serverHandle,
+            itemHandle,
+            statuses);
+        mxEvent.OnWriteComplete = new OnWriteCompleteEvent();
+
+        return mxEvent;
+    }
+
+    public MxEvent CreateOperationComplete(
+        string sessionId,
+        int serverHandle,
+        int itemHandle,
+        Array? statuses)
+    {
+        MxEvent mxEvent = CreateBaseEvent(
+            MxEventFamily.OperationComplete,
+            sessionId,
+            serverHandle,
+            itemHandle,
+            statuses);
+        mxEvent.OperationComplete = new OperationCompleteEvent();
+
+        return mxEvent;
+    }
+
+    public MxEvent CreateOnBufferedDataChange(
+        string sessionId,
+        int serverHandle,
+        int itemHandle,
+        int rawDataType,
+        object? value,
+        object? quality,
+        object? timestamp,
+        Array? statuses)
+    {
+        MxDataType dataType = MapMxDataType(rawDataType);
+        MxEvent mxEvent = CreateBaseEvent(
+            MxEventFamily.OnBufferedDataChange,
+            sessionId,
+            serverHandle,
+            itemHandle,
+            statuses);
+        mxEvent.Value = variantConverter.Convert(value, dataType);
+        mxEvent.OnBufferedDataChange = new OnBufferedDataChangeEvent
+        {
+            DataType = dataType,
+            RawDataType = rawDataType,
+            QualityValues = ConvertBufferedArray(quality, MxDataType.Integer),
+            TimestampValues = ConvertBufferedArray(timestamp, MxDataType.Time),
+        };
+
+        return mxEvent;
+    }
+
+    public static MxDataType MapMxDataType(int rawDataType)
+    {
+        return rawDataType switch
+        {
+            -1 => MxDataType.Unknown,
+            0 => MxDataType.NoData,
+            1 => MxDataType.Boolean,
+            2 => MxDataType.Integer,
+            3 => MxDataType.Float,
+            4 => MxDataType.Double,
+            5 => MxDataType.String,
+            6 => MxDataType.Time,
+            7 => MxDataType.ElapsedTime,
+            8 => MxDataType.ReferenceType,
+            9 => MxDataType.StatusType,
+            10 => MxDataType.Enum,
+            11 => MxDataType.SecurityClassificationEnum,
+            12 => MxDataType.DataQualityType,
+            13 => MxDataType.QualifiedEnum,
+            14 => MxDataType.QualifiedStruct,
+            15 => MxDataType.InternationalizedString,
+            16 => MxDataType.BigString,
+            17 => MxDataType.End,
+            _ => MxDataType.Unknown,
+        };
+    }
+
+    private MxEvent CreateBaseEvent(
+        MxEventFamily family,
+        string sessionId,
+        int serverHandle,
+        int itemHandle,
+        Array? statuses)
+    {
+        MxEvent mxEvent = new()
+        {
+            Family = family,
+            SessionId = sessionId ?? string.Empty,
+            ServerHandle = serverHandle,
+            ItemHandle = itemHandle,
+        };
+        mxEvent.Statuses.Add(statusProxyConverter.ConvertMany(statuses));
+
+        return mxEvent;
+    }
+
+    private void ApplySourceTimestamp(
+        MxEvent mxEvent,
+        object? timestamp)
+    {
+        MxValue convertedTimestamp = variantConverter.Convert(timestamp, MxDataType.Time);
+        if (convertedTimestamp.KindCase == MxValue.KindOneofCase.TimestampValue)
+        {
+            mxEvent.SourceTimestamp = convertedTimestamp.TimestampValue;
+            return;
+        }
+
+        if (!string.IsNullOrWhiteSpace(convertedTimestamp.RawDiagnostic))
+        {
+            mxEvent.RawStatus = string.IsNullOrWhiteSpace(mxEvent.RawStatus)
+                ? convertedTimestamp.RawDiagnostic
+                : $"{mxEvent.RawStatus}; {convertedTimestamp.RawDiagnostic}";
+        }
+    }
+
+    private MxArray ConvertBufferedArray(
+        object? value,
+        MxDataType expectedElementDataType)
+    {
+        if (value is Array array)
+        {
+            return variantConverter.ConvertArray(array, expectedElementDataType);
+        }
+
+        MxValue converted = variantConverter.Convert(value, expectedElementDataType);
+        if (converted.KindCase == MxValue.KindOneofCase.ArrayValue)
+        {
+            return converted.ArrayValue;
+        }
+
+        MxArray mxArray = new()
+        {
+            ElementDataType = converted.DataType,
+            VariantType = converted.VariantType,
+            RawElementDataType = converted.RawDataType,
+            RawDiagnostic = string.IsNullOrWhiteSpace(converted.RawDiagnostic)
+                ? "Buffered MXAccess event argument was not a SAFEARRAY."
+                : converted.RawDiagnostic,
+        };
+
+        switch (converted.KindCase)
+        {
+            case MxValue.KindOneofCase.Int32Value:
+                mxArray.Int32Values = new Int32Array();
+                mxArray.Int32Values.Values.Add(converted.Int32Value);
+                break;
+
+            case MxValue.KindOneofCase.Int64Value:
+                mxArray.Int64Values = new Int64Array();
+                mxArray.Int64Values.Values.Add(converted.Int64Value);
+                break;
+
+            case MxValue.KindOneofCase.TimestampValue:
+                mxArray.TimestampValues = new TimestampArray();
+                mxArray.TimestampValues.Values.Add(converted.TimestampValue);
+                break;
+        }
+
+        return mxArray;
+    }
+}

src/MxGateway.Worker/MxAccess/MxAccessEventQueue.cs

@@ -0,0 +1,180 @@
+using System;
+using System.Collections.Generic;
+using Google.Protobuf.WellKnownTypes;
+using MxGateway.Contracts.Proto;
+
+namespace MxGateway.Worker.MxAccess;
+
+public sealed class MxAccessEventQueue
+{
+    public const int DefaultCapacity = 10000;
+
+    private readonly int capacity;
+    private readonly Queue<WorkerEvent> events;
+    private readonly object syncRoot = new();
+    private ulong lastEventSequence;
+    private WorkerFault? fault;
+
+    public MxAccessEventQueue()
+        : this(DefaultCapacity)
+    {
+    }
+
+    public MxAccessEventQueue(int capacity)
+    {
+        if (capacity <= 0)
+        {
+            throw new ArgumentOutOfRangeException(
+                nameof(capacity),
+                "MXAccess event queue capacity must be greater than zero.");
+        }
+
+        this.capacity = capacity;
+        events = new Queue<WorkerEvent>(capacity);
+    }
+
+    public int Capacity => capacity;
+
+    public int Count
+    {
+        get
+        {
+            lock (syncRoot)
+            {
+                return events.Count;
+            }
+        }
+    }
+
+    public ulong LastEventSequence
+    {
+        get
+        {
+            lock (syncRoot)
+            {
+                return lastEventSequence;
+            }
+        }
+    }
+
+    public bool IsFaulted
+    {
+        get
+        {
+            lock (syncRoot)
+            {
+                return fault is not null;
+            }
+        }
+    }
+
+    public WorkerFault? Fault
+    {
+        get
+        {
+            lock (syncRoot)
+            {
+                return fault?.Clone();
+            }
+        }
+    }
+
+    public WorkerEvent Enqueue(MxEvent mxEvent)
+    {
+        if (mxEvent is null)
+        {
+            throw new ArgumentNullException(nameof(mxEvent));
+        }
+
+        lock (syncRoot)
+        {
+            if (fault is not null)
+            {
+                throw new InvalidOperationException("MXAccess outbound event queue is faulted.");
+            }
+
+            if (events.Count >= capacity)
+            {
+                fault = CreateOverflowFault();
+                throw new MxAccessEventQueueOverflowException(capacity);
+            }
+
+            MxEvent queuedEvent = mxEvent.Clone();
+            queuedEvent.WorkerSequence = ++lastEventSequence;
+            queuedEvent.WorkerTimestamp = Timestamp.FromDateTime(DateTime.UtcNow);
+
+            WorkerEvent workerEvent = new()
+            {
+                Event = queuedEvent,
+            };
+            events.Enqueue(workerEvent);
+
+            return workerEvent.Clone();
+        }
+    }
+
+    public bool TryDequeue(out WorkerEvent? workerEvent)
+    {
+        lock (syncRoot)
+        {
+            if (events.Count == 0)
+            {
+                workerEvent = null;
+                return false;
+            }
+
+            workerEvent = events.Dequeue().Clone();
+            return true;
+        }
+    }
+
+    public IReadOnlyList<WorkerEvent> Drain(uint maxEvents)
+    {
+        lock (syncRoot)
+        {
+            int drainCount = maxEvents == 0
+                ? events.Count
+                : Math.Min(events.Count, checked((int)Math.Min(maxEvents, int.MaxValue)));
+            if (drainCount == 0)
+            {
+                return Array.Empty<WorkerEvent>();
+            }
+
+            List<WorkerEvent> drained = new(drainCount);
+            for (int index = 0; index < drainCount; index++)
+            {
+                drained.Add(events.Dequeue().Clone());
+            }
+
+            return drained;
+        }
+    }
+
+    public void RecordFault(WorkerFault workerFault)
+    {
+        if (workerFault is null)
+        {
+            throw new ArgumentNullException(nameof(workerFault));
+        }
+
+        lock (syncRoot)
+        {
+            fault ??= workerFault.Clone();
+        }
+    }
+
+    private WorkerFault CreateOverflowFault()
+    {
+        string message = $"MXAccess outbound event queue reached capacity {capacity}.";
+        return new WorkerFault
+        {
+            Category = WorkerFaultCategory.QueueOverflow,
+            DiagnosticMessage = message,
+            ProtocolStatus = new ProtocolStatus
+            {
+                Code = ProtocolStatusCode.WorkerUnavailable,
+                Message = message,
+            },
+        };
+    }
+}

src/MxGateway.Worker/MxAccess/MxAccessEventQueueOverflowException.cs

@@ -0,0 +1,14 @@
+using System;
+
+namespace MxGateway.Worker.MxAccess;
+
+public sealed class MxAccessEventQueueOverflowException : Exception
+{
+    public MxAccessEventQueueOverflowException(int capacity)
+        : base($"MXAccess outbound event queue reached its configured capacity of {capacity}.")
+    {
+        Capacity = capacity;
+    }
+
+    public int Capacity { get; }
+}

src/MxGateway.Worker/MxAccess/MxAccessHandleRegistry.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Collections.Generic;
 using System.Linq;
 
@@ -7,6 +8,7 @@ public sealed class MxAccessHandleRegistry
 {
     private readonly Dictionary<int, RegisteredServerHandle> serverHandles = new();
     private readonly Dictionary<long, RegisteredItemHandle> itemHandles = new();
+    private readonly Dictionary<AdviceHandleKey, RegisteredAdviceHandle> adviceHandles = new();
 
     public IReadOnlyList<RegisteredServerHandle> ServerHandles => serverHandles
         .Values
@@ -19,6 +21,13 @@ public sealed class MxAccessHandleRegistry
         .ThenBy(handle => handle.ItemHandle)
         .ToArray();
 
+    public IReadOnlyList<RegisteredAdviceHandle> AdviceHandles => adviceHandles
+        .Values
+        .OrderBy(handle => handle.ServerHandle)
+        .ThenBy(handle => handle.ItemHandle)
+        .ThenBy(handle => handle.AdviceKind)
+        .ToArray();
+
     public void RegisterServerHandle(
         int serverHandle,
         string clientName)
@@ -37,6 +46,14 @@ public sealed class MxAccessHandleRegistry
         {
             itemHandles.Remove(key);
         }
+
+        foreach (AdviceHandleKey key in adviceHandles
+            .Where(pair => pair.Value.ServerHandle == serverHandle)
+            .Select(pair => pair.Key)
+            .ToArray())
+        {
+            adviceHandles.Remove(key);
+        }
     }
 
     public bool ContainsServerHandle(int serverHandle)
@@ -64,6 +81,7 @@ public sealed class MxAccessHandleRegistry
         int itemHandle)
     {
         itemHandles.Remove(CreateItemKey(serverHandle, itemHandle));
+        RemoveAdviceHandles(serverHandle, itemHandle);
     }
 
     public bool ContainsItemHandle(
@@ -73,10 +91,84 @@ public sealed class MxAccessHandleRegistry
         return itemHandles.ContainsKey(CreateItemKey(serverHandle, itemHandle));
     }
 
+    public void RegisterAdviceHandle(
+        int serverHandle,
+        int itemHandle,
+        MxAccessAdviceKind adviceKind)
+    {
+        AdviceHandleKey key = new(serverHandle, itemHandle, adviceKind);
+        adviceHandles[key] = new RegisteredAdviceHandle(
+            serverHandle,
+            itemHandle,
+            adviceKind);
+    }
+
+    public void RemoveAdviceHandles(
+        int serverHandle,
+        int itemHandle)
+    {
+        foreach (AdviceHandleKey key in adviceHandles
+            .Where(pair => pair.Value.ServerHandle == serverHandle && pair.Value.ItemHandle == itemHandle)
+            .Select(pair => pair.Key)
+            .ToArray())
+        {
+            adviceHandles.Remove(key);
+        }
+    }
+
+    public bool ContainsAdviceHandle(
+        int serverHandle,
+        int itemHandle,
+        MxAccessAdviceKind adviceKind)
+    {
+        return adviceHandles.ContainsKey(new AdviceHandleKey(serverHandle, itemHandle, adviceKind));
+    }
+
     private static long CreateItemKey(
         int serverHandle,
         int itemHandle)
     {
         return ((long)serverHandle << 32) | (uint)itemHandle;
     }
+
+    private readonly struct AdviceHandleKey : IEquatable<AdviceHandleKey>
+    {
+        private readonly int serverHandle;
+        private readonly int itemHandle;
+        private readonly MxAccessAdviceKind adviceKind;
+
+        public AdviceHandleKey(
+            int serverHandle,
+            int itemHandle,
+            MxAccessAdviceKind adviceKind)
+        {
+            this.serverHandle = serverHandle;
+            this.itemHandle = itemHandle;
+            this.adviceKind = adviceKind;
+        }
+
+        public bool Equals(AdviceHandleKey other)
+        {
+            return serverHandle == other.serverHandle
+                && itemHandle == other.itemHandle
+                && adviceKind == other.adviceKind;
+        }
+
+        public override bool Equals(object? obj)
+        {
+            return obj is AdviceHandleKey other && Equals(other);
+        }
+
+        public override int GetHashCode()
+        {
+            unchecked
+            {
+                int hashCode = serverHandle;
+                hashCode = (hashCode * 397) ^ itemHandle;
+                hashCode = (hashCode * 397) ^ (int)adviceKind;
+
+                return hashCode;
+            }
+        }
+    }
 }

src/MxGateway.Worker/MxAccess/MxAccessSession.cs

@@ -44,7 +44,8 @@ public sealed class MxAccessSession : IDisposable
 
     public static MxAccessSession Create(
         IMxAccessComObjectFactory factory,
-        IMxAccessEventSink eventSink)
+        IMxAccessEventSink eventSink,
+        string sessionId)
     {
         if (factory is null)
         {
@@ -66,7 +67,7 @@ public sealed class MxAccessSession : IDisposable
                 throw new InvalidOperationException("MXAccess COM factory returned null.");
             }
 
-            eventSink.Attach(mxAccessComObject);
+            eventSink.Attach(mxAccessComObject, sessionId);
 
             return new MxAccessSession(
                 mxAccessComObject,
@@ -151,6 +152,42 @@ public sealed class MxAccessSession : IDisposable
         handleRegistry.RemoveItemHandle(serverHandle, itemHandle);
     }
 
+    public void Advise(
+        int serverHandle,
+        int itemHandle)
+    {
+        ThrowIfDisposed();
+
+        mxAccessServer.Advise(serverHandle, itemHandle);
+        handleRegistry.RegisterAdviceHandle(
+            serverHandle,
+            itemHandle,
+            MxAccessAdviceKind.Plain);
+    }
+
+    public void UnAdvise(
+        int serverHandle,
+        int itemHandle)
+    {
+        ThrowIfDisposed();
+
+        mxAccessServer.UnAdvise(serverHandle, itemHandle);
+        handleRegistry.RemoveAdviceHandles(serverHandle, itemHandle);
+    }
+
+    public void AdviseSupervisory(
+        int serverHandle,
+        int itemHandle)
+    {
+        ThrowIfDisposed();
+
+        mxAccessServer.AdviseSupervisory(serverHandle, itemHandle);
+        handleRegistry.RegisterAdviceHandle(
+            serverHandle,
+            itemHandle,
+            MxAccessAdviceKind.Supervisory);
+    }
+
     public void Dispose()
     {
         if (disposed)

src/MxGateway.Worker/MxAccess/MxAccessStaSession.cs

@@ -11,6 +11,7 @@ public sealed class MxAccessStaSession : IDisposable
 {
     private readonly IMxAccessComObjectFactory factory;
     private readonly IMxAccessEventSink eventSink;
+    private readonly MxAccessEventQueue eventQueue;
     private readonly StaRuntime staRuntime;
     private StaCommandDispatcher? commandDispatcher;
     private MxAccessSession? session;
@@ -20,7 +21,7 @@ public sealed class MxAccessStaSession : IDisposable
         : this(
             new StaRuntime(),
             new MxAccessComObjectFactory(),
-            new MxAccessBaseEventSink())
+            new MxAccessEventQueue())
     {
     }
 
@@ -28,13 +29,41 @@ public sealed class MxAccessStaSession : IDisposable
         StaRuntime staRuntime,
         IMxAccessComObjectFactory factory,
         IMxAccessEventSink eventSink)
+        : this(staRuntime, factory, eventSink, new MxAccessEventQueue())
+    {
+    }
+
+    public MxAccessStaSession(
+        StaRuntime staRuntime,
+        IMxAccessComObjectFactory factory,
+        MxAccessEventQueue eventQueue)
+        : this(staRuntime, factory, new MxAccessBaseEventSink(eventQueue), eventQueue)
+    {
+    }
+
+    public MxAccessStaSession(
+        StaRuntime staRuntime,
+        IMxAccessComObjectFactory factory,
+        IMxAccessEventSink eventSink,
+        MxAccessEventQueue eventQueue)
     {
         this.staRuntime = staRuntime ?? throw new ArgumentNullException(nameof(staRuntime));
         this.factory = factory ?? throw new ArgumentNullException(nameof(factory));
         this.eventSink = eventSink ?? throw new ArgumentNullException(nameof(eventSink));
+        this.eventQueue = eventQueue ?? throw new ArgumentNullException(nameof(eventQueue));
+    }
+
+    public MxAccessEventQueue EventQueue => eventQueue;
+
+    public Task<WorkerReady> StartAsync(
+        int workerProcessId,
+        CancellationToken cancellationToken = default)
+    {
+        return StartAsync(string.Empty, workerProcessId, cancellationToken);
     }
 
     public Task<WorkerReady> StartAsync(
+        string sessionId,
         int workerProcessId,
         CancellationToken cancellationToken = default)
     {
@@ -48,7 +77,7 @@ public sealed class MxAccessStaSession : IDisposable
                     throw new InvalidOperationException("MXAccess COM session has already been created.");
                 }
 
-                session = MxAccessSession.Create(factory, eventSink);
+                session = MxAccessSession.Create(factory, eventSink, sessionId);
                 commandDispatcher = new StaCommandDispatcher(
                     staRuntime,
                     new MxAccessCommandExecutor(session));
@@ -68,6 +97,11 @@ public sealed class MxAccessStaSession : IDisposable
         return commandDispatcher.DispatchAsync(command);
     }
 
+    public IReadOnlyList<WorkerEvent> DrainEvents(uint maxEvents)
+    {
+        return eventQueue.Drain(maxEvents);
+    }
+
     public Task<IReadOnlyList<RegisteredServerHandle>> GetRegisteredServerHandlesAsync(
         CancellationToken cancellationToken = default)
     {
@@ -94,6 +128,19 @@ public sealed class MxAccessStaSession : IDisposable
             cancellationToken);
     }
 
+    public Task<IReadOnlyList<RegisteredAdviceHandle>> GetRegisteredAdviceHandlesAsync(
+        CancellationToken cancellationToken = default)
+    {
+        if (session is null)
+        {
+            throw new InvalidOperationException("MXAccess COM session has not been started.");
+        }
+
+        return staRuntime.InvokeAsync(
+            () => session.HandleRegistry.AdviceHandles,
+            cancellationToken);
+    }
+
     public void Dispose()
     {
         if (disposed)

src/MxGateway.Worker/MxAccess/RegisteredAdviceHandle.cs

@@ -0,0 +1,20 @@
+namespace MxGateway.Worker.MxAccess;
+
+public sealed class RegisteredAdviceHandle
+{
+    public RegisteredAdviceHandle(
+        int serverHandle,
+        int itemHandle,
+        MxAccessAdviceKind adviceKind)
+    {
+        ServerHandle = serverHandle;
+        ItemHandle = itemHandle;
+        AdviceKind = adviceKind;
+    }
+
+    public int ServerHandle { get; }
+
+    public int ItemHandle { get; }
+
+    public MxAccessAdviceKind AdviceKind { get; }
+}