From 963e2e60a5261b8e208f7359091cb8c7db84aca2 Mon Sep 17 00:00:00 2001 From: hby7921 <133756746+hby7921@users.noreply.github.com> Date: Sat, 15 Aug 2026 23:46:56 +0800 Subject: [PATCH] feat: add queued host-thread game action handler --- docs/engine-integration.md | 22 +- src/OpenGameAgent/QueuedGameActionHandler.cs | 345 ++++++++++++++++++ .../PublicApiCompatibilityTests.cs | 2 +- .../QueuedGameActionHandlerTests.cs | 195 ++++++++++ 4 files changed, 557 insertions(+), 7 deletions(-) create mode 100644 src/OpenGameAgent/QueuedGameActionHandler.cs create mode 100644 tests/OpenGameAgent.Tests/QueuedGameActionHandlerTests.cs diff --git a/docs/engine-integration.md b/docs/engine-integration.md index fef5804..e5c7980 100644 --- a/docs/engine-integration.md +++ b/docs/engine-integration.md @@ -82,14 +82,24 @@ Streaming `MessageUpdated` wire events carry only the new delta for that event. ## Main-thread actions -The adapters marshal public events, not arbitrary context providers or tool handlers. If a tool mutates a scene, create an `IGameActionHandler` that: +The adapters marshal public events, not arbitrary context providers or tool handlers. If a tool mutates a scene, use `QueuedGameActionHandler` when the host needs a reusable bounded handoff from background work to its game thread: -1. validates arguments without touching engine state; -2. queues a command onto the engine thread; -3. awaits a completion source; -4. returns the game-generated `GameActionReceipt`. +```csharp +var actionHandler = new QueuedGameActionHandler( + intent => + { + ValidateArguments(intent); + ValidateGeneration(intent.GenerationId); + return ExecuteOnGameThread(intent); + }, + intent => RecoverOnGameThread(intent), + capacity: 256); + +// Call this from the engine's tick/update callback. +actionHandler.Pump(maximumWorkItems: 32); +``` -Bound that queue and cancel waiting callers during scene or application shutdown. Keep the operation ID in the game save/command ledger so `RecoverAsync` can answer after a crash. +`ExecuteAsync` and `RecoverAsync` wait for the host thread without moving the game callback to a worker thread. Requests that have not started can be cancelled; a request already claimed by `Pump` is allowed to finish. Call `Stop` when a scene or application is shutting down so waiting requests fail and new requests are rejected. Keep the operation ID in the game save/command ledger so `RecoverAsync` can answer after a crash. The host remains responsible for authoritative rules and `generationId` validation. ## Versions diff --git a/src/OpenGameAgent/QueuedGameActionHandler.cs b/src/OpenGameAgent/QueuedGameActionHandler.cs new file mode 100644 index 0000000..68d5d00 --- /dev/null +++ b/src/OpenGameAgent/QueuedGameActionHandler.cs @@ -0,0 +1,345 @@ +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace OpenGameAgent; + +/// +/// Bridges background action requests to a host-owned thread. +/// The host must call from the thread that owns game state. +/// +public sealed class QueuedGameActionHandler : IGameActionHandler, IDisposable +{ + private const int DefaultCapacity = 256; + private const int MaximumCapacity = 100_000; + + private readonly object _gate = new(); + private readonly Queue _requests = new(); + private readonly Func _execute; + private readonly Func _recover; + private readonly int _capacity; + private int _queuedCount; + private int _stopped; + + public QueuedGameActionHandler( + Func execute, + Func recover, + int capacity = DefaultCapacity) + { + _execute = execute ?? throw new ArgumentNullException(nameof(execute)); + _recover = recover ?? throw new ArgumentNullException(nameof(recover)); + if (capacity <= 0 || capacity > MaximumCapacity) + { + throw new ArgumentOutOfRangeException(nameof(capacity)); + } + + _capacity = capacity; + } + + public int PendingCount + { + get + { + lock (_gate) + { + return _queuedCount; + } + } + } + + public bool IsStopped => Volatile.Read(ref _stopped) != 0; + + public async ValueTask ExecuteAsync( + GameActionIntent intent, + CancellationToken cancellationToken) + { + var request = Enqueue(intent, isRecovery: false, cancellationToken); + var receipt = await request.Completion.Task.ConfigureAwait(false); + return receipt ?? throw new InvalidOperationException("The execute callback returned a null receipt."); + } + + public async ValueTask RecoverAsync( + GameActionIntent intent, + CancellationToken cancellationToken) + { + var request = Enqueue(intent, isRecovery: true, cancellationToken); + return await request.Completion.Task.ConfigureAwait(false); + } + + /// + /// Executes up to non-cancelled requests on the caller's thread. + /// + public int Pump(int maximumWorkItems = 64) + { + if (maximumWorkItems <= 0) + { + throw new ArgumentOutOfRangeException(nameof(maximumWorkItems)); + } + + var executed = 0; + while (executed < maximumWorkItems) + { + PendingRequest? request; + lock (_gate) + { + if (_requests.Count == 0) + { + break; + } + + request = _requests.Dequeue(); + if (!request.TryStart()) + { + request.DisposeRegistration(); + request = null; + } + else + { + _queuedCount--; + } + } + + if (request is null) + { + continue; + } + + executed++; + request.Run(_execute, _recover); + } + + return executed; + } + + /// + /// Rejects new requests and fails requests that have not started. + /// A request already running inside is allowed to finish. + /// + public void Stop(Exception? reason = null) + { + var stoppedRequests = new List(); + lock (_gate) + { + if (Interlocked.Exchange(ref _stopped, 1) != 0) + { + return; + } + + while (_requests.Count > 0) + { + var request = _requests.Dequeue(); + if (request.TryStop()) + { + _queuedCount--; + stoppedRequests.Add(request); + } + } + } + + var failure = reason ?? new InvalidOperationException("The game action handler has stopped."); + foreach (var request in stoppedRequests) + { + request.Fail(failure); + } + } + + public void Dispose() => Stop(new ObjectDisposedException(nameof(QueuedGameActionHandler))); + + private PendingRequest Enqueue( + GameActionIntent intent, + bool isRecovery, + CancellationToken cancellationToken) + { + if (intent is null) + { + throw new ArgumentNullException(nameof(intent)); + } + + var request = new PendingRequest(intent, isRecovery, cancellationToken); + lock (_gate) + { + ThrowIfStopped(); + if (_queuedCount >= _capacity) + { + throw new GameRuntimeLimitException( + nameof(_capacity), + "The queued game action handler reached its capacity."); + } + + _requests.Enqueue(request); + _queuedCount++; + if (cancellationToken.CanBeCanceled) + { + request.AttachRegistration(cancellationToken.Register( + static state => + { + var cancellation = (CancellationRegistrationState)state!; + cancellation.Owner.Cancel(cancellation.Request); + }, + new CancellationRegistrationState(this, request))); + } + } + + return request; + } + + private void Cancel(PendingRequest request) + { + var cancelled = false; + lock (_gate) + { + if (request.TryCancel()) + { + _queuedCount--; + cancelled = true; + } + } + + if (cancelled) + { + request.CancelCompletion(); + request.DisposeRegistration(); + } + } + + private void ThrowIfStopped() + { + if (IsStopped) + { + throw new InvalidOperationException("The game action handler has stopped."); + } + } + + private sealed class CancellationRegistrationState + { + public CancellationRegistrationState(QueuedGameActionHandler owner, PendingRequest request) + { + Owner = owner; + Request = request; + } + + public QueuedGameActionHandler Owner { get; } + + public PendingRequest Request { get; } + } + + private sealed class PendingRequest + { + private readonly object _stateGate = new(); + private readonly bool _isRecovery; + private readonly CancellationToken _cancellationToken; + private RequestState _state; + private CancellationTokenRegistration _registration; + private bool _registrationAttached; + + public PendingRequest( + GameActionIntent intent, + bool isRecovery, + CancellationToken cancellationToken) + { + Intent = intent; + _isRecovery = isRecovery; + _cancellationToken = cancellationToken; + Completion = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + } + + public GameActionIntent Intent { get; } + + public TaskCompletionSource Completion { get; } + + public bool TryStart() => TryTransition(RequestState.Queued, RequestState.Started); + + public bool TryCancel() => TryTransition(RequestState.Queued, RequestState.Cancelled); + + public bool TryStop() => TryTransition(RequestState.Queued, RequestState.Stopped); + + public void AttachRegistration(CancellationTokenRegistration registration) + { + var dispose = false; + lock (_stateGate) + { + _registration = registration; + _registrationAttached = true; + dispose = _state != RequestState.Queued; + } + + if (dispose) + { + registration.Dispose(); + } + } + + public void DisposeRegistration() + { + CancellationTokenRegistration registration; + lock (_stateGate) + { + if (!_registrationAttached) + { + return; + } + + registration = _registration; + _registrationAttached = false; + } + + registration.Dispose(); + } + + public void CancelCompletion() => Completion.TrySetCanceled(_cancellationToken); + + public void Fail(Exception exception) + { + Completion.TrySetException(exception); + DisposeRegistration(); + } + + public void Run( + Func execute, + Func recover) + { + try + { + var receipt = _isRecovery ? recover(Intent) : execute(Intent); + if (!_isRecovery && receipt is null) + { + throw new InvalidOperationException("The execute callback returned a null receipt."); + } + + Completion.TrySetResult(receipt); + } + catch (Exception exception) + { + Completion.TrySetException(exception); + } + finally + { + DisposeRegistration(); + } + } + + private bool TryTransition(RequestState expected, RequestState next) + { + lock (_stateGate) + { + if (_state != expected) + { + return false; + } + + _state = next; + return true; + } + } + } + + private enum RequestState + { + Queued, + Started, + Cancelled, + Stopped, + } +} diff --git a/tests/OpenGameAgent.Tests/PublicApiCompatibilityTests.cs b/tests/OpenGameAgent.Tests/PublicApiCompatibilityTests.cs index 22a9a08..0b2158b 100644 --- a/tests/OpenGameAgent.Tests/PublicApiCompatibilityTests.cs +++ b/tests/OpenGameAgent.Tests/PublicApiCompatibilityTests.cs @@ -5,7 +5,7 @@ namespace OpenGameAgent.Tests; public sealed class PublicApiCompatibilityTests { - private const string ApprovedApiHash = "436EAE0B0E6367F7532232DD799B22DB6CA3C2A01EC0E9D4009EE9B157427AEB"; + private const string ApprovedApiHash = "BDD12E715A4D1F42B6FD33D93888AF379E2173F891F1179F26EF30773ABE5E5D"; [Fact] public void RuntimePublicApiMatchesTheApprovedStableSurface() diff --git a/tests/OpenGameAgent.Tests/QueuedGameActionHandlerTests.cs b/tests/OpenGameAgent.Tests/QueuedGameActionHandlerTests.cs new file mode 100644 index 0000000..a1300d8 --- /dev/null +++ b/tests/OpenGameAgent.Tests/QueuedGameActionHandlerTests.cs @@ -0,0 +1,195 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using OpenGameAgent; +using Xunit; + +namespace OpenGameAgent.Tests; + +public sealed class QueuedGameActionHandlerTests +{ + [Fact] + public async Task RequestsWaitForPumpAndRespectThePumpLimit() + { + var executed = new int[2]; + using var handler = new QueuedGameActionHandler( + intent => + { + executed[int.Parse(intent.OperationId, System.Globalization.CultureInfo.InvariantCulture) - 1]++; + return GameActionReceipt.Committed(intent, "{}"); + }, + _ => null, + capacity: 4); + + var first = handler.ExecuteAsync(Intent("1"), CancellationToken.None).AsTask(); + var second = handler.ExecuteAsync(Intent("2"), CancellationToken.None).AsTask(); + + Assert.False(first.IsCompleted); + Assert.Equal(2, handler.PendingCount); + Assert.Equal(1, handler.Pump(1)); + Assert.Equal("1", (await first).OperationId); + Assert.False(second.IsCompleted); + Assert.Equal(1, handler.PendingCount); + + Assert.Equal(1, handler.Pump()); + Assert.Equal("2", (await second).OperationId); + Assert.Equal(new[] { 1, 1 }, executed); + } + + [Fact] + public async Task QueueCapacityIsBoundedAndQueuedCancellationFreesCapacity() + { + using var handler = new QueuedGameActionHandler( + intent => GameActionReceipt.Committed(intent, "{}"), + _ => null, + capacity: 1); + using var cancellation = new CancellationTokenSource(); + + var cancelled = handler.ExecuteAsync(Intent("cancelled"), cancellation.Token).AsTask(); + cancellation.Cancel(); + await Assert.ThrowsAnyAsync(() => cancelled); + Assert.Equal(0, handler.PendingCount); + + var accepted = handler.ExecuteAsync(Intent("accepted"), CancellationToken.None).AsTask(); + await Assert.ThrowsAsync(async () => + await handler.ExecuteAsync(Intent("overflow"), CancellationToken.None).AsTask()); + + handler.Pump(); + await accepted; + } + + [Fact] + public async Task StartedRequestIsNotCancelledWhenItsCallerTimesOut() + { + using var cancellation = new CancellationTokenSource(); + using var handler = new QueuedGameActionHandler( + intent => + { + cancellation.Cancel(); + return GameActionReceipt.Committed(intent, "{}"); + }, + _ => null); + + var task = handler.ExecuteAsync(Intent("started"), cancellation.Token).AsTask(); + Assert.Equal(1, handler.Pump()); + + var receipt = await task; + Assert.Equal(GameActionStatus.Committed, receipt.Status); + } + + [Fact] + public async Task StopDoesNotCancelARequestAlreadyClaimedByPump() + { + QueuedGameActionHandler? handler = null; + handler = new QueuedGameActionHandler( + intent => + { + handler!.Stop(); + return GameActionReceipt.Committed(intent, "{}"); + }, + _ => null); + + var task = handler.ExecuteAsync(Intent("shutdown-race"), CancellationToken.None).AsTask(); + Assert.Equal(1, handler.Pump()); + + Assert.Equal(GameActionStatus.Committed, (await task).Status); + } + + [Fact] + public async Task RecoveryUsesTheSameHostThreadPump() + { + var pumpThread = Environment.CurrentManagedThreadId; + var recoveryThread = -1; + using var handler = new QueuedGameActionHandler( + intent => GameActionReceipt.Committed(intent, "{}"), + _ => + { + recoveryThread = Environment.CurrentManagedThreadId; + return null; + }); + + var task = handler.RecoverAsync(Intent("recover"), CancellationToken.None).AsTask(); + Assert.Equal(1, handler.Pump()); + + Assert.Null(await task); + Assert.Equal(pumpThread, recoveryThread); + } + + [Fact] + public async Task StopFailsPendingRequestsAndRejectsNewRequests() + { + using var handler = new QueuedGameActionHandler( + intent => GameActionReceipt.Committed(intent, "{}"), + _ => null); + var pending = handler.ExecuteAsync(Intent("pending"), CancellationToken.None).AsTask(); + + handler.Stop(); + + await Assert.ThrowsAsync(() => pending); + await Assert.ThrowsAsync(async () => + await handler.ExecuteAsync(Intent("after-stop"), CancellationToken.None).AsTask()); + Assert.True(handler.IsStopped); + Assert.Equal(0, handler.PendingCount); + } + + [Fact] + public async Task CallbackFailureDoesNotPreventLaterRequests() + { + using var handler = new QueuedGameActionHandler( + intent => + { + if (intent.OperationId == "fail") + { + throw new InvalidOperationException("test failure"); + } + + return GameActionReceipt.Committed(intent, "{}"); + }, + _ => null); + var failed = handler.ExecuteAsync(Intent("fail"), CancellationToken.None).AsTask(); + var completed = handler.ExecuteAsync(Intent("complete"), CancellationToken.None).AsTask(); + + Assert.Equal(2, handler.Pump(2)); + await Assert.ThrowsAsync(() => failed); + Assert.Equal(GameActionStatus.Committed, (await completed).Status); + } + + [Fact] + public async Task DurableDispatcherStillDeduplicatesQueuedOperations() + { + var executeCount = 0; + using var handler = new QueuedGameActionHandler( + intent => + { + executeCount++; + return GameActionReceipt.Committed(intent, "{}"); + }, + _ => null); + var dispatcher = new DurableGameActionDispatcher(new InMemoryGameActionJournal(), handler); + var intent = Intent("duplicate"); + + var first = dispatcher.ExecuteAsync(intent, CancellationToken.None).AsTask(); + var second = dispatcher.ExecuteAsync(intent, CancellationToken.None).AsTask(); + for (var attempt = 0; attempt < 100 && handler.PendingCount == 0; attempt++) + { + await Task.Yield(); + } + + Assert.Equal(1, handler.PendingCount); + handler.Pump(); + + Assert.Equal(GameActionStatus.Committed, (await first).Status); + Assert.Equal(GameActionStatus.Committed, (await second).Status); + Assert.Equal(1, executeCount); + } + + private static GameActionIntent Intent(string operationId) => + new( + operationId, + "input", + "session", + "actor", + "action", + "{}", + new GameMoment("world", 1)); +}