diff --git a/src/GsdOrchestrator.Tests/FailureStateTests.cs b/src/GsdOrchestrator.Tests/FailureStateTests.cs new file mode 100644 index 0000000..3ce0d93 --- /dev/null +++ b/src/GsdOrchestrator.Tests/FailureStateTests.cs @@ -0,0 +1,67 @@ +using GsdOrchestrator.Loop; +using GsdOrchestrator.Scheduling; +using Xunit; + +namespace GsdOrchestrator.Tests; + +public sealed class FailureStateTests +{ + [Fact] + public void Timeout_Is_Classified_As_Transient_And_Retryable() + { + var result = FailureClassifier.Classify(new TimeoutException("timed out"), "component"); + + Assert.Equal(FailureClass.Transient, result.FailureClass); + Assert.True(result.Retryable); + Assert.Equal(5d, result.RetryAfterSeconds); + } + + [Fact] + public void OperationCanceled_Is_Classified_As_Cancellation() + { + var result = FailureClassifier.Classify(new OperationCanceledException("cancelled"), "component"); + + Assert.Equal(FailureClass.Cancellation, result.FailureClass); + Assert.False(result.Retryable); + Assert.Null(result.RetryAfterSeconds); + } + + [Fact] + public void InvalidOperation_Is_Classified_As_Deterministic() + { + var result = FailureClassifier.Classify(new InvalidOperationException("Goal 'x' was not found."), "component"); + + Assert.Equal(FailureClass.Deterministic, result.FailureClass); + Assert.False(result.Retryable); + } + + [Fact] + public void UnauthorizedAccess_Is_Classified_As_Policy() + { + var result = FailureClassifier.Classify(new UnauthorizedAccessException("denied"), "component"); + + Assert.Equal(FailureClass.Policy, result.FailureClass); + Assert.False(result.Retryable); + } + + [Fact] + public void Unknown_Exception_Falls_Back_To_Unrecoverable() + { + var result = FailureClassifier.Classify(new UnknownFailureException("boom"), "component"); + + Assert.Equal(FailureClass.Unrecoverable, result.FailureClass); + Assert.False(result.Retryable); + } + + [Fact] + public void Cause_Chain_Is_Collected_Outermost_First() + { + var exception = new InvalidOperationException("outer", new TimeoutException("inner")); + + var result = FailureClassifier.Classify(exception, "component"); + + Assert.Equal(["outer", "inner"], result.CauseChain); + } + + private sealed class UnknownFailureException(string message) : Exception(message); +} diff --git a/src/GsdOrchestrator.Tests/LoopCoordinatorTests.cs b/src/GsdOrchestrator.Tests/LoopCoordinatorTests.cs index b696600..7ce8737 100644 --- a/src/GsdOrchestrator.Tests/LoopCoordinatorTests.cs +++ b/src/GsdOrchestrator.Tests/LoopCoordinatorTests.cs @@ -38,6 +38,23 @@ public async Task RunAsync_CreatesBoundedRepairAndReverifies() Assert.Single(fixture.Learning.Outcomes); } + [Fact] + public async Task RunAsync_Converts_Worker_Exception_Into_Typed_Failure_State() + { + await using var fixture = await Fixture.CreateAsync([Passed()], worker: new ThrowingWorker(new TimeoutException("worker timeout"))); + + var result = await fixture.Coordinator.RunAsync("goal-1", Budget()); + + Assert.Equal(GoalStatus.Failed, result.Aggregate.Goal.Status); + Assert.Contains(result.Aggregate.Evidence, item => item.Kind == "failure-state"); + var failureEvent = Assert.Single(result.Aggregate.Events, item => item.Type == "goal.failed"); + Assert.Contains("Transient", failureEvent.PayloadJson); + Assert.Contains("worker timeout", failureEvent.PayloadJson); + Assert.Single(fixture.Learning.Outcomes); + Assert.Contains("worker timeout", fixture.Learning.Outcomes[0].Summary); + Assert.Equal(GoalStatus.Failed, fixture.Learning.Outcomes[0].Status); + } + [Fact] public void PolicyGuard_DeniesEnvironmentAndHoldsExternalActions() { @@ -60,14 +77,14 @@ private sealed class Fixture : IAsyncDisposable private Fixture(string path, LoopCoordinator coordinator, CapturingLearning learning) => (_path, Coordinator, Learning) = (path, coordinator, learning); - public static async Task CreateAsync(IReadOnlyList results) + public static async Task CreateAsync(IReadOnlyList results, ILoopWorker? worker = null) { var path = Path.Combine(Path.GetTempPath(), $"loop-{Guid.NewGuid():N}.db"); var store = new SqliteGoalStore(path, NullLogger.Instance); await store.InitializeAsync(); await store.SaveAsync(Aggregate()); var learning = new CapturingLearning(); - return new(path, new(store, new SuccessfulWorker(), new ScriptedVerifier(results), learning), learning); + return new(path, new(store, worker ?? new SuccessfulWorker(), new ScriptedVerifier(results), learning, NullLogger.Instance), learning); } public ValueTask DisposeAsync() @@ -88,6 +105,12 @@ public Task ExecuteAsync(LoopWorkRequest request, CancellationTo Task.FromResult(new LoopWorkResult(true, [$"cas://evidence/worker/{request.Attempt}"], request.IsRepair ? "repair" : "feature")); } + private sealed class ThrowingWorker(Exception exception) : ILoopWorker + { + public Task ExecuteAsync(LoopWorkRequest request, CancellationToken cancellationToken) => + Task.FromException(exception); + } + private sealed class ScriptedVerifier(IReadOnlyList results) : ILoopVerifier { private int _index; diff --git a/src/GsdOrchestrator/Loop/FailureState.cs b/src/GsdOrchestrator/Loop/FailureState.cs new file mode 100644 index 0000000..5b78995 --- /dev/null +++ b/src/GsdOrchestrator/Loop/FailureState.cs @@ -0,0 +1,61 @@ +using System.Net.Http; +using System.Net.Sockets; +using GsdOrchestrator.Scheduling; + +namespace GsdOrchestrator.Loop; + +public sealed record FailureState( + FailureClass FailureClass, + string Component, + string Message, + bool Retryable, + string? ExceptionType = null, + IReadOnlyList? CauseChain = null, + double? RetryAfterSeconds = null); + +public static class FailureClassifier +{ + private const int MaxMessageLength = 1024; + private const int MaxCauseChainEntries = 10; + + public static FailureState Classify(Exception exception, string component) + { + var (failureClass, retryable, retryAfterSeconds) = exception switch + { + OperationCanceledException => (FailureClass.Cancellation, false, (double?)null), + TimeoutException => (FailureClass.Transient, true, 5d), + HttpRequestException => (FailureClass.Transient, true, 5d), + SocketException => (FailureClass.Transient, true, 5d), + UnauthorizedAccessException => (FailureClass.Policy, false, (double?)null), + ArgumentOutOfRangeException => (FailureClass.Policy, false, (double?)null), + InvalidOperationException => (FailureClass.Deterministic, false, (double?)null), + ArgumentException => (FailureClass.Deterministic, false, (double?)null), + _ => (FailureClass.Unrecoverable, false, (double?)null), + }; + + return new FailureState( + failureClass, + component, + Truncate(exception.Message), + retryable, + exception.GetType().FullName, + BuildCauseChain(exception), + retryAfterSeconds); + } + + private static string Truncate(string message) => + message.Length <= MaxMessageLength ? message : message[..MaxMessageLength]; + + private static IReadOnlyList? BuildCauseChain(Exception exception) + { + var chain = new List(capacity: 4); + var current = exception; + while (current is not null && chain.Count < MaxCauseChainEntries) + { + chain.Add(Truncate(current.Message)); + current = current.InnerException; + } + + return chain.Count == 0 ? null : chain; + } +} diff --git a/src/GsdOrchestrator/Loop/LoopCoordinator.cs b/src/GsdOrchestrator/Loop/LoopCoordinator.cs index 2928637..d5ba374 100644 --- a/src/GsdOrchestrator/Loop/LoopCoordinator.cs +++ b/src/GsdOrchestrator/Loop/LoopCoordinator.cs @@ -4,6 +4,7 @@ using GsdOrchestrator.Mcp; using GsdOrchestrator.Scheduling; using GsdOrchestrator.Verification; +using Microsoft.Extensions.Logging; namespace GsdOrchestrator.Loop; @@ -87,7 +88,7 @@ public async Task PublishAsync(TerminalLoopOutcome outcome, CancellationToken ca } } -public sealed class LoopCoordinator(IGoalStore store, ILoopWorker worker, ILoopVerifier verifier, ITerminalOutcomePublisher learning) +public sealed class LoopCoordinator(IGoalStore store, ILoopWorker worker, ILoopVerifier verifier, ITerminalOutcomePublisher learning, ILogger logger) { public async Task RunAsync( string goalId, @@ -107,55 +108,93 @@ public async Task RunAsync( ?? throw new InvalidOperationException($"Work item '{workItem.Id}' could not be leased."); aggregate = (await store.LoadAsync(goalId, cancellationToken))!; - var first = await ExecuteAttemptAsync(aggregate, workItem, 1, false, cancellationToken); - aggregate = first.Aggregate; - var workerAttempts = 1; - var peakConcurrency = first.Work.PeakConcurrency; + var workerAttempts = 0; + var peakConcurrency = 0; var verificationRuns = 0; var repairCreated = false; - if (first.Work.Succeeded) + try { - var verification = await verifier.VerifyAsync(goalId, first.Work, cancellationToken); - verificationRuns++; - aggregate = AddVerification(aggregate, workItem.Id, verification); - if (verification.Outcome == VerificationOutcome.Failed) + var first = await ExecuteAttemptAsync(aggregate, workItem, 1, false, cancellationToken); + aggregate = first.Aggregate; + workerAttempts = 1; + peakConcurrency = first.Work.PeakConcurrency; + + if (first.Work.Succeeded) { - var consumedBudget = repairBudget with - { - ConsumedAttempts = Math.Max(repairBudget.ConsumedAttempts, 1), - ConsumedIterations = Math.Max(repairBudget.ConsumedIterations, 1), - }; - var repair = RepairPolicy.CreateRepair(verification, consumedBudget, workItem.Id); - if (repair is not null) + var verification = await verifier.VerifyAsync(goalId, first.Work, cancellationToken); + verificationRuns++; + aggregate = AddVerification(aggregate, workItem.Id, verification); + if (verification.Outcome == VerificationOutcome.Failed) { - repairCreated = true; - aggregate = AddEvent(aggregate, "repair.created", $"attempt-{repair.AttemptNumber}"); - await store.SaveAsync(aggregate, cancellationToken); - var repaired = await ExecuteAttemptAsync(aggregate, workItem, repair.AttemptNumber, true, cancellationToken); - aggregate = repaired.Aggregate; - workerAttempts++; - peakConcurrency = Math.Max(peakConcurrency, repaired.Work.PeakConcurrency); - if (repaired.Work.Succeeded) + var consumedBudget = repairBudget with + { + ConsumedAttempts = Math.Max(repairBudget.ConsumedAttempts, 1), + ConsumedIterations = Math.Max(repairBudget.ConsumedIterations, 1), + }; + var repair = RepairPolicy.CreateRepair(verification, consumedBudget, workItem.Id); + if (repair is not null) { - verification = await verifier.VerifyAsync(goalId, repaired.Work, cancellationToken); - verificationRuns++; - aggregate = AddVerification(aggregate, workItem.Id, verification); + repairCreated = true; + aggregate = AddEvent(aggregate, "repair.created", $"attempt-{repair.AttemptNumber}"); + await store.SaveAsync(aggregate, cancellationToken); + var repaired = await ExecuteAttemptAsync(aggregate, workItem, repair.AttemptNumber, true, cancellationToken); + aggregate = repaired.Aggregate; + workerAttempts++; + peakConcurrency = Math.Max(peakConcurrency, repaired.Work.PeakConcurrency); + if (repaired.Work.Succeeded) + { + verification = await verifier.VerifyAsync(goalId, repaired.Work, cancellationToken); + verificationRuns++; + aggregate = AddVerification(aggregate, workItem.Id, verification); + } } } } + + var outcome = aggregate.Evidence.Any(item => item.Kind == "verification-failed") + && !aggregate.Evidence.Any(item => item.Kind == "verification-passed" && item.Id.EndsWith($"-{verificationRuns}", StringComparison.Ordinal)) + ? GoalStatus.Failed + : aggregate.Evidence.Any(item => item.Kind == "verification-passed") ? GoalStatus.Completed : GoalStatus.Failed; + aggregate = Complete(aggregate, workItem.Id, outcome, lease.Id); + await store.SaveAsync(aggregate, cancellationToken); + + var evidence = aggregate.Evidence.Select(item => item.Uri).Distinct(StringComparer.Ordinal).ToArray(); + await learning.PublishAsync(new(goalId, aggregate.Goal.CorrelationId, outcome, $"Goal {outcome} after {workerAttempts} worker attempt(s).", evidence), cancellationToken); + return new(aggregate, workerAttempts, verificationRuns, repairCreated, peakConcurrency); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + var failure = FailureClassifier.Classify(new OperationCanceledException("Loop execution cancelled by caller."), "gsd-orchestrator.LoopCoordinator"); + logger.LogError( + "Loop execution failed for goal {GoalId} with failure class {FailureClass}: {Message}", + goalId, + failure.FailureClass, + failure.Message); + throw; } + catch (Exception ex) + { + var failure = FailureClassifier.Classify(ex, "gsd-orchestrator.LoopCoordinator"); + logger.LogError( + ex, + "Loop execution failed for goal {GoalId} with failure class {FailureClass}: {Message}", + goalId, + failure.FailureClass, + failure.Message); - var outcome = aggregate.Evidence.Any(item => item.Kind == "verification-failed") - && !aggregate.Evidence.Any(item => item.Kind == "verification-passed" && item.Id.EndsWith($"-{verificationRuns}", StringComparison.Ordinal)) - ? GoalStatus.Failed - : aggregate.Evidence.Any(item => item.Kind == "verification-passed") ? GoalStatus.Completed : GoalStatus.Failed; - aggregate = Complete(aggregate, workItem.Id, outcome, lease.Id); - await store.SaveAsync(aggregate, cancellationToken); + aggregate = await store.LoadAsync(goalId, cancellationToken) ?? aggregate; + var status = failure.FailureClass == FailureClass.Cancellation ? GoalStatus.Cancelled : GoalStatus.Failed; + var evidenceUri = $"cas://evidence/failure/{failure.FailureClass.ToString().ToLowerInvariant()}/{Guid.NewGuid():N}"; + aggregate = CompleteWithFailure(aggregate, workItem.Id, lease.Id, status, failure, evidenceUri); + await store.SaveAsync(aggregate, cancellationToken); - var evidence = aggregate.Evidence.Select(item => item.Uri).Distinct(StringComparer.Ordinal).ToArray(); - await learning.PublishAsync(new(goalId, aggregate.Goal.CorrelationId, outcome, $"Goal {outcome} after {workerAttempts} worker attempt(s).", evidence), cancellationToken); - return new(aggregate, workerAttempts, verificationRuns, repairCreated, peakConcurrency); + var evidence = aggregate.Evidence.Select(item => item.Uri).Distinct(StringComparer.Ordinal).ToArray(); + await learning.PublishAsync( + new(goalId, aggregate.Goal.CorrelationId, status, failure.Message, evidence), + cancellationToken); + return new(aggregate, workerAttempts, verificationRuns, repairCreated, peakConcurrency); + } } private async Task<(GoalAggregate Aggregate, LoopWorkResult Work)> ExecuteAttemptAsync( @@ -215,6 +254,46 @@ private static GoalAggregate Complete(GoalAggregate aggregate, string workItemId }; } + private static GoalAggregate CompleteWithFailure( + GoalAggregate aggregate, + string workItemId, + string leaseId, + GoalStatus status, + FailureState failure, + string evidenceUri) + { + var now = DateTimeOffset.UtcNow; + var eventPayload = JsonSerializer.Serialize(new + { + failureClass = failure.FailureClass.ToString(), + component = failure.Component, + message = failure.Message, + retryable = failure.Retryable, + exceptionType = failure.ExceptionType, + causeChain = failure.CauseChain, + retryAfterSeconds = failure.RetryAfterSeconds, + }); + + return aggregate with + { + Goal = aggregate.Goal with { Status = status }, + WorkItems = aggregate.WorkItems.Select(item => item.Id == workItemId + ? item with { Status = status == GoalStatus.Cancelled ? WorkItemStatus.Cancelled : WorkItemStatus.Failed } + : item).ToArray(), + Leases = aggregate.Leases.Where(item => item.Id != leaseId).ToArray(), + BudgetReservations = [], + Evidence = [.. aggregate.Evidence, new EvidenceRecord(Guid.NewGuid().ToString("N"), aggregate.Goal.Id, workItemId, "failure-state", evidenceUri)], + Transitions = [.. aggregate.Transitions, new(Guid.NewGuid().ToString("N"), aggregate.Goal.Id, aggregate.Goal.Status.ToString(), status.ToString(), status == GoalStatus.Cancelled ? GoalStopReason.Cancellation.ToString() : GoalStopReason.UnrecoverableFault.ToString(), now)], + Events = + [ + .. aggregate.Events, + new GoalEventRecord(Guid.NewGuid().ToString("N"), aggregate.Goal.Id, + aggregate.Events.Count == 0 ? 1 : aggregate.Events.Max(item => item.Sequence) + 1, + "goal.failed", eventPayload, now) + ] + }; + } + private static GoalEventRecord NewEvent(GoalAggregate aggregate, string type, string payload) => new(Guid.NewGuid().ToString("N"), aggregate.Goal.Id, aggregate.Events.Count == 0 ? 1 : aggregate.Events.Max(item => item.Sequence) + 1, diff --git a/tools/LoopPilotRunner/Program.cs b/tools/LoopPilotRunner/Program.cs index b5148f0..3adf49a 100644 --- a/tools/LoopPilotRunner/Program.cs +++ b/tools/LoopPilotRunner/Program.cs @@ -36,7 +36,7 @@ async Task RunCoordinatorScenario(string scenario, IReadOnlyList< await store.SaveAsync(Seed($"goal-{scenario}")); var learning = new CapturingLearning(); var worker = new MafProcessLoopWorker(arguments.Python, arguments.MafRoot); - var coordinator = new LoopCoordinator(store, worker, new ScriptedVerifier(results), learning); + var coordinator = new LoopCoordinator(store, worker, new ScriptedVerifier(results), learning, NullLogger.Instance); var run = await coordinator.RunAsync($"goal-{scenario}", new(true, 0, 2, 0, 2, 0, 10)); var peak = run.PeakConcurrency; var isolatedImplementation = scenario != "feature" || VerifyIsolatedWorktree();