Skip to content

Commit 7cbfde9

Browse files
OgeonX-Airoot
OgeonX-Ai
and
root
authored
Add durable goal scheduling and executable loop coordination (#9)
* fix(01-02): resume bounded failed workflow states * test(01-03): cover finite multi-repo watch state * feat(01-03): add durable finite watch coordinator * fix(01-03): poll all repositories per watch interval * test(scheduler): cover persistence leasing and outcomes * feat(scheduler): add durable SQLite goal control plane * feat(verification): enforce evidence-gated completion * feat(observability): expose complete goal projection * feat(loop): integrate scheduler workers verification and pilots * fix(pilots): record integrated source revisions * test(01-02): add failed-state recovery coverage * fix(pilots): reference publishable component revisions * fix(scheduler): release SQLite handles on Windows --------- Co-authored-by: root <root@Kimi.localdomain>
1 parent 82e313a commit 7cbfde9

33 files changed

Lines changed: 2225 additions & 76 deletions

‎GithubMCP.slnx‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,4 +3,7 @@
33
<Project Path="src/GsdOrchestrator/GsdOrchestrator.csproj" />
44
<Project Path="src/GsdOrchestrator.Tests/GsdOrchestrator.Tests.csproj" />
55
</Folder>
6+
<Folder Name="/tools/">
7+
<Project Path="tools/LoopPilotRunner/LoopPilotRunner.csproj" />
8+
</Folder>
69
</Solution>

‎src/GsdOrchestrator.Tests/FileWatchStateStoreTests.cs‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
using GsdOrchestrator.Workflows.Services;
2+
using Microsoft.Extensions.Logging.Abstractions;
3+
using Xunit;
4+
5+
namespace GsdOrchestrator.Tests;
6+
7+
public sealed class FileWatchStateStoreTests : IDisposable
8+
{
9+
private readonly string _root = Path.Combine(Path.GetTempPath(), $"watch-store-{Guid.NewGuid():N}");
10+
11+
[Fact]
12+
public async Task MarkProcessedAsync_SurvivesStoreRecreation()
13+
{
14+
var first = new FileWatchStateStore(_root, NullLogger<FileWatchStateStore>.Instance);
15+
await first.MarkProcessedAsync("org", "repo", "issue-42", CancellationToken.None);
16+
17+
var second = new FileWatchStateStore(_root, NullLogger<FileWatchStateStore>.Instance);
18+
19+
Assert.True(await second.IsProcessedAsync("org", "repo", "issue-42", CancellationToken.None));
20+
}
21+
22+
[Fact]
23+
public async Task MarkProcessedAsync_SanitizesPathAndLeavesNoTemporaryFile()
24+
{
25+
var sut = new FileWatchStateStore(_root, NullLogger<FileWatchStateStore>.Instance);
26+
27+
await sut.MarkProcessedAsync("../org", "repo/../../escape", "issue:7", CancellationToken.None);
28+
29+
var files = Directory.GetFiles(Path.Combine(_root, ".gsd", "watch"));
30+
Assert.Single(files);
31+
Assert.EndsWith(".done", files[0]);
32+
Assert.DoesNotContain("..", Path.GetFileName(files[0]));
33+
Assert.Empty(Directory.GetFiles(Path.Combine(_root, ".gsd", "watch"), "*.tmp"));
34+
}
35+
36+
public void Dispose()
37+
{
38+
if (Directory.Exists(_root)) Directory.Delete(_root, recursive: true);
39+
}
40+
}

‎src/GsdOrchestrator.Tests/GoalControlPlaneTests.cs‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
using GsdOrchestrator.Scheduling;
2+
using Microsoft.Extensions.Logging.Abstractions;
3+
using Xunit;
4+
5+
namespace GsdOrchestrator.Tests;
6+
7+
public sealed class GoalControlPlaneTests : IDisposable
8+
{
9+
private readonly string _root = Path.Combine(Path.GetTempPath(), $"control-plane-{Guid.NewGuid():N}");
10+
11+
[Fact]
12+
public async Task StartInspectCancel_PersistsExplicitLifecycle()
13+
{
14+
var (store, sut) = await Create();
15+
var aggregate = GoalFixtures.CompleteAggregate() with
16+
{
17+
Goal = GoalFixtures.CompleteAggregate().Goal with { Status = GoalStatus.Planned },
18+
Transitions = [], Events = [], Evidence = [], Leases = [], BudgetReservations = [], Attempts = [], IdempotencyKeys = []
19+
};
20+
21+
await sut.StartAsync(aggregate);
22+
Assert.Equal(GoalStatus.Running, (await sut.InspectAsync("goal-1"))!.Goal.Status);
23+
24+
await sut.CancelAsync("goal-1", "operator request", "cas://evidence/cancel");
25+
var cancelled = await store.LoadAsync("goal-1");
26+
Assert.Equal(GoalStatus.Cancelled, cancelled!.Goal.Status);
27+
Assert.Contains(cancelled.Transitions, transition => transition.To == nameof(GoalStatus.Cancelled));
28+
Assert.Contains(cancelled.Evidence, evidence => evidence.Uri == "cas://evidence/cancel");
29+
}
30+
31+
[Fact]
32+
public async Task Resume_PausedGoal_ReturnsToRunningWithoutLosingEvidence()
33+
{
34+
var (store, sut) = await Create();
35+
var aggregate = GoalFixtures.CompleteAggregate() with
36+
{
37+
Goal = GoalFixtures.CompleteAggregate().Goal with { Status = GoalStatus.Paused }
38+
};
39+
await store.SaveAsync(aggregate);
40+
41+
await sut.ResumeAsync("goal-1", "operator resume");
42+
var resumed = await sut.InspectAsync("goal-1");
43+
Assert.Equal(GoalStatus.Running, resumed!.Goal.Status);
44+
Assert.Equal(aggregate.Evidence.Count, resumed.Evidence.Count);
45+
}
46+
47+
private async Task<(SqliteGoalStore Store, GoalControlPlane Sut)> Create()
48+
{
49+
var store = new SqliteGoalStore(Path.Combine(_root, "goals.db"), NullLogger<SqliteGoalStore>.Instance);
50+
await store.InitializeAsync();
51+
return (store, new GoalControlPlane(store, NullLogger<GoalControlPlane>.Instance));
52+
}
53+
54+
public void Dispose() { if (Directory.Exists(_root)) Directory.Delete(_root, true); }
55+
}

‎src/GsdOrchestrator.Tests/GoalDecisionPolicyTests.cs‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
using GsdOrchestrator.Scheduling;
2+
using Xunit;
3+
4+
namespace GsdOrchestrator.Tests;
5+
6+
public class GoalDecisionPolicyTests
7+
{
8+
[Theory]
9+
[InlineData(FailureClass.Transient, DecisionAction.Retry)]
10+
[InlineData(FailureClass.Deterministic, DecisionAction.CreateRepair)]
11+
[InlineData(FailureClass.Policy, DecisionAction.Stop)]
12+
[InlineData(FailureClass.Cancellation, DecisionAction.Stop)]
13+
[InlineData(FailureClass.Unrecoverable, DecisionAction.Stop)]
14+
public void Decide_ClassifiesEveryFailureWithBudgetAndEvidence(FailureClass failure, DecisionAction action)
15+
{
16+
var result = GoalDecisionPolicy.DecideFailure(failure, consumedAttempts: 1, maxAttempts: 3, noProgressCount: 0, noProgressLimit: 2, ["evidence-1"]);
17+
Assert.Equal(action, result.Action);
18+
Assert.Equal(1, result.ConsumedAttempts);
19+
Assert.NotEmpty(result.Reason);
20+
Assert.Equal(["evidence-1"], result.EvidenceIds);
21+
}
22+
23+
[Fact]
24+
public void Decide_ExhaustedAttempts_StopsWithExhaustion()
25+
{
26+
var result = GoalDecisionPolicy.DecideFailure(FailureClass.Transient, 3, 3, 0, 2, ["e"]);
27+
Assert.Equal(GoalStopReason.Exhaustion, result.StopReason);
28+
}
29+
30+
[Fact]
31+
public void Decide_NoProgressLimit_StopsBeforeRetry()
32+
{
33+
var result = GoalDecisionPolicy.DecideFailure(FailureClass.Deterministic, 1, 3, 2, 2, ["e"]);
34+
Assert.Equal(GoalStopReason.NoProgress, result.StopReason);
35+
}
36+
37+
[Theory]
38+
[InlineData(GoalStopReason.Passed)]
39+
[InlineData(GoalStopReason.Exhaustion)]
40+
[InlineData(GoalStopReason.Cancellation)]
41+
[InlineData(GoalStopReason.Denial)]
42+
[InlineData(GoalStopReason.ApprovalWait)]
43+
[InlineData(GoalStopReason.Deadlock)]
44+
[InlineData(GoalStopReason.UnrecoverableFault)]
45+
[InlineData(GoalStopReason.NoProgress)]
46+
public void Stop_ExposesEveryDistinctReasonWithEvidence(GoalStopReason reason)
47+
{
48+
var result = GoalDecisionPolicy.Stop(reason, "explicit outcome", ["evidence-1"]);
49+
Assert.Equal(reason, result.StopReason);
50+
Assert.Equal(DecisionAction.Stop, result.Action);
51+
Assert.NotEmpty(result.EvidenceIds);
52+
}
53+
54+
[Fact]
55+
public void Stop_WithoutEvidence_IsRejected()
56+
{
57+
Assert.Throws<ArgumentException>(() => GoalDecisionPolicy.Stop(GoalStopReason.Deadlock, "deadlock", []));
58+
}
59+
}

‎src/GsdOrchestrator.Tests/GoalSchedulerTests.cs‎

Lines changed: 118 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,118 @@
1+
using GsdOrchestrator.Scheduling;
2+
using Microsoft.Extensions.Logging.Abstractions;
3+
using Xunit;
4+
5+
namespace GsdOrchestrator.Tests;
6+
7+
public sealed class GoalSchedulerTests : IDisposable
8+
{
9+
private readonly string _root = Path.Combine(Path.GetTempPath(), $"scheduler-{Guid.NewGuid():N}");
10+
11+
[Fact]
12+
public async Task TryAcquireLease_IncompleteDependency_IsDenied()
13+
{
14+
var store = await CreateStore();
15+
var aggregate = AggregateWithTwoItems(WorkItemStatus.Pending);
16+
await store.SaveAsync(aggregate);
17+
var scheduler = new GoalScheduler(store, NullLogger<GoalScheduler>.Instance);
18+
19+
var lease = await scheduler.TryAcquireLeaseAsync(Request("work-2"));
20+
21+
Assert.Null(lease);
22+
}
23+
24+
[Fact]
25+
public async Task TryAcquireLease_CompetingCalls_RespectsGlobalLimitAtomically()
26+
{
27+
var store = await CreateStore();
28+
await store.SaveAsync(IndependentAggregate());
29+
var scheduler = new GoalScheduler(store, NullLogger<GoalScheduler>.Instance);
30+
31+
var results = await Task.WhenAll(
32+
scheduler.TryAcquireLeaseAsync(Request("work-1", global: 1)),
33+
scheduler.TryAcquireLeaseAsync(Request("work-2", global: 1)));
34+
35+
Assert.Single(results, result => result is not null);
36+
}
37+
38+
[Theory]
39+
[InlineData(10, 1, 10)]
40+
[InlineData(10, 10, 1)]
41+
public async Task TryAcquireLease_ProviderOrRepositoryLimit_DeniesSecond(int global, int provider, int repository)
42+
{
43+
var store = await CreateStore();
44+
await store.SaveAsync(IndependentAggregate());
45+
var scheduler = new GoalScheduler(store, NullLogger<GoalScheduler>.Instance);
46+
Assert.NotNull(await scheduler.TryAcquireLeaseAsync(Request("work-1", global, provider, repository)));
47+
Assert.Null(await scheduler.TryAcquireLeaseAsync(Request("work-2", global, provider, repository)));
48+
}
49+
50+
[Fact]
51+
public async Task RecoverExpiredLeases_AllowsReacquire()
52+
{
53+
var store = await CreateStore();
54+
await store.SaveAsync(IndependentAggregate());
55+
var scheduler = new GoalScheduler(store, NullLogger<GoalScheduler>.Instance);
56+
var now = DateTimeOffset.UtcNow;
57+
Assert.NotNull(await scheduler.TryAcquireLeaseAsync(
58+
new LeaseRequest("goal-1", "work-1", "worker-1", now.AddMinutes(-2), now.AddMinutes(-1), 3, 3, 3)));
59+
60+
Assert.Equal(1, await scheduler.RecoverExpiredLeasesAsync(now));
61+
Assert.NotNull(await scheduler.TryAcquireLeaseAsync(Request("work-1")));
62+
}
63+
64+
[Fact]
65+
public async Task ReserveIdempotencyKey_RestartRejectsDuplicateEffect()
66+
{
67+
var path = Path.Combine(_root, "goals.db");
68+
var first = new SqliteGoalStore(path, NullLogger<SqliteGoalStore>.Instance);
69+
await first.InitializeAsync();
70+
await first.SaveAsync(IndependentAggregate());
71+
Assert.True(await first.TryReserveIdempotencyKeyAsync("goal-1", "work-1", "commit:abc", "commit"));
72+
73+
var restarted = new SqliteGoalStore(path, NullLogger<SqliteGoalStore>.Instance);
74+
await restarted.InitializeAsync();
75+
Assert.False(await restarted.TryReserveIdempotencyKeyAsync("goal-1", "work-1", "commit:abc", "commit"));
76+
}
77+
78+
private LeaseRequest Request(string workItemId, int global = 3, int provider = 3, int repository = 3, DateTimeOffset? expiresAt = null) =>
79+
new("goal-1", workItemId, "worker-1", DateTimeOffset.UtcNow, expiresAt ?? DateTimeOffset.UtcNow.AddMinutes(5), global, provider, repository);
80+
81+
private async Task<SqliteGoalStore> CreateStore()
82+
{
83+
var store = new SqliteGoalStore(Path.Combine(_root, "goals.db"), NullLogger<SqliteGoalStore>.Instance);
84+
await store.InitializeAsync();
85+
return store;
86+
}
87+
88+
private static GoalAggregate AggregateWithTwoItems(WorkItemStatus dependencyStatus)
89+
{
90+
var seed = GoalFixtures.CompleteAggregate();
91+
return seed with
92+
{
93+
WorkItems =
94+
[
95+
new WorkItemRecord("work-1", "goal-1", "org/repo", "local", dependencyStatus, 3, "idem-1"),
96+
new WorkItemRecord("work-2", "goal-1", "org/repo", "local", WorkItemStatus.Ready, 3, "idem-2")
97+
],
98+
Dependencies = [new DependencyRecord("goal-1", "work-2", "work-1")],
99+
Attempts = [], Leases = [], BudgetReservations = [], IdempotencyKeys = []
100+
};
101+
}
102+
103+
private static GoalAggregate IndependentAggregate()
104+
{
105+
var seed = GoalFixtures.CompleteAggregate();
106+
return seed with
107+
{
108+
WorkItems =
109+
[
110+
new WorkItemRecord("work-1", "goal-1", "org/repo", "local", WorkItemStatus.Ready, 3, "idem-1"),
111+
new WorkItemRecord("work-2", "goal-1", "org/repo", "local", WorkItemStatus.Ready, 3, "idem-2")
112+
],
113+
Dependencies = [], Attempts = [], Leases = [], BudgetReservations = [], IdempotencyKeys = []
114+
};
115+
}
116+
117+
public void Dispose() { if (Directory.Exists(_root)) Directory.Delete(_root, true); }
118+
}

‎src/GsdOrchestrator.Tests/GsdStateMachineTests.cs‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -141,6 +141,80 @@ public async Task ResumeAsync_CheckpointExists_ResumesFromSavedState()
141141
Assert.Equal(WorkflowState.Done, ctx.CurrentState);
142142
}
143143

144+
[Fact]
145+
public async Task ResumeAsync_RecoverableFailure_ReentersFailedStateWithoutReplayingHistory()
146+
{
147+
var checkpoints = Substitute.For<ICheckpointStore>();
148+
var priorTransition = new StateTransitionEvent(
149+
WorkflowState.Idle,
150+
WorkflowState.Analyzing,
151+
DateTimeOffset.UtcNow.AddMinutes(-1));
152+
var savedCtx = new GsdWorkflowContext
153+
{
154+
WorkflowId = "recoverable-wf",
155+
CurrentState = WorkflowState.Failed,
156+
FailedState = WorkflowState.Analyzing,
157+
RetryCount = 0,
158+
FailureReason = "transient failure",
159+
Issue = new IssueContext(7, "Test issue", "", [], "owner", "repo", "main"),
160+
History = [priorTransition]
161+
};
162+
checkpoints.LoadAsync(savedCtx.WorkflowId, Arg.Any<CancellationToken>()).Returns(savedCtx);
163+
checkpoints.SaveAsync(Arg.Any<GsdWorkflowContext>(), Arg.Any<CancellationToken>()).Returns(Task.CompletedTask);
164+
checkpoints.ArchiveAsync(Arg.Any<string>(), Arg.Any<CancellationToken>()).Returns(Task.CompletedTask);
165+
166+
var analyzing = MakeState(WorkflowState.Analyzing, WorkflowState.Done);
167+
var sut = BuildSut(checkpoints, [analyzing]);
168+
169+
var result = await sut.ResumeAsync(savedCtx.WorkflowId, CancellationToken.None);
170+
171+
Assert.Equal(WorkflowState.Done, result.CurrentState);
172+
Assert.Contains(result.History, transition =>
173+
transition.From == WorkflowState.Failed && transition.To == WorkflowState.Analyzing);
174+
Assert.Contains(priorTransition, result.History);
175+
await analyzing.Received(1).ExecuteAsync(
176+
Arg.Is<GsdWorkflowContext>(context => context.CurrentState == WorkflowState.Analyzing && context.RetryCount == 1),
177+
Arg.Any<CancellationToken>());
178+
}
179+
180+
[Fact]
181+
public async Task ResumeAsync_RetryExhausted_ThrowsClearError()
182+
{
183+
var checkpoints = Substitute.For<ICheckpointStore>();
184+
var savedCtx = new GsdWorkflowContext
185+
{
186+
WorkflowId = "exhausted-wf",
187+
CurrentState = WorkflowState.Failed,
188+
FailedState = WorkflowState.Analyzing,
189+
RetryCount = 1
190+
};
191+
checkpoints.LoadAsync(savedCtx.WorkflowId, Arg.Any<CancellationToken>()).Returns(savedCtx);
192+
var sut = BuildSut(checkpoints, []);
193+
194+
var error = await Assert.ThrowsAsync<InvalidOperationException>(
195+
() => sut.ResumeAsync(savedCtx.WorkflowId, CancellationToken.None));
196+
197+
Assert.Contains("retry limit", error.Message, StringComparison.OrdinalIgnoreCase);
198+
}
199+
200+
[Fact]
201+
public async Task ResumeAsync_LegacyFailedCheckpoint_ThrowsClearError()
202+
{
203+
var checkpoints = Substitute.For<ICheckpointStore>();
204+
var savedCtx = new GsdWorkflowContext
205+
{
206+
WorkflowId = "legacy-failed-wf",
207+
CurrentState = WorkflowState.Failed
208+
};
209+
checkpoints.LoadAsync(savedCtx.WorkflowId, Arg.Any<CancellationToken>()).Returns(savedCtx);
210+
var sut = BuildSut(checkpoints, []);
211+
212+
var error = await Assert.ThrowsAsync<InvalidOperationException>(
213+
() => sut.ResumeAsync(savedCtx.WorkflowId, CancellationToken.None));
214+
215+
Assert.Contains("recoverable state", error.Message, StringComparison.OrdinalIgnoreCase);
216+
}
217+
144218
[Fact]
145219
public async Task ResumeAsync_NoCheckpointExists_ThrowsInvalidOperationException()
146220
{

0 commit comments

Comments
 (0)