From b31b2edb76c3fbc0c1e408f761d2cc6c73289909 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kim=20Harjam=C3=A4ki?= Date: Tue, 7 Jul 2026 18:39:41 +0300 Subject: [PATCH] fix(recovery): preserve failed MCP state for fault injection --- .../CheckpointCorruptionTests.cs | 110 ++++++++++++ .../FaultInjectionTests.cs | 163 ++++++++++++++++++ .../Workflows/GsdStateMachine.cs | 2 +- 3 files changed, 274 insertions(+), 1 deletion(-) create mode 100644 src/GsdOrchestrator.Tests/CheckpointCorruptionTests.cs create mode 100644 src/GsdOrchestrator.Tests/FaultInjectionTests.cs diff --git a/src/GsdOrchestrator.Tests/CheckpointCorruptionTests.cs b/src/GsdOrchestrator.Tests/CheckpointCorruptionTests.cs new file mode 100644 index 0000000..3cdb3ce --- /dev/null +++ b/src/GsdOrchestrator.Tests/CheckpointCorruptionTests.cs @@ -0,0 +1,110 @@ +using System.Text.Json; +using GsdOrchestrator.Checkpointing; +using GsdOrchestrator.Workflows.Models; +using Microsoft.Extensions.Logging.Abstractions; +using Xunit; + +namespace GsdOrchestrator.Tests; + +public sealed class CheckpointCorruptionTests +{ + [Fact] + public async Task LoadAsync_CorruptedFinalCheckpoint_ThrowsJsonException() + { + using var fixture = new CheckpointFixture(); + var store = fixture.CreateStore(); + var context = ValidContext("corrupt-final"); + var path = fixture.StatePath(context.WorkflowId); + + await store.SaveAsync(context); + await File.WriteAllTextAsync(path, "{\"workflowId\":\"corrupt-final\",\"currentSt"); + + await Assert.ThrowsAsync(() => store.LoadAsync(context.WorkflowId)); + } + + [Fact] + public async Task LoadAsync_OrphanedTmpFile_IgnoresGarbageAndReturnsLastGoodCheckpoint() + { + using var fixture = new CheckpointFixture(); + var store = fixture.CreateStore(); + var context = ValidContext("orphaned-tmp"); + var path = fixture.StatePath(context.WorkflowId); + var tmpPath = path + ".tmp"; + + await store.SaveAsync(context); + await File.WriteAllTextAsync(tmpPath, "garbage bytes from interrupted second save"); + + var loaded = await store.LoadAsync(context.WorkflowId); + + Assert.NotNull(loaded); + Assert.Equal(context.WorkflowId, loaded!.WorkflowId); + Assert.Equal(context.CurrentState, loaded.CurrentState); + Assert.Equal("1.1", loaded.SchemaVersion); + } + + [Fact] + public async Task LoadAsync_UnsupportedSchemaVersion_ThrowsInvalidDataException() + { + using var fixture = new CheckpointFixture(); + var store = fixture.CreateStore(); + var context = ValidContext("unsupported-version"); + + await store.SaveAsync(context); + await RewriteSchemaVersionAsync(fixture.StatePath(context.WorkflowId), "99.0"); + + var error = await Assert.ThrowsAsync(() => store.LoadAsync(context.WorkflowId)); + Assert.Contains("unsupported schema version", error.Message, StringComparison.OrdinalIgnoreCase); + } + + [Fact] + public async Task LoadAsync_LegacySchemaVersion_UpgradesToCurrentVersion() + { + using var fixture = new CheckpointFixture(); + var store = fixture.CreateStore(); + var context = ValidContext("legacy-version"); + + await store.SaveAsync(context); + await RewriteSchemaVersionAsync(fixture.StatePath(context.WorkflowId), "1.0"); + + var loaded = await store.LoadAsync(context.WorkflowId); + + Assert.NotNull(loaded); + Assert.Equal("1.1", loaded!.SchemaVersion); + Assert.Equal(context.WorkflowId, loaded.WorkflowId); + } + + private static async Task RewriteSchemaVersionAsync(string path, string schemaVersion) + { + var checkpoint = await File.ReadAllTextAsync(path); + var rewritten = checkpoint.Replace("\"schemaVersion\": \"1.1\"", $"\"schemaVersion\": \"{schemaVersion}\"", StringComparison.Ordinal); + await File.WriteAllTextAsync(path, rewritten); + } + + private static GsdWorkflowContext ValidContext(string workflowId) => + new() + { + WorkflowId = workflowId, + SchemaVersion = "1.1", + CurrentState = WorkflowState.Analyzing, + Issue = null + }; + + private sealed class CheckpointFixture : IDisposable + { + private readonly string _root = Directory.CreateTempSubdirectory("gsd-checkpoint-tests-").FullName; + + public FileCheckpointStore CreateStore() => + new(_root, NullLogger.Instance); + + public string StatePath(string workflowId) => + Path.Combine(_root, ".gsd", "state", $"{workflowId}.json"); + + public void Dispose() + { + if (Directory.Exists(_root)) + { + Directory.Delete(_root, recursive: true); + } + } + } +} \ No newline at end of file diff --git a/src/GsdOrchestrator.Tests/FaultInjectionTests.cs b/src/GsdOrchestrator.Tests/FaultInjectionTests.cs new file mode 100644 index 0000000..3ff79d6 --- /dev/null +++ b/src/GsdOrchestrator.Tests/FaultInjectionTests.cs @@ -0,0 +1,163 @@ +using GsdOrchestrator.Checkpointing; +using GsdOrchestrator.Mcp; +using GsdOrchestrator.Workflows; +using GsdOrchestrator.Workflows.Models; +using GsdOrchestrator.Workflows.States; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using NSubstitute.ExceptionExtensions; +using Polly.Registry; +using Xunit; + +namespace GsdOrchestrator.Tests; + +public sealed class FaultInjectionTests +{ + private static GsdStateMachine BuildSut( + ICheckpointStore checkpoints, + IWorkflowState[] states, + IMcpClient? mcpClient = null) + { + var client = mcpClient ?? Substitute.For(); + var registry = new ResiliencePipelineRegistry(); + registry.TryAddBuilder("mcp-tools", (b, _) => { }); + var dispatcher = new McpToolDispatcher( + client, + registry, + NullLogger.Instance); + return new GsdStateMachine( + checkpoints, + dispatcher, + states, + NullLogger.Instance); + } + + private static IWorkflowState MakeTransitionState(WorkflowState from, WorkflowState to) + { + var state = Substitute.For(); + state.State.Returns(from); + state.ExecuteAsync(Arg.Any(), Arg.Any()) + .Returns(ci => Task.FromResult(ci.Arg().Transition(to))); + return state; + } + + [Fact] + public async Task RunAsync_McpFailureMidGoal_ProducesTypedFailedStateAndCheckpoint() + { + var checkpoints = Substitute.For(); + checkpoints.SaveAsync(Arg.Any(), Arg.Any()) + .Returns(Task.CompletedTask); + + var idle = MakeTransitionState(WorkflowState.Idle, WorkflowState.Analyzing); + var analyzing = MakeTransitionState(WorkflowState.Analyzing, WorkflowState.Editing); + var editing = Substitute.For(); + editing.State.Returns(WorkflowState.Editing); + editing.ExecuteAsync(Arg.Any(), Arg.Any()) + .ThrowsAsync(new McpException("simulated MCP failure", isTransient: true)); + + var sut = BuildSut(checkpoints, [idle, analyzing, editing]); + + var ctx = await sut.RunAsync("owner", "repo", 42, triageModeOnly: false, CancellationToken.None); + + Assert.Equal(WorkflowState.Failed, ctx.CurrentState); + Assert.Equal(WorkflowState.Editing, ctx.FailedState); + Assert.Contains("simulated MCP failure", ctx.FailureReason); + await checkpoints.Received().SaveAsync( + Arg.Is(saved => + saved.CurrentState == WorkflowState.Failed && + saved.FailedState == WorkflowState.Editing && + saved.FailureReason != null && + saved.FailureReason.Contains("simulated MCP failure", StringComparison.Ordinal)), + Arg.Any()); + } + + [Fact] + public async Task ResumeAsync_RetriesFailedStateOnce_ThenHaltsDeterministically() + { + var checkpoints = Substitute.For(); + checkpoints.SaveAsync(Arg.Any(), Arg.Any()) + .Returns(Task.CompletedTask); + + var firstCheckpoint = FailedCheckpoint("retry-wf", retryCount: 0); + var secondCheckpoint = FailedCheckpoint("retry-wf", retryCount: 1); + checkpoints.LoadAsync("retry-wf", Arg.Any()) + .Returns(firstCheckpoint, secondCheckpoint); + + var editing = Substitute.For(); + editing.State.Returns(WorkflowState.Editing); + editing.ExecuteAsync(Arg.Any(), Arg.Any()) + .ThrowsAsync(new McpException("editing exploded again", isTransient: true)); + + var sut = BuildSut(checkpoints, [editing]); + + var firstResult = await sut.ResumeAsync("retry-wf", CancellationToken.None); + + Assert.Equal(WorkflowState.Failed, firstResult.CurrentState); + Assert.Equal(1, firstResult.RetryCount); + Assert.Equal(WorkflowState.Editing, firstResult.FailedState); + Assert.Contains("editing exploded again", firstResult.FailureReason); + await checkpoints.Received().SaveAsync( + Arg.Is(saved => + saved.CurrentState == WorkflowState.Failed && + saved.RetryCount == 1 && + saved.FailedState == WorkflowState.Editing), + Arg.Any()); + + var error = await Assert.ThrowsAsync( + () => sut.ResumeAsync("retry-wf", CancellationToken.None)); + Assert.Contains("recovery retry limit", error.Message, StringComparison.OrdinalIgnoreCase); + } + + [Fact] + public async Task RunAsync_TruncatesFailureReasonTo1024Characters() + { + var checkpoints = Substitute.For(); + checkpoints.SaveAsync(Arg.Any(), Arg.Any()) + .Returns(Task.CompletedTask); + + var oversizedMessage = new string('x', 1200); + var idle = Substitute.For(); + idle.State.Returns(WorkflowState.Idle); + idle.ExecuteAsync(Arg.Any(), Arg.Any()) + .ThrowsAsync(new InvalidOperationException(oversizedMessage)); + + var sut = BuildSut(checkpoints, [idle]); + + var ctx = await sut.RunAsync("owner", "repo", 7, triageModeOnly: false, CancellationToken.None); + + Assert.Equal(WorkflowState.Failed, ctx.CurrentState); + Assert.NotNull(ctx.FailureReason); + Assert.Equal(1024, ctx.FailureReason!.Length); + } + + [Fact] + public async Task RunAsync_BudgetFailureMessage_SetsBudgetExhaustedStopReason() + { + var checkpoints = Substitute.For(); + checkpoints.SaveAsync(Arg.Any(), Arg.Any()) + .Returns(Task.CompletedTask); + + var idle = Substitute.For(); + idle.State.Returns(WorkflowState.Idle); + idle.ExecuteAsync(Arg.Any(), Arg.Any()) + .ThrowsAsync(new InvalidOperationException("budget exhausted during planning")); + + var sut = BuildSut(checkpoints, [idle]); + + var ctx = await sut.RunAsync("owner", "repo", 8, triageModeOnly: false, CancellationToken.None); + + Assert.Equal(WorkflowState.Failed, ctx.CurrentState); + Assert.Equal(TerminalStopReason.BudgetExhausted, ctx.StopReason); + } + + private static GsdWorkflowContext FailedCheckpoint(string workflowId, int retryCount) => + new() + { + WorkflowId = workflowId, + CurrentState = WorkflowState.Failed, + FailedState = WorkflowState.Editing, + RetryCount = retryCount, + FailureReason = "prior failure", + Issue = new IssueContext(1, "Retry workflow", "", [], "owner", "repo", "main") + }; +} diff --git a/src/GsdOrchestrator/Workflows/GsdStateMachine.cs b/src/GsdOrchestrator/Workflows/GsdStateMachine.cs index 92b52b3..d0dd651 100644 --- a/src/GsdOrchestrator/Workflows/GsdStateMachine.cs +++ b/src/GsdOrchestrator/Workflows/GsdStateMachine.cs @@ -235,7 +235,7 @@ private async Task ExecuteLoopAsync(GsdWorkflowContext ctx, InvalidatedPhaseIds: [phase.RollbackTo ?? phase.Id], Reason: failureReason)); } - if (ctx.PendingRollback is not null) + if (ctx.PendingRollback is not null && ex is not McpException) { _logger.LogWarning( "Workflow {WorkflowId} rollback requested from {FromPhase} to {ToPhase}: {Reason}",