From 7efee36345a6031edcf4f9ac98dd741eb3ebdc46 Mon Sep 17 00:00:00 2001 From: Eric Sun <141227631+EricSun0218@users.noreply.github.com> Date: Thu, 13 Aug 2026 21:15:48 +0800 Subject: [PATCH 1/2] Add durable task plan pause and resume --- CHANGELOG.md | 1 + README.md | 2 +- README.zh-CN.md | 2 +- docs/features.md | 2 +- docs/game-integration-patterns.md | 8 +- .../TaskPlanExtension.cs | 68 ++++- .../TaskPlanExtensionTests.cs | 278 ++++++++++++++++++ .../TaskPlanPersistenceTests.cs | 65 ++++ 8 files changed, 407 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b2e2ec2..a7ad0b4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,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. +- Add backward-compatible durable task-plan pause/resume with revision checks, preserved in-progress steps, non-runnable paused routing, typed change reasons, and restart coverage. ## 0.3.0-alpha.2 diff --git a/README.md b/README.md index 5dc615a..748a088 100644 --- a/README.md +++ b/README.md @@ -109,7 +109,7 @@ Read [Architecture](docs/architecture.md) for the ownership and failure boundari | Tool execution | Provider-request schema preflight plus execution-time validation over a bounded JSON Schema subset, guaranteed result for every accepted call, safe parallel reads, conflict-key serialization, policy blocking/termination, timeouts, uncertain write outcomes | | 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 | +| Official extensions | Tool policy and search, structured player questions/recommended replies, goals, host-verified ordered task plans with durable pause/resume, memory, artifacts, knowledge, delegation, tracing, and durable parallel workflow graphs | | 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 | diff --git a/README.zh-CN.md b/README.zh-CN.md index 2732ba6..9997ad9 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -107,7 +107,7 @@ GameAgentRuntime | 工具执行 | provider 请求前 schema 预检及执行期有界 JSON Schema 子集校验、每个已接受调用都有结果、安全并行读、冲突键串行、策略拦截/终止、超时与写入结果未知语义 | | 游戏 Runtime | 任意 JSON 输入、游戏时钟/时间线、快速/完整/Workflow 路由、乐观并发会话、输入去重、角色并发、运行中 steering/abort | | 扩展 API | 不可变构建器；提示词/上下文/工具/Skills/路由/Workflow/Hooks/提供方/服务注册；类型化生命周期事件与通道；命名空间持久状态 | -| 官方扩展 | 工具策略与搜索、玩家结构化提问/推荐回复、目标、宿主证据校验的有序任务清单、记忆、产物、外部知识、委派、追踪和可持久并行工作流图 | +| 官方扩展 | 工具策略与搜索、玩家结构化提问/推荐回复、目标、支持持久暂停/恢复且由宿主校验证据的有序任务清单、记忆、产物、外部知识、委派、追踪和可持久并行工作流图 | | 世界原语 | 可恢复动作、可续跑 Workflow、记忆、Skills、信号、游戏时间调度、支持批量只读待处理状态的角色邮箱 | | 模型与认证 | 内置模型能力/上下文/推理级别/成本目录、动态刷新、API Key/环境/存储/OAuth/本地认证、开发者托管短期凭证网关 | | 外部工具 | 默认按需搜索/描述/调用；小型可信目录可显式选择原生直连暴露 | diff --git a/docs/features.md b/docs/features.md index 245578c..c42e2c8 100644 --- a/docs/features.md +++ b/docs/features.md @@ -33,7 +33,7 @@ This page maps product needs to the smallest reusable OpenGameAgent primitive. | Search a large tool catalog on demand | `ToolCatalogExtension`, `IGameToolCatalog` | | Ask the player structured questions and recommend choices | `StructuredInteractionExtension`, `IGameInteractionBroker` | | Track goals and resume them after game-time waits | `GoalLoopExtension` | -| Keep a persistent ordered checklist with host-verified progress | `TaskPlanExtension` | +| Keep a persistent ordered checklist with host-verified progress and durable pause/resume | `TaskPlanExtension` | | Delegate bounded foreground or background work | `AgentDelegationExtension` | | Query a game-owned knowledge source | `ExternalKnowledgeExtension` | | Capture bounded lifecycle traces | `GameAgentTracingExtension` | diff --git a/docs/game-integration-patterns.md b/docs/game-integration-patterns.md index 77721f3..ade47d0 100644 --- a/docs/game-integration-patterns.md +++ b/docs/game-integration-patterns.md @@ -68,7 +68,7 @@ ui.Render(goals.SessionRevision, goals.Goals, taskPlans.Plans); These readers are read-only projections over `IGameSessionStore`. They do not run routing, providers, tools, pruning, or other extension lifecycle work. A missing session returns revision `0` and an empty collection. The caller must authorize the `GameSessionKey` before querying it; the readers deliberately do not replace host ownership policy. -Use `TaskPlanExtension` for an ordered checklist that must survive later inputs. It is separate from `GoalLoopExtension`: goals describe durable intent and game-time waits, while a task plan records an ordered execution path. An active plan always has one `InProgress` step, a completed prefix, and a pending suffix. The model cannot advance a step merely by claiming success; the host-supplied `GameTaskPlanEvidenceValidator` must accept the evidence against the current input, plan, and step. +Use `TaskPlanExtension` for an ordered checklist that must survive later inputs. It is separate from `GoalLoopExtension`: goals describe durable intent and game-time waits, while a task plan records an ordered execution path. An active or paused plan always retains one `InProgress` step, a completed prefix, and a pending suffix. The model cannot advance a step merely by claiming success; the host-supplied `GameTaskPlanEvidenceValidator` must accept the evidence against the current input, plan, and step. ```csharp var plans = new TaskPlanExtension( @@ -96,7 +96,9 @@ var runtime = new GameAgentBuilder(provider, model) .Build(); ``` -`advance` requires the plan revision and accepted evidence and can succeed only once per input. `replace_remaining` preserves completed steps and replaces only unfinished work. `fail` and `cancel` are terminal. Active plans contribute pending work to routing; terminal retention is independently bounded and never consumes active-plan capacity. State is namespaced by the runtime's session/actor key and persists through any `IGameSessionStore`. +`advance` requires the plan revision and accepted evidence and can succeed only once per input. `replace_remaining` preserves completed steps and replaces only unfinished work. `pause` changes `Active` to `Paused` without changing any step, and `resume` restores that same plan to `Active`; both require `expectedRevision`. A repeated pause of an already paused plan, or resume of an already active plan, is an idempotent success only when the supplied revision still matches: it does not write state, increment the revision, or emit another change event. A stale revision is always a conflict. `fail` and `cancel` remain terminal and terminal plans cannot resume. + +Paused plans remain visible through `list_task_plans` and `TaskPlanExtension.ReadAsync` without requesting terminal records. They continue to count toward `MaximumActivePlans`, but do not contribute pending work. While paused, every mutation except idempotent `pause` and `resume` is rejected; the checklist must resume before it can advance, replan, fail, or cancel. A successful transition increments the plan revision, persists the current game moment, and publishes `GameTaskPlanChanged` with reason `pause` or `resume`; as with every extension change event, wait for the matching `SessionSaved` event before treating it as committed UI state. State is namespaced by the runtime's session/actor key and persists through any `IGameSessionStore`. The tool payload cannot select an owner, session, or actor scope. Plans always use the already-authorized `GameInput`/`GameSessionKey`; a server host must resolve and authorize that key before invoking the runtime. @@ -108,6 +110,8 @@ The evidence validator is a read-only authority check, not another world mutatio Hosts that previously inspected `GameSessionSnapshot.ExtensionState` should migrate to `GoalLoopExtension.ReadAsync` and `TaskPlanExtension.ReadAsync`. Treat extension-state key encoding and JSON documents as private storage details. `GameGoalChanged` now follows `GameTaskPlanChanged`: its constructor and every published event include `GameSessionKey` and `InputId`, so event consumers should correlate the change with the matching saved input before updating authoritative UI. +Existing task-plan documents remain valid without migration. `Paused` was appended to the public status enum and is serialized by name; the numeric values and stored JSON names of `Active`, `Completed`, `Failed`, and `Cancelled` are unchanged. Hosts that switch exhaustively on plan status should add `Paused` as a visible, non-terminal, non-runnable state. + ## Monthly or turn-based evolution Represent the calendar in `GameMoment.CalendarJson` while using `Tick` for ordering. A monthly advance can be a named `DurableGameWorkflow`: diff --git a/src/OpenGameAgent.Extensions/TaskPlanExtension.cs b/src/OpenGameAgent.Extensions/TaskPlanExtension.cs index d7151d3..349a117 100644 --- a/src/OpenGameAgent.Extensions/TaskPlanExtension.cs +++ b/src/OpenGameAgent.Extensions/TaskPlanExtension.cs @@ -17,6 +17,7 @@ public enum GameTaskPlanStatus Completed, Failed, Cancelled, + Paused, } [JsonConverter(typeof(JsonStringEnumConverter))] @@ -213,7 +214,7 @@ public sealed class TaskPlanExtension : IGameAgentExtension "type":"object", "required":["action","planId"], "properties":{ - "action":{"type":"string","enum":["create","advance","replace_remaining","fail","cancel"]}, + "action":{"type":"string","enum":["create","advance","replace_remaining","pause","resume","fail","cancel"]}, "planId":{"type":"string","minLength":1,"maxLength":128}, "expectedRevision":{"type":"integer","minimum":1}, "objective":{"type":"string","minLength":1,"maxLength":4096}, @@ -244,9 +245,9 @@ public TaskPlanExtension( public GameAgentExtensionDescriptor Descriptor { get; } = new( ExtensionId, - "1.0.0", - "Persistent ordered task checklists with host-validated advancement.", - new[] { "task-plan", "checklist", "pending-work", "evidence" }); + "1.1.0", + "Persistent ordered task checklists with host-validated advancement and durable pause/resume.", + new[] { "task-plan", "checklist", "pending-work", "evidence", "pause-resume" }); public static async ValueTask ReadAsync( IGameSessionStore sessionStore, @@ -275,7 +276,7 @@ public static async ValueTask ReadAsync( var plans = ReadAll( StoredExtensionStateReader.Read(snapshot, ExtensionId), AbsoluteMaximumStepsPerPlan) - .Where(plan => includeTerminal || plan.Status == GameTaskPlanStatus.Active) + .Where(plan => includeTerminal || !IsTerminal(plan.Status)) .OrderBy(plan => plan.Id, StringComparer.Ordinal) .ToArray(); return new GameTaskPlanQueryResult(key, snapshot.Revision, plans); @@ -285,7 +286,7 @@ public void Configure(GameAgentExtensionApi api) { api.RegisterPromptFragment( "task-plan-guidance", - "Use manage_task_plan for multi-step work that must survive later inputs. An active plan has exactly one in-progress step. Advance only with evidence the host can verify, never by assertion. Use replace_remaining when new world state invalidates unfinished work; completed steps remain immutable."); + "Use manage_task_plan for multi-step work that must survive later inputs. An active or paused plan retains exactly one in-progress step. Paused plans do not drive pending work and must be resumed before advancing. Advance only with evidence the host can verify, never by assertion. Use replace_remaining when new world state invalidates unfinished work; completed steps remain immutable."); api.RegisterToolProvider( "task-plan-tools", (context, _) => new ValueTask>(new[] @@ -307,7 +308,7 @@ private AgentTool CreateManageTool(GameAgentExtensionApi api, GameAgentExtension new( new ToolDefinition( "manage_task_plan", - "Create, advance, replan, fail, or cancel a persistent ordered checklist for the current actor session. Advancing the final step completes the plan.", + "Create, advance, replan, pause, resume, fail, or cancel a persistent ordered checklist for the current actor session. Advancing the final step completes the plan.", ManageSchema), async (arguments, _, cancellationToken) => { @@ -323,11 +324,11 @@ private AgentTool CreateManageTool(GameAgentExtensionApi api, GameAgentExtension } PruneTerminalPlans(context.State); - var activeCount = ReadAll(context.State).Count(plan => plan.Status == GameTaskPlanStatus.Active); + var activeCount = ReadAll(context.State).Count(plan => !IsTerminal(plan.Status)); if (activeCount >= _options.MaximumActivePlans) { return ToolResult.Error( - $"At most {_options.MaximumActivePlans} active task plans may exist in one actor session."); + $"At most {_options.MaximumActivePlans} active or paused task plans may exist in one actor session."); } if (!arguments.TryGetProperty("objective", out var objectiveElement) @@ -370,7 +371,7 @@ private AgentTool CreateManageTool(GameAgentExtensionApi api, GameAgentExtension } document = existing; - if (document.Status != GameTaskPlanStatus.Active) + if (IsTerminal(document.Status)) { return ToolResult.Error($"Task plan '{planId}' is terminal and immutable."); } @@ -382,6 +383,14 @@ private AgentTool CreateManageTool(GameAgentExtensionApi api, GameAgentExtension $"Task plan '{planId}' revision conflict. Current revision is {document.Revision}."); } + if (document.Status == GameTaskPlanStatus.Paused + && action is not "pause" and not "resume") + { + return ToolResult.Error( + $"Task plan '{planId}' is paused and must be resumed before it can change."); + } + + var changed = true; switch (action) { case "advance": @@ -447,6 +456,28 @@ private AgentTool CreateManageTool(GameAgentExtensionApi api, GameAgentExtension : GameTaskPlanStepStatus.Pending, })); document.Steps = completed; + break; + case "pause": + if (document.Status == GameTaskPlanStatus.Paused) + { + changed = false; + } + else + { + document.Status = GameTaskPlanStatus.Paused; + } + + break; + case "resume": + if (document.Status == GameTaskPlanStatus.Active) + { + changed = false; + } + else + { + document.Status = GameTaskPlanStatus.Active; + } + break; case "fail": document.Status = GameTaskPlanStatus.Failed; @@ -464,6 +495,11 @@ private AgentTool CreateManageTool(GameAgentExtensionApi api, GameAgentExtension return ToolResult.Error($"Unsupported task-plan action '{action}'."); } + if (!changed) + { + return JsonResult(new GameTaskPlanSnapshot(document)); + } + document.Revision = checked(document.Revision + 1); document.LastTimelineId = context.Input.Moment.TimelineId; document.LastTick = context.Input.Moment.Tick; @@ -504,7 +540,7 @@ private AgentTool CreateListTool(GameAgentExtensionRunContext context) => var includeTerminal = arguments.TryGetProperty("includeTerminal", out var include) && include.GetBoolean(); var plans = ReadAll(context.State) - .Where(plan => includeTerminal || plan.Status == GameTaskPlanStatus.Active) + .Where(plan => includeTerminal || !IsTerminal(plan.Status)) .OrderBy(plan => plan.Id, StringComparer.Ordinal) .ToArray(); return new ValueTask(JsonResult(new { plans })); @@ -667,12 +703,16 @@ private static void ValidateDocument(TaskPlanDocument document, string expectedI } var inProgress = document.Steps.Count(step => step.Status == GameTaskPlanStepStatus.InProgress); - if ((document.Status == GameTaskPlanStatus.Active && inProgress != 1) - || (document.Status != GameTaskPlanStatus.Active && inProgress != 0) + var resumable = document.Status is GameTaskPlanStatus.Active or GameTaskPlanStatus.Paused; + if ((resumable && inProgress != 1) + || (!resumable && inProgress != 0) || (document.Status == GameTaskPlanStatus.Completed && document.Steps.Any(step => step.Status != GameTaskPlanStepStatus.Completed)) || IsTerminal(document.Status) != (document.TerminalSequence > 0) - || ((document.Status is GameTaskPlanStatus.Active or GameTaskPlanStatus.Completed) && document.Error is not null)) + || ((document.Status is GameTaskPlanStatus.Active + or GameTaskPlanStatus.Paused + or GameTaskPlanStatus.Completed) + && document.Error is not null)) { throw new InvalidOperationException("The task-plan status does not match its checklist."); } diff --git a/tests/OpenGameAgent.Extensions.Tests/TaskPlanExtensionTests.cs b/tests/OpenGameAgent.Extensions.Tests/TaskPlanExtensionTests.cs index 3ba63ed..a2829b6 100644 --- a/tests/OpenGameAgent.Extensions.Tests/TaskPlanExtensionTests.cs +++ b/tests/OpenGameAgent.Extensions.Tests/TaskPlanExtensionTests.cs @@ -8,6 +8,21 @@ namespace OpenGameAgent.Extensions.Tests; public sealed class TaskPlanExtensionTests { + [Fact] + public void PauseStatusPreservesExistingEnumValuesAndJsonNames() + { + Assert.Equal(0, (int)GameTaskPlanStatus.Active); + Assert.Equal(1, (int)GameTaskPlanStatus.Completed); + Assert.Equal(2, (int)GameTaskPlanStatus.Failed); + Assert.Equal(3, (int)GameTaskPlanStatus.Cancelled); + Assert.Equal(4, (int)GameTaskPlanStatus.Paused); + Assert.Equal("\"Active\"", JsonSerializer.Serialize(GameTaskPlanStatus.Active)); + Assert.Equal("\"Completed\"", JsonSerializer.Serialize(GameTaskPlanStatus.Completed)); + Assert.Equal("\"Failed\"", JsonSerializer.Serialize(GameTaskPlanStatus.Failed)); + Assert.Equal("\"Cancelled\"", JsonSerializer.Serialize(GameTaskPlanStatus.Cancelled)); + Assert.Equal("\"Paused\"", JsonSerializer.Serialize(GameTaskPlanStatus.Paused)); + } + [Fact] public async Task OrderedChecklistAdvancesOncePerInputAndPublishesScopedChanges() { @@ -104,6 +119,168 @@ await RunAsync( Assert.Equal("step-1", steps[0].GetProperty("Id").GetString()); } + [Fact] + public async Task PauseAndResumeAreDurableIdempotentAndPreserveProgress() + { + var store = new InMemoryGameSessionStore(); + var extension = new TaskPlanExtension((_, _) => new ValueTask(true)); + var changes = new ConcurrentQueue(); + await RunAsync( + store, + extension, + Input("create"), + ToolCall("create", "{\"action\":\"create\",\"planId\":\"journey\",\"objective\":\"travel\",\"steps\":[\"prepare\",\"walk\",\"arrive\"]}"), + TextResponse("created")); + await RunAsync( + store, + extension, + Input("advance"), + ToolCall("advance", "{\"action\":\"advance\",\"planId\":\"journey\",\"expectedRevision\":1,\"evidence\":{\"kind\":\"receipt\",\"reference\":\"prepared\"}}"), + TextResponse("advanced")); + + await RunWithChangesAsync( + store, + extension, + changes, + Input("pause"), + ToolCall("pause", "{\"action\":\"pause\",\"planId\":\"journey\",\"expectedRevision\":2}"), + ToolCall("pause-again", "{\"action\":\"pause\",\"planId\":\"journey\",\"expectedRevision\":3}"), + TextResponse("paused")); + + var pausedQuery = await TaskPlanExtension.ReadAsync( + store, + new GameSessionKey("session", "actor"), + cancellationToken: TestContext.Current.CancellationToken); + var paused = Assert.Single(pausedQuery.Plans); + Assert.Equal(GameTaskPlanStatus.Paused, paused.Status); + Assert.Equal(3, paused.Revision); + Assert.Equal( + new[] + { + GameTaskPlanStepStatus.Completed, + GameTaskPlanStepStatus.InProgress, + GameTaskPlanStepStatus.Pending, + }, + paused.Steps.Select(step => step.Status).ToArray()); + Assert.Equal("pause", Assert.Single(changes).Reason); + + var listProvider = new ScriptedProvider(new[] + { + ListToolCall("list", "{}"), + TextResponse("listed"), + }); + await using (var listRuntime = new GameAgentBuilder(listProvider, "model") + .UseSessionStore(store) + .UseExtension(extension) + .Build()) + { + var result = await listRuntime.RunAsync(Input("list"), TestContext.Current.CancellationToken); + Assert.True(result.Succeeded, result.Error ?? result.AgentResult?.Error); + } + + var listMessage = listProvider.Requests.ElementAt(1).Messages.Last(message => message.Role == AgentRole.Tool); + var listJson = Assert.IsType(Assert.Single(listMessage.Content)).Json; + Assert.Contains("\"Paused\"", listJson, StringComparison.Ordinal); + + var pending = new ConcurrentQueue(); + await using (var pendingRuntime = new GameAgentBuilder( + new ScriptedProvider(new[] { TextResponse("observed") }), + "model") + .UseSessionStore(store) + .UseExtension(extension) + .UseExtension( + "paused.pending-work.observer", + "1", + api => api.RegisterRouteRule( + "capture", + (_, _, hasPendingWork, _) => + { + pending.Enqueue(hasPendingWork); + return new ValueTask(GameRouteDecision.Agent("captured")); + }, + priority: 1_000)) + .Build()) + { + var result = await pendingRuntime.RunAsync(Input("paused-pending"), TestContext.Current.CancellationToken); + Assert.True(result.Succeeded, result.Error ?? result.AgentResult?.Error); + } + + Assert.False(Assert.Single(pending)); + + await RunAsync( + store, + extension, + Input("blocked"), + ToolCall("advance-paused", "{\"action\":\"advance\",\"planId\":\"journey\",\"expectedRevision\":3,\"evidence\":{\"kind\":\"receipt\",\"reference\":\"must-not-run\"}}"), + ToolCall("replace-paused", "{\"action\":\"replace_remaining\",\"planId\":\"journey\",\"expectedRevision\":3,\"steps\":[\"changed\"]}"), + ToolCall("fail-paused", "{\"action\":\"fail\",\"planId\":\"journey\",\"expectedRevision\":3}"), + ToolCall("cancel-paused", "{\"action\":\"cancel\",\"planId\":\"journey\",\"expectedRevision\":3}"), + ToolCall("stale-resume", "{\"action\":\"resume\",\"planId\":\"journey\",\"expectedRevision\":2}"), + TextResponse("blocked")); + var stillPaused = Assert.Single((await TaskPlanExtension.ReadAsync( + store, + new GameSessionKey("session", "actor"), + cancellationToken: TestContext.Current.CancellationToken)).Plans); + Assert.Equal(GameTaskPlanStatus.Paused, stillPaused.Status); + Assert.Equal(3, stillPaused.Revision); + + changes.Clear(); + await RunWithChangesAsync( + store, + extension, + changes, + Input("resume"), + ToolCall("resume", "{\"action\":\"resume\",\"planId\":\"journey\",\"expectedRevision\":3}"), + ToolCall("resume-again", "{\"action\":\"resume\",\"planId\":\"journey\",\"expectedRevision\":4}"), + TextResponse("resumed")); + var resumed = Assert.Single((await TaskPlanExtension.ReadAsync( + store, + new GameSessionKey("session", "actor"), + cancellationToken: TestContext.Current.CancellationToken)).Plans); + Assert.Equal(GameTaskPlanStatus.Active, resumed.Status); + Assert.Equal(4, resumed.Revision); + Assert.Equal(paused.Steps.Select(step => step.Id), resumed.Steps.Select(step => step.Id)); + Assert.Equal(paused.Steps.Select(step => step.Status), resumed.Steps.Select(step => step.Status)); + Assert.Equal("resume", Assert.Single(changes).Reason); + } + + [Fact] + public async Task TerminalTaskPlanCannotResume() + { + var store = new InMemoryGameSessionStore(); + var extension = new TaskPlanExtension((_, _) => new ValueTask(true)); + await RunAsync( + store, + extension, + Input("create"), + ToolCall("create", "{\"action\":\"create\",\"planId\":\"done\",\"objective\":\"finish\",\"steps\":[\"only\"]}"), + TextResponse("created")); + await RunAsync( + store, + extension, + Input("complete"), + ToolCall("complete", "{\"action\":\"advance\",\"planId\":\"done\",\"expectedRevision\":1,\"evidence\":{\"kind\":\"receipt\",\"reference\":\"done\"}}"), + TextResponse("completed")); + await RunAsync( + store, + extension, + Input("resume"), + ToolCall("resume", "{\"action\":\"resume\",\"planId\":\"done\",\"expectedRevision\":2}"), + TextResponse("immutable")); + + Assert.Empty((await TaskPlanExtension.ReadAsync( + store, + new GameSessionKey("session", "actor"), + cancellationToken: TestContext.Current.CancellationToken)).Plans); + var completed = Assert.Single((await TaskPlanExtension.ReadAsync( + store, + new GameSessionKey("session", "actor"), + includeTerminal: true, + cancellationToken: TestContext.Current.CancellationToken)).Plans); + Assert.Equal(GameTaskPlanStatus.Completed, completed.Status); + Assert.Equal(2, completed.Revision); + } + [Fact] public async Task ConcurrentMutationsUseSessionCas() { @@ -142,6 +319,44 @@ await RunAsync( Assert.Equal(2, document.RootElement.GetProperty("Revision").GetInt64()); } + [Fact] + public async Task ConcurrentPauseUsesSessionCas() + { + var store = new InMemoryGameSessionStore(); + await RunAsync( + store, + new TaskPlanExtension((_, _) => new ValueTask(true)), + Input("seed"), + ToolCall("create", "{\"action\":\"create\",\"planId\":\"shared-pause\",\"objective\":\"shared work\",\"steps\":[\"one\",\"two\"]}"), + TextResponse("created")); + + var gate = new ConcurrentRunGate(2); + var response = ToolCall( + "pause", + "{\"action\":\"pause\",\"planId\":\"shared-pause\",\"expectedRevision\":1}"); + await using var left = new GameAgentBuilder(new FirstCallBarrierProvider(gate, response), "model") + .UseSessionStore(store) + .UseExtension(new TaskPlanExtension((_, _) => new ValueTask(true))) + .Build(); + await using var right = new GameAgentBuilder(new FirstCallBarrierProvider(gate, response), "model") + .UseSessionStore(store) + .UseExtension(new TaskPlanExtension((_, _) => new ValueTask(true))) + .Build(); + + var results = await Task.WhenAll( + left.RunAsync(Input("left-pause"), TestContext.Current.CancellationToken), + right.RunAsync(Input("right-pause"), TestContext.Current.CancellationToken)); + + Assert.Contains(results, result => result.Status == GameAgentRunStatus.SessionConflict); + var paused = Assert.Single((await TaskPlanExtension.ReadAsync( + store, + new GameSessionKey("session", "actor"), + cancellationToken: TestContext.Current.CancellationToken)).Plans); + Assert.Equal(GameTaskPlanStatus.Paused, paused.Status); + Assert.Equal(2, paused.Revision); + Assert.Single(paused.Steps, step => step.Status == GameTaskPlanStepStatus.InProgress); + } + [Fact] public async Task StateIsIsolatedBySessionAndActor() { @@ -241,6 +456,40 @@ await RunAsync( Assert.Equal("Completed", retained.RootElement.GetProperty("Status").GetString()); } + [Fact] + public async Task PausedPlanStillConsumesNonTerminalCapacity() + { + var store = new InMemoryGameSessionStore(); + var extension = new TaskPlanExtension( + (_, _) => new ValueTask(true), + new TaskPlanOptions { MaximumActivePlans = 1 }); + await RunAsync( + store, + extension, + Input("create"), + ToolCall("create", "{\"action\":\"create\",\"planId\":\"first\",\"objective\":\"work\",\"steps\":[\"one\"]}"), + TextResponse("created")); + await RunAsync( + store, + extension, + Input("pause"), + ToolCall("pause", "{\"action\":\"pause\",\"planId\":\"first\",\"expectedRevision\":1}"), + TextResponse("paused")); + await RunAsync( + store, + extension, + Input("second"), + ToolCall("create-second", "{\"action\":\"create\",\"planId\":\"second\",\"objective\":\"other\",\"steps\":[\"one\"]}"), + TextResponse("rejected")); + + var plan = Assert.Single((await TaskPlanExtension.ReadAsync( + store, + new GameSessionKey("session", "actor"), + cancellationToken: TestContext.Current.CancellationToken)).Plans); + Assert.Equal("first", plan.Id); + Assert.Equal(GameTaskPlanStatus.Paused, plan.Status); + } + [Fact] public async Task ActivePlanIsExposedAsPendingWorkOnLaterInputs() { @@ -347,6 +596,29 @@ private static async Task RunAsync( Assert.True(result.Succeeded, result.Error ?? result.AgentResult?.Error); } + private static async Task RunWithChangesAsync( + IGameSessionStore store, + TaskPlanExtension extension, + ConcurrentQueue changes, + GameInput input, + params ModelResponse[] responses) + { + await using var runtime = new GameAgentBuilder(new ScriptedProvider(responses), "model") + .UseSessionStore(store) + .UseExtension(extension) + .UseExtension( + "task-plan.change-listener", + "1", + api => api.Subscribe(TaskPlanExtension.PlanChanged, (change, _) => + { + changes.Enqueue(change); + return ValueTask.CompletedTask; + })) + .Build(); + var result = await runtime.RunAsync(input, TestContext.Current.CancellationToken); + Assert.True(result.Succeeded, result.Error ?? result.AgentResult?.Error); + } + private static GameInput Input( string inputId, string sessionId = "session", @@ -373,6 +645,9 @@ private static string[] Statuses(JsonElement plan) => private static ModelResponse ToolCall(string id, string arguments) => new(new AgentContent[] { new ToolCallContent(id, "manage_task_plan", arguments) }, ModelStopReason.ToolUse); + private static ModelResponse ListToolCall(string id, string arguments) => + new(new AgentContent[] { new ToolCallContent(id, "list_task_plans", arguments) }, ModelStopReason.ToolUse); + private static ModelResponse TextResponse(string text) => new(new AgentContent[] { new TextContent(text) }, ModelStopReason.Stop); @@ -391,11 +666,14 @@ public ScriptedProvider(Func response) _response = response; } + public ConcurrentQueue Requests { get; } = new(); + public async IAsyncEnumerable StreamAsync( ModelRequest request, [EnumeratorCancellation] CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); + Requests.Enqueue(request); yield return ModelStreamEvent.Terminal(_response(Interlocked.Increment(ref _calls))); await Task.CompletedTask; } diff --git a/tests/OpenGameAgent.Persistence.Tests/TaskPlanPersistenceTests.cs b/tests/OpenGameAgent.Persistence.Tests/TaskPlanPersistenceTests.cs index 29109aa..b5993a7 100644 --- a/tests/OpenGameAgent.Persistence.Tests/TaskPlanPersistenceTests.cs +++ b/tests/OpenGameAgent.Persistence.Tests/TaskPlanPersistenceTests.cs @@ -61,6 +61,71 @@ public async Task ChecklistRevisionAndAdvanceGuardSurviveProcessRestart() Assert.Equal(1, Volatile.Read(ref evidenceCalls)); } + [Fact] + public async Task PausedChecklistAndInProgressStepSurviveProcessRestart() + { + using var directory = new TemporaryDirectory(); + var key = new GameSessionKey("session", "actor"); + GameTaskPlanEvidenceValidator validator = (_, _) => new ValueTask(true); + await using (var runtime = new GameAgentBuilder( + new ScriptedProvider(call => call == 1 + ? ToolCall("create", "{\"action\":\"create\",\"planId\":\"paused\",\"objective\":\"persist pause\",\"steps\":[\"one\",\"two\"]}") + : TextResponse("created")), + "model") + .UseSessionStore(new FileGameSessionStore(directory.Path)) + .UseExtension(new TaskPlanExtension(validator)) + .Build()) + { + var result = await runtime.RunAsync(Input("create"), TestContext.Current.CancellationToken); + Assert.True(result.Succeeded); + } + + await using (var runtime = new GameAgentBuilder( + new ScriptedProvider(call => call == 1 + ? ToolCall("pause", "{\"action\":\"pause\",\"planId\":\"paused\",\"expectedRevision\":1}") + : TextResponse("paused")), + "model") + .UseSessionStore(new FileGameSessionStore(directory.Path)) + .UseExtension(new TaskPlanExtension(validator)) + .Build()) + { + var result = await runtime.RunAsync(Input("pause"), TestContext.Current.CancellationToken); + Assert.True(result.Succeeded); + } + + var restartedStore = new FileGameSessionStore(directory.Path); + var paused = Assert.Single((await TaskPlanExtension.ReadAsync( + restartedStore, + key, + cancellationToken: TestContext.Current.CancellationToken)).Plans); + Assert.Equal(GameTaskPlanStatus.Paused, paused.Status); + Assert.Equal(2, paused.Revision); + var inProgress = Assert.Single(paused.Steps, step => step.Status == GameTaskPlanStepStatus.InProgress); + + await using (var runtime = new GameAgentBuilder( + new ScriptedProvider(call => call == 1 + ? ToolCall("resume", "{\"action\":\"resume\",\"planId\":\"paused\",\"expectedRevision\":2}") + : TextResponse("resumed")), + "model") + .UseSessionStore(new FileGameSessionStore(directory.Path)) + .UseExtension(new TaskPlanExtension(validator)) + .Build()) + { + var result = await runtime.RunAsync(Input("resume"), TestContext.Current.CancellationToken); + Assert.True(result.Succeeded); + } + + var resumed = Assert.Single((await TaskPlanExtension.ReadAsync( + new FileGameSessionStore(directory.Path), + key, + cancellationToken: TestContext.Current.CancellationToken)).Plans); + Assert.Equal(GameTaskPlanStatus.Active, resumed.Status); + Assert.Equal(3, resumed.Revision); + Assert.Equal(inProgress.Id, Assert.Single( + resumed.Steps, + step => step.Status == GameTaskPlanStepStatus.InProgress).Id); + } + private static GameInput Input(string inputId) => new("session", "actor", "request", "{}", new GameMoment("world", 1), inputId); From e075992634a775ca19d0a657156f6365312d03ac Mon Sep 17 00:00:00 2001 From: Eric Sun <141227631+EricSun0218@users.noreply.github.com> Date: Thu, 13 Aug 2026 21:25:13 +0800 Subject: [PATCH 2/2] Prevent canceled credential mutations from running --- src/OpenGameAgent.Models/Credentials.cs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/OpenGameAgent.Models/Credentials.cs b/src/OpenGameAgent.Models/Credentials.cs index 2f207e2..ad807f8 100644 --- a/src/OpenGameAgent.Models/Credentials.cs +++ b/src/OpenGameAgent.Models/Credentials.cs @@ -218,6 +218,7 @@ public async ValueTask RemoveAsync(GameCredentialKey key, CancellationToke CancellationToken cancellationToken) { using var lease = await AcquireAsync(key, cancellationToken).ConfigureAwait(false); + cancellationToken.ThrowIfCancellationRequested(); GameCredential? current; lock (_stateGate) {