Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 67 additions & 0 deletions src/GsdOrchestrator.Tests/FailureStateTests.cs
Original file line number Diff line number Diff line change
@@ -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);
}
27 changes: 25 additions & 2 deletions src/GsdOrchestrator.Tests/LoopCoordinatorTests.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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()
{
Expand All @@ -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<Fixture> CreateAsync(IReadOnlyList<VerificationRunResult> results)
public static async Task<Fixture> CreateAsync(IReadOnlyList<VerificationRunResult> results, ILoopWorker? worker = null)
{
var path = Path.Combine(Path.GetTempPath(), $"loop-{Guid.NewGuid():N}.db");
var store = new SqliteGoalStore(path, NullLogger<SqliteGoalStore>.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<LoopCoordinator>.Instance), learning);
}

public ValueTask DisposeAsync()
Expand All @@ -88,6 +105,12 @@ public Task<LoopWorkResult> 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<LoopWorkResult> ExecuteAsync(LoopWorkRequest request, CancellationToken cancellationToken) =>
Task.FromException<LoopWorkResult>(exception);
}

private sealed class ScriptedVerifier(IReadOnlyList<VerificationRunResult> results) : ILoopVerifier
{
private int _index;
Expand Down
61 changes: 61 additions & 0 deletions src/GsdOrchestrator/Loop/FailureState.cs
Original file line number Diff line number Diff line change
@@ -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<string>? 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<string>? BuildCauseChain(Exception exception)
{
var chain = new List<string>(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;
}
}
153 changes: 116 additions & 37 deletions src/GsdOrchestrator/Loop/LoopCoordinator.cs
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
using GsdOrchestrator.Mcp;
using GsdOrchestrator.Scheduling;
using GsdOrchestrator.Verification;
using Microsoft.Extensions.Logging;

namespace GsdOrchestrator.Loop;

Expand Down Expand Up @@ -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<LoopCoordinator> logger)
{
public async Task<LoopRunResult> RunAsync(
string goalId,
Expand All @@ -107,55 +108,93 @@ public async Task<LoopRunResult> 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);

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Keep terminal publishing failures from rewriting outcomes

When learning.PublishAsync throws after store.SaveAsync has already persisted a completed outcome, this call is still inside the broad try; the catch reloads that terminal aggregate and calls CompleteWithFailure, so a telemetry/MCP publication outage rewrites a successful goal/work item to Failed with a synthetic failure-state. Keep terminal publication failures outside the failure-mapping path or avoid overwriting already-terminal outcomes.

Useful? React with 👍 / 👎.

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(
Expand Down Expand Up @@ -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)],

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Preserve class-specific stop reasons for typed failures

For any non-cancellation typed failure, including UnauthorizedAccessException classified as Policy and TimeoutException classified as retryable Transient, this transition is recorded as UnrecoverableFault. That makes consumers of the transition stream unable to distinguish policy denials or retryable transient failures from actual unrecoverable faults despite the new FailureState classification; map the FailureClass to the corresponding stop reason/action instead of collapsing everything to unrecoverable.

Useful? React with 👍 / 👎.

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,
Expand Down
2 changes: 1 addition & 1 deletion tools/LoopPilotRunner/Program.cs
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ async Task<PilotEvidence> 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<LoopCoordinator>.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();
Expand Down
Loading