From 5716a9e7a31f112100152a5d4435fb58e1ff00ef Mon Sep 17 00:00:00 2001 From: Eric Sun <141227631+EricSun0218@users.noreply.github.com> Date: Thu, 13 Aug 2026 20:44:17 +0800 Subject: [PATCH] Add read-only mailbox pending queries --- CHANGELOG.md | 1 + README.md | 4 +- README.zh-CN.md | 4 +- docs/game-integration-patterns.md | 17 ++ .../FileGameMailbox.cs | 58 ++++++ src/OpenGameAgent/Mailbox.cs | 172 ++++++++++++++++++ .../PersistenceTests.cs | 137 ++++++++++++++ .../PublicApiCompatibilityTests.cs | 2 +- tests/OpenGameAgent.Tests/RuntimeTests.cs | 118 ++++++++++++ 9 files changed, 508 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f09976a..b2e2ec2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ - Add the optional `TaskPlanExtension` for session/actor-scoped persistent ordered checklists, revision-checked mutations, host-validated evidence, per-input advancement guards, pending-work routing, typed UI projection events, and bounded terminal retention. - Add typed, model-free host queries for persisted goals and task plans, including session revisions, and scope goal-change events with their session/actor key and input ID. +- Add batched, payload-free mailbox pending-status queries that distinguish ready work from active leases without claiming delivery or incrementing attempts. ## 0.3.0-alpha.2 diff --git a/README.md b/README.md index e84588c..5dc615a 100644 --- a/README.md +++ b/README.md @@ -67,7 +67,7 @@ OpenGameAgent keeps the reusable agent machinery independent from the game while - game-time memory filtering, expiry, and optional custom ranking; - optional local/remote embeddings, rebuildable vector indexes, and lexical/vector hybrid recall; - skills selected by input type and available tools; -- recurring game-time triggers and persistent actor mailboxes; +- recurring game-time triggers and persistent actor mailboxes with payload-free backlog queries; - a typed extension API for tools, skills, routes, workflows, hooks, events, and services; - capability-aware model catalogs and developer-hosted short-lived credentials; - lazy external-tool discovery and large-result artifact spill; @@ -110,7 +110,7 @@ Read [Architecture](docs/architecture.md) for the ownership and failure boundari | Game runtime | Arbitrary JSON input, game clocks/timelines, fast/full/workflow routing, optimistic sessions, duplicate-input protection, actor concurrency, active-run steering/abort | | Extension API | Immutable builder; prompt/context/tool/skill/route/workflow/hook/provider/service registration; typed lifecycle events and channels; namespaced persistent state | | Official extensions | Tool policy and search, structured player questions/recommended replies, goals, host-verified ordered task plans, memory, artifacts, knowledge, delegation, tracing, and durable parallel workflow graphs | -| World primitives | Durable actions, resumable workflows, memories, skills, signals, game-time schedules, actor mailboxes | +| World primitives | Durable actions, resumable workflows, memories, skills, signals, game-time schedules, actor mailboxes with batch read-only pending status | | Models and auth | Bundled capability/context/reasoning/cost directory, dynamic refresh, API-key/environment/stored/OAuth/local auth, developer-hosted short-lived credential gateway | | External tools | Lazy on-demand search/describe/call by default; explicit direct exposure for small trusted catalogs | | Portable plugins | [Agent Plugins 1.0.0](docs/agent-plugins.md) `plugin.json`, immediate-child `SKILL.md` discovery, MCP stdio/Streamable HTTP, client namespaces, containment, and component-level failure isolation | diff --git a/README.zh-CN.md b/README.zh-CN.md index 1b6a152..2732ba6 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -67,7 +67,7 @@ OpenGameAgent 不替游戏规定玩法，而是提供可复用的游戏坐标与 - 按游戏时间过滤、过期并可自定义排序的记忆； - 可选本地/远程嵌入、可重建向量索引与词法/向量混合召回； - 根据输入类型和可用工具选择的 Skills； -- 游戏时间触发器与持久邮箱； +- 游戏时间触发器，以及支持无 payload 积压查询的持久邮箱； - 可扩展工具、Skills、路由、Workflow、Hooks、事件与服务的类型化接口； - 能力感知模型目录与开发者托管的短期凭证； - 外部工具按需发现与大型结果产物化； @@ -108,7 +108,7 @@ GameAgentRuntime | 游戏 Runtime | 任意 JSON 输入、游戏时钟/时间线、快速/完整/Workflow 路由、乐观并发会话、输入去重、角色并发、运行中 steering/abort | | 扩展 API | 不可变构建器；提示词/上下文/工具/Skills/路由/Workflow/Hooks/提供方/服务注册；类型化生命周期事件与通道；命名空间持久状态 | | 官方扩展 | 工具策略与搜索、玩家结构化提问/推荐回复、目标、宿主证据校验的有序任务清单、记忆、产物、外部知识、委派、追踪和可持久并行工作流图 | -| 世界原语 | 可恢复动作、可续跑 Workflow、记忆、Skills、信号、游戏时间调度、角色邮箱 | +| 世界原语 | 可恢复动作、可续跑 Workflow、记忆、Skills、信号、游戏时间调度、支持批量只读待处理状态的角色邮箱 | | 模型与认证 | 内置模型能力/上下文/推理级别/成本目录、动态刷新、API Key/环境/存储/OAuth/本地认证、开发者托管短期凭证网关 | | 外部工具 | 默认按需搜索/描述/调用；小型可信目录可显式选择原生直连暴露 | | 可移植插件 | [Agent Plugins 1.0.0](docs/agent-plugins.md) `plugin.json`、直接子目录 `SKILL.md` 发现、MCP stdio/Streamable HTTP、客户端命名空间、路径限制与组件级故障隔离 | diff --git a/docs/game-integration-patterns.md b/docs/game-integration-patterns.md index a34c817..77721f3 100644 --- a/docs/game-integration-patterns.md +++ b/docs/game-integration-patterns.md @@ -30,6 +30,23 @@ game tick / month advance `MultiActorScheduler` gives per-actor ordering and global concurrency. `GameTimeScheduler` emits bounded recurring occurrences. `IGameMailbox` carries durable work to actors that are not currently resident. The game supplies activation, distance, importance, and budget policy. +When an AI budget ends exactly at a game-time boundary, inspect mailbox backlog without claiming work or invoking a model: + +```csharp +var recipients = activeActors + .Select(actorId => new GameMailboxRecipientKey(sessionId, actorId)) + .ToArray(); +var pending = await mailbox.GetPendingStatusAsync( + recipients, + DateTimeOffset.UtcNow, + cancellationToken); + +var mustPauseAtBoundary = pending.Any(status => status.IncompleteCount > 0); +var canRunImmediately = pending.Any(status => status.ReadyCount > 0); +``` + +`GetPendingStatusAsync` is a typed, read-only snapshot. It returns one result per requested key in input order, including zero counts for missing mailboxes, and never returns message payloads. `ReadyCount` includes unleased messages and messages whose operational lease has expired; `LeasedCount` contains incomplete messages whose operational lease is still active; `IncompleteCount` is their sum. Querying does not acquire a lease, increment `Attempt`, complete or abandon a message, or call the model. The built-in file store evaluates the whole recipient batch in one directory pass rather than scanning all mailbox files once per NPC. Supply the same trusted operational clock used for `ClaimAsync`. A concurrent claim or settlement may make any snapshot stale, so use it for scheduling and causal-boundary admission, not as authority to complete a specific message. + Use `GoalLoopExtension` when an actor owns semantic goals that can wait for a tick or event and continue later. `GoalLoopOptions.MaximumActiveGoals` bounds active and waiting work, while `MaximumRetainedTerminalGoals` independently retains only the most recent completed, failed, or cancelled records for audit. Terminal retention never removes active or waiting goals, so long-running sessions do not exhaust their future goal capacity. Use `AgentDelegationExtension` when one actor needs bounded background research or planning without sharing its mutable transcript. Delegates still receive explicitly scoped context and tools; delegation is not permission escalation. Delegation status can be persisted, but the included local executor runs child work in the current process and does not automatically resume an in-flight child after a process restart. Use a host-owned durable workflow or executor when child execution itself must survive restarts. The host can project goals and task plans after loading a save without invoking a model and without parsing extension-owned JSON keys: diff --git a/src/OpenGameAgent.Persistence/FileGameMailbox.cs b/src/OpenGameAgent.Persistence/FileGameMailbox.cs index 10a6753..180862f 100644 --- a/src/OpenGameAgent.Persistence/FileGameMailbox.cs +++ b/src/OpenGameAgent.Persistence/FileGameMailbox.cs @@ -203,6 +203,64 @@ public ValueTask CompleteAsync(string messageId, string leaseToken, Cancellation public ValueTask AbandonAsync(string messageId, string leaseToken, CancellationToken cancellationToken) => SettleAsync(messageId, leaseToken, complete: false, cancellationToken); + public async ValueTask> GetPendingStatusAsync( + IReadOnlyList recipients, + DateTimeOffset operationalNow, + CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + var requested = GameMailboxPendingQuery.Validate(recipients); + var counts = new Dictionary(); + foreach (var recipient in requested) + { + counts[recipient] = default; + } + + if (counts.Count == 0) + { + return Array.Empty(); + } + + // A batch is deliberately evaluated with one directory pass. Hosts can inspect + // many actors without turning recipient count into recipient count x file count. + foreach (var path in Directory.EnumerateFiles(_files.DirectoryPath, "*" + Suffix, SearchOption.TopDirectoryOnly) + .Take(_capacity)) + { + cancellationToken.ThrowIfCancellationRequested(); + var document = await _files.ReadAsync(path, cancellationToken).ConfigureAwait(false); + if (document is null) + { + continue; + } + + _files.EnsurePathFor(path, document.MessageId, Suffix, "mailbox message"); + var message = DecodeMessage(document); + if (document.Completed) + { + continue; + } + + var recipient = new GameMailboxRecipientKey(message.SessionId, message.RecipientId); + if (!counts.TryGetValue(recipient, out var count)) + { + continue; + } + + if (document.LeaseToken is not null && document.OperationalLeaseExpiresAt > operationalNow) + { + count.Leased = checked(count.Leased + 1); + } + else + { + count.Ready = checked(count.Ready + 1); + } + + counts[recipient] = count; + } + + return GameMailboxPendingQuery.Materialize(requested, counts); + } + private async ValueTask SettleAsync( string messageId, string leaseToken, diff --git a/src/OpenGameAgent/Mailbox.cs b/src/OpenGameAgent/Mailbox.cs index 1ec83d8..df5107a 100644 --- a/src/OpenGameAgent/Mailbox.cs +++ b/src/OpenGameAgent/Mailbox.cs @@ -68,6 +68,74 @@ public GameMailboxDelivery( public DateTimeOffset OperationalLeaseExpiresAt { get; } } +public readonly struct GameMailboxRecipientKey : IEquatable +{ + public GameMailboxRecipientKey(string sessionId, string recipientId) + { + SessionId = GameJson.RequireId(sessionId, nameof(sessionId)); + RecipientId = GameJson.RequireId(recipientId, nameof(recipientId)); + } + + public string SessionId { get; } + + public string RecipientId { get; } + + public bool Equals(GameMailboxRecipientKey other) => + string.Equals(SessionId, other.SessionId, StringComparison.Ordinal) + && string.Equals(RecipientId, other.RecipientId, StringComparison.Ordinal); + + public override bool Equals(object? obj) => obj is GameMailboxRecipientKey other && Equals(other); + + public override int GetHashCode() + { + unchecked + { + return ((SessionId is null ? 0 : StringComparer.Ordinal.GetHashCode(SessionId)) * 397) + ^ (RecipientId is null ? 0 : StringComparer.Ordinal.GetHashCode(RecipientId)); + } + } + + public override string ToString() => (SessionId ?? string.Empty) + ":" + (RecipientId ?? string.Empty); + + public static bool operator ==(GameMailboxRecipientKey left, GameMailboxRecipientKey right) => + left.Equals(right); + + public static bool operator !=(GameMailboxRecipientKey left, GameMailboxRecipientKey right) => + !left.Equals(right); + + internal GameMailboxRecipientKey EnsureValid(string parameterName) + { + if (string.IsNullOrWhiteSpace(SessionId) || string.IsNullOrWhiteSpace(RecipientId)) + { + throw new ArgumentException("A valid mailbox recipient key is required.", parameterName); + } + + return this; + } +} + +public sealed class GameMailboxPendingStatus +{ + public GameMailboxPendingStatus( + GameMailboxRecipientKey recipient, + int readyCount, + int leasedCount) + { + Recipient = recipient.EnsureValid(nameof(recipient)); + ReadyCount = readyCount >= 0 ? readyCount : throw new ArgumentOutOfRangeException(nameof(readyCount)); + LeasedCount = leasedCount >= 0 ? leasedCount : throw new ArgumentOutOfRangeException(nameof(leasedCount)); + IncompleteCount = checked(readyCount + leasedCount); + } + + public GameMailboxRecipientKey Recipient { get; } + + public int ReadyCount { get; } + + public int LeasedCount { get; } + + public int IncompleteCount { get; } +} + public interface IGameMailbox { ValueTask EnqueueAsync(GameMailboxMessage message, CancellationToken cancellationToken); @@ -83,6 +151,11 @@ ValueTask> ClaimAsync( ValueTask CompleteAsync(string messageId, string leaseToken, CancellationToken cancellationToken); ValueTask AbandonAsync(string messageId, string leaseToken, CancellationToken cancellationToken); + + ValueTask> GetPendingStatusAsync( + IReadOnlyList recipients, + DateTimeOffset operationalNow, + CancellationToken cancellationToken); } public sealed class InMemoryGameMailbox : IGameMailbox @@ -214,6 +287,54 @@ public ValueTask AbandonAsync(string messageId, string leaseToken, CancellationT return default; } + public ValueTask> GetPendingStatusAsync( + IReadOnlyList recipients, + DateTimeOffset operationalNow, + CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + var requested = GameMailboxPendingQuery.Validate(recipients); + var counts = new Dictionary(); + foreach (var recipient in requested) + { + counts[recipient] = default; + } + + lock (_gate) + { + foreach (var entry in _entries.Values) + { + cancellationToken.ThrowIfCancellationRequested(); + if (entry.Completed) + { + continue; + } + + var recipient = new GameMailboxRecipientKey( + entry.Message.SessionId, + entry.Message.RecipientId); + if (!counts.TryGetValue(recipient, out var count)) + { + continue; + } + + if (entry.LeaseToken is not null && entry.LeaseExpiresAt > operationalNow) + { + count.Leased = checked(count.Leased + 1); + } + else + { + count.Ready = checked(count.Ready + 1); + } + + counts[recipient] = count; + } + } + + return new ValueTask>( + GameMailboxPendingQuery.Materialize(requested, counts)); + } + private Entry RequireLease(string messageId, string leaseToken) { GameJson.RequireId(messageId, nameof(messageId)); @@ -272,3 +393,54 @@ public Entry(GameMailboxMessage message, long sequence) public bool Completed { get; set; } } } + +internal static class GameMailboxPendingQuery +{ + internal const int MaximumRecipients = 4_096; + + internal static GameMailboxRecipientKey[] Validate( + IReadOnlyList recipients) + { + if (recipients is null) + { + throw new ArgumentNullException(nameof(recipients)); + } + + if (recipients.Count > MaximumRecipients) + { + throw new GameRuntimeLimitException( + nameof(MaximumRecipients), + "A mailbox pending query contains too many recipients."); + } + + var copy = new GameMailboxRecipientKey[recipients.Count]; + for (var index = 0; index < recipients.Count; index++) + { + copy[index] = recipients[index].EnsureValid(nameof(recipients)); + } + + return copy; + } + + internal static IReadOnlyList Materialize( + IReadOnlyList requested, + IReadOnlyDictionary counts) + { + var result = new GameMailboxPendingStatus[requested.Count]; + for (var index = 0; index < requested.Count; index++) + { + var recipient = requested[index]; + var count = counts[recipient]; + result[index] = new GameMailboxPendingStatus(recipient, count.Ready, count.Leased); + } + + return Array.AsReadOnly(result); + } + + internal struct Counts + { + public int Ready; + + public int Leased; + } +} diff --git a/tests/OpenGameAgent.Persistence.Tests/PersistenceTests.cs b/tests/OpenGameAgent.Persistence.Tests/PersistenceTests.cs index d4ee644..58b80d9 100644 --- a/tests/OpenGameAgent.Persistence.Tests/PersistenceTests.cs +++ b/tests/OpenGameAgent.Persistence.Tests/PersistenceTests.cs @@ -1054,6 +1054,143 @@ await finalRestart.CompleteAsync( TestContext.Current.CancellationToken)); } + [Fact] + public async Task FileMailboxPendingStatusIsReadOnlyAndSurvivesRestart() + { + using var directory = new TemporaryDirectory(); + var mailbox = new FileGameMailbox(directory.Path); + var recipient = new GameMailboxRecipientKey("session", "npc"); + var other = new GameMailboxRecipientKey("session", "other"); + var missing = new GameMailboxRecipientKey("missing", "npc"); + var now = DateTimeOffset.Parse("2026-01-01T00:00:00Z"); + await mailbox.EnqueueAsync( + new GameMailboxMessage("mail-a", "session", "npc", "event", "{\"secret\":1}", new GameMoment("world", 1)), + TestContext.Current.CancellationToken); + await mailbox.EnqueueAsync( + new GameMailboxMessage("mail-b", "session", "npc", "event", "{\"secret\":2}", new GameMoment("world", 2)), + TestContext.Current.CancellationToken); + await mailbox.EnqueueAsync( + new GameMailboxMessage("mail-c", "session", "other", "event", "{}", new GameMoment("world", 3)), + TestContext.Current.CancellationToken); + + var requested = new[] { recipient, other, missing }; + var initial = await mailbox.GetPendingStatusAsync( + requested, + now, + TestContext.Current.CancellationToken); + Assert.Equal(2, initial[0].ReadyCount); + Assert.Equal(1, initial[1].ReadyCount); + Assert.Equal(0, initial[2].IncompleteCount); + + var delivery = Assert.Single(await mailbox.ClaimAsync( + recipient.SessionId, + recipient.RecipientId, + 1, + now, + TimeSpan.FromMinutes(1), + TestContext.Current.CancellationToken)); + Assert.Equal(1, delivery.Attempt); + + var restarted = new FileGameMailbox(directory.Path); + var activeLease = await restarted.GetPendingStatusAsync( + requested, + now.AddSeconds(30), + TestContext.Current.CancellationToken); + Assert.Equal(1, activeLease[0].ReadyCount); + Assert.Equal(1, activeLease[0].LeasedCount); + Assert.Equal(2, activeLease[0].IncompleteCount); + Assert.Equal(1, activeLease[1].ReadyCount); + + await restarted.AbandonAsync( + delivery.Message.MessageId, + delivery.LeaseToken, + TestContext.Current.CancellationToken); + var abandoned = Assert.Single(await restarted.GetPendingStatusAsync( + new[] { recipient }, + now, + TestContext.Current.CancellationToken)); + Assert.Equal(2, abandoned.ReadyCount); + Assert.Equal(0, abandoned.LeasedCount); + + var claimed = await restarted.ClaimAsync( + recipient.SessionId, + recipient.RecipientId, + 2, + now, + TimeSpan.FromMinutes(1), + TestContext.Current.CancellationToken); + Assert.Equal(2, claimed.Count); + Assert.Contains(claimed, item => item.Message.MessageId == delivery.Message.MessageId && item.Attempt == 2); + Assert.Contains(claimed, item => item.Message.MessageId != delivery.Message.MessageId && item.Attempt == 1); + foreach (var item in claimed) + { + await restarted.CompleteAsync( + item.Message.MessageId, + item.LeaseToken, + TestContext.Current.CancellationToken); + } + + var completed = Assert.Single(await new FileGameMailbox(directory.Path).GetPendingStatusAsync( + new[] { recipient }, + now, + TestContext.Current.CancellationToken)); + Assert.Equal(0, completed.IncompleteCount); + } + + [Fact] + public async Task FileMailboxPendingBatchReturnsEveryRequestedRecipient() + { + using var directory = new TemporaryDirectory(); + var mailbox = new FileGameMailbox(directory.Path, capacity: 512); + var recipients = Enumerable.Range(0, 128) + .Select(index => new GameMailboxRecipientKey("session", "npc-" + index)) + .ToArray(); + for (var index = 0; index < recipients.Length; index++) + { + await mailbox.EnqueueAsync( + new GameMailboxMessage( + "mail-" + index, + recipients[index].SessionId, + recipients[index].RecipientId, + "event", + "{}", + new GameMoment("world", index)), + TestContext.Current.CancellationToken); + } + + var statuses = await mailbox.GetPendingStatusAsync( + recipients, + DateTimeOffset.UnixEpoch, + TestContext.Current.CancellationToken); + + Assert.Equal(recipients.Length, statuses.Count); + Assert.All(statuses, status => + { + Assert.Equal(1, status.ReadyCount); + Assert.Equal(0, status.LeasedCount); + Assert.Equal(1, status.IncompleteCount); + }); + Assert.Equal(recipients, statuses.Select(status => status.Recipient)); + } + + [Fact] + public async Task FileMailboxPendingStatusRejectsCorruptState() + { + using var directory = new TemporaryDirectory(); + var mailbox = new FileGameMailbox(directory.Path); + await mailbox.EnqueueAsync( + new GameMailboxMessage("mail", "session", "npc", "event", "{}", new GameMoment("world", 1)), + TestContext.Current.CancellationToken); + var path = Assert.Single(Directory.GetFiles(directory.Path, "*.mailbox.json")); + await File.WriteAllTextAsync(path, "{\"formatVersion\":1", TestContext.Current.CancellationToken); + + await Assert.ThrowsAsync(async () => + await mailbox.GetPendingStatusAsync( + new[] { new GameMailboxRecipientKey("session", "npc") }, + DateTimeOffset.UnixEpoch, + TestContext.Current.CancellationToken)); + } + [Fact] public async Task IndependentMailboxWorkersCannotClaimTheSameMessageLease() { diff --git a/tests/OpenGameAgent.Tests/PublicApiCompatibilityTests.cs b/tests/OpenGameAgent.Tests/PublicApiCompatibilityTests.cs index 04f57b6..03f14e1 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 = "2669F795459C54443CF3807963CAACB76AB5986709F1D3D2D46275E0CF4E6AD9"; + private const string ApprovedApiHash = "D57DE3663DE8EA3EAB266C4A586B2A10C9E03676D554F4E70F5C31F2F7E9C19D"; [Fact] public void RuntimePublicApiMatchesTheApprovedStableSurface() diff --git a/tests/OpenGameAgent.Tests/RuntimeTests.cs b/tests/OpenGameAgent.Tests/RuntimeTests.cs index f696b21..aab6cb4 100644 --- a/tests/OpenGameAgent.Tests/RuntimeTests.cs +++ b/tests/OpenGameAgent.Tests/RuntimeTests.cs @@ -3094,6 +3094,124 @@ await mailbox.ClaimAsync( Assert.Equal(1, delivery.Attempt); } + [Fact] + public async Task MailboxPendingStatusIsReadOnlyAndTracksLeaseLifecycle() + { + var mailbox = new InMemoryGameMailbox(); + var recipient = new GameMailboxRecipientKey("session", "npc"); + var missing = new GameMailboxRecipientKey("session", "missing"); + var now = DateTimeOffset.Parse("2026-01-01T00:00:00Z"); + await mailbox.EnqueueAsync( + new GameMailboxMessage( + "mail", + recipient.SessionId, + recipient.RecipientId, + "event", + "{\"private\":\"not returned\"}", + new GameMoment("world", 1)), + TestContext.Current.CancellationToken); + + var requested = new[] { recipient, missing, recipient }; + var initial = await mailbox.GetPendingStatusAsync( + requested, + now, + TestContext.Current.CancellationToken); + var repeated = await mailbox.GetPendingStatusAsync( + requested, + now, + TestContext.Current.CancellationToken); + + Assert.Equal(3, initial.Count); + Assert.Equal(1, initial[0].ReadyCount); + Assert.Equal(0, initial[0].LeasedCount); + Assert.Equal(1, initial[0].IncompleteCount); + Assert.Equal(0, initial[1].IncompleteCount); + Assert.Equal(initial[0].IncompleteCount, initial[2].IncompleteCount); + Assert.Equal(initial.Select(StatusTuple), repeated.Select(StatusTuple)); + + var delivery = Assert.Single(await mailbox.ClaimAsync( + recipient.SessionId, + recipient.RecipientId, + 1, + now, + TimeSpan.FromMinutes(1), + TestContext.Current.CancellationToken)); + Assert.Equal(1, delivery.Attempt); + + var leased = Assert.Single(await mailbox.GetPendingStatusAsync( + new[] { recipient }, + now.AddSeconds(30), + TestContext.Current.CancellationToken)); + Assert.Equal(0, leased.ReadyCount); + Assert.Equal(1, leased.LeasedCount); + Assert.Equal(1, leased.IncompleteCount); + + var expired = Assert.Single(await mailbox.GetPendingStatusAsync( + new[] { recipient }, + now.AddMinutes(2), + TestContext.Current.CancellationToken)); + Assert.Equal(1, expired.ReadyCount); + Assert.Equal(0, expired.LeasedCount); + + await mailbox.AbandonAsync( + delivery.Message.MessageId, + delivery.LeaseToken, + TestContext.Current.CancellationToken); + var abandoned = Assert.Single(await mailbox.GetPendingStatusAsync( + new[] { recipient }, + now, + TestContext.Current.CancellationToken)); + Assert.Equal(1, abandoned.ReadyCount); + Assert.Equal(0, abandoned.LeasedCount); + + var reclaimed = Assert.Single(await mailbox.ClaimAsync( + recipient.SessionId, + recipient.RecipientId, + 1, + now, + TimeSpan.FromMinutes(1), + TestContext.Current.CancellationToken)); + Assert.Equal(2, reclaimed.Attempt); + await mailbox.CompleteAsync( + reclaimed.Message.MessageId, + reclaimed.LeaseToken, + TestContext.Current.CancellationToken); + + var completed = Assert.Single(await mailbox.GetPendingStatusAsync( + new[] { recipient }, + now, + TestContext.Current.CancellationToken)); + Assert.Equal(0, completed.IncompleteCount); + + static (GameMailboxRecipientKey Recipient, int Ready, int Leased, int Incomplete) StatusTuple( + GameMailboxPendingStatus status) => + (status.Recipient, status.ReadyCount, status.LeasedCount, status.IncompleteCount); + } + + [Fact] + public async Task MailboxPendingStatusValidatesBoundsAndCancellation() + { + var mailbox = new InMemoryGameMailbox(); + var recipients = Enumerable.Range(0, 4_097) + .Select(index => new GameMailboxRecipientKey("session", "npc-" + index)) + .ToArray(); + + var limit = await Assert.ThrowsAsync(async () => + await mailbox.GetPendingStatusAsync( + recipients, + DateTimeOffset.UnixEpoch, + TestContext.Current.CancellationToken)); + Assert.Equal("MaximumRecipients", limit.Limit); + + using var cancellation = new CancellationTokenSource(); + cancellation.Cancel(); + await Assert.ThrowsAnyAsync(async () => + await mailbox.GetPendingStatusAsync( + new[] { new GameMailboxRecipientKey("session", "npc") }, + DateTimeOffset.UnixEpoch, + cancellation.Token)); + } + [Fact] public void SchedulerCatchesUpAnOverdueTriggerExactlyOnce() {