diff --git a/src/CodeyBox.Core/AgentPhase.cs b/src/CodeyBox.Core/AgentPhase.cs new file mode 100644 index 00000000..0c6875cb --- /dev/null +++ b/src/CodeyBox.Core/AgentPhase.cs @@ -0,0 +1,188 @@ +namespace CodeyBox.Core; + +/// +/// Sandbox-backed agent phase served by . +/// Replaces the old isInitial boolean: work and rework are distinct +/// values instead of two meanings of one flag. +/// +public enum AgentPhaseKind +{ + Work, + Rework, + Merge, +} + +/// +/// Outcome of a phase executed through . +/// Phases signal failure by throwing; a returned result always completed. +/// +public enum AgentPhaseOutcome +{ + Completed, +} + +/// +/// How the phase executor treats a required-build failure on the work branch. +/// Anti-corruption naming for the orchestrator-internal build policy: the +/// policy itself still lives with the build gate, this enum only carries the +/// caller's choice across the seam. Mapped 1:1 at the seam edge; the mapping +/// is exhaustive so a new value fails loudly instead of silently degrading. +/// +public enum AgentPhaseBuildPolicy +{ + Terminal, + DeferToAuditLoop, +} + +/// +/// How the phase executor treats a clean-exit-but-empty diff on a rework turn. +/// Anti-corruption naming for the orchestrator-internal handling: mapped 1:1 +/// at the seam edge with an exhaustive mapping. +/// +public enum AgentPhaseReworkNoDiffHandling +{ + TerminalError, + AuditEmptyRework, +} + +/// +/// Agent route declared for a phase execution: the runner kind plus the +/// model and reasoning mode the turn runs under. Derived from the request's +/// runner and work item (see ), so +/// a request can never declare a route that contradicts what will run. +/// +public sealed record AgentPhaseAgentRoute( + AgentKind Kind, + string? ModelId, + string? ReasoningMode); + +/// +/// One sandbox-backed agent phase for one work item, shaped as a request +/// object instead of a positional parameter list. Every field the previous +/// positional phase signatures carried is present here; cancellation tokens +/// stay method parameters of +/// because they are ambient, not phase state. +/// +public sealed record AgentPhaseRequest +{ + /// Work item under execution. Carries resume state + /// (PreemptCheckpoint, AgentTurnResumeCheckpoint, + /// AgentTurnRecoveryLease) and the model/reasoning mode. + public required WorkItem Item { get; init; } + + /// Project the work item is bound to. + public required Project Project { get; init; } + + /// Which phase to run. Replaces the old isInitial bool. + public required AgentPhaseKind Phase { get; init; } + + /// Host-side repository id the sandbox clones from. + public required string RepositoryId { get; init; } + + /// Branch the work branch is created from (work) or merged into (merge). + public required string BaseBranch { get; init; } + + /// Branch the agent works on (work/rework) or merges (merge). + public required string Branch { get; init; } + + /// Exact runner instance to invoke, resolved by the caller + /// (including quota-fallback swaps). Never re-resolved from the route: + /// the executor runs this instance. + public required IAgentRunner Runner { get; init; } + + /// Prompt for work/rework turns. Null for merge, which builds + /// its own resolution prompt. + public string? Prompt { get; init; } + + /// Sandbox network profile for the phase. + public string? NetworkProfile { get; init; } + + /// Sandbox flavour for the phase. + public SandboxProfileFlavor SandboxFlavor { get; init; } = SandboxProfileFlavor.Headless; + + /// Required-build failure policy for work/rework. Null for merge, + /// which does not consult the work-phase build gate. + public AgentPhaseBuildPolicy? BuildPolicy { get; init; } + + /// Audit/rework iteration number. Null for the initial work turn. + public int? Iteration { get; init; } + + /// Auditors composing the pre-emptive self-review turn. + /// Only consulted by the work phase. + public IReadOnlyList? AuditorsForPreemptiveSelfReview { get; init; } + + /// Empty-diff handling for rework turns. + public AgentPhaseReworkNoDiffHandling ReworkNoDiffHandling { get; init; } = + AgentPhaseReworkNoDiffHandling.TerminalError; + + /// Pre-turn source commit for durable-turn resume. Null when the + /// item carries no git checkpoint; the executor resolves it then. + public string? ResumePreTurnCommitSha { get; init; } + + /// Skips the no-changes circuit-breaker contribution for this turn. + public bool SuppressNoChangesBreaker { get; init; } + + /// + /// Declared agent route for this turn. Always derived from + /// and so it round-trips + /// consistently and can never contradict the invocation. + /// + public AgentPhaseAgentRoute AgentRoute => + new(Runner.Kind, Item.ModelId, Item.ReasoningMode); +} + +/// +/// Result of one phase executed through . +/// Fields a phase does not produce stay at their empty value: agent phases +/// produce no audit findings (empty list), and cost is recorded directly to +/// the cost store rather than returned (null usage). +/// +public sealed record AgentPhaseResult +{ + /// Phase that produced this result. + public required AgentPhaseKind Phase { get; init; } + + /// How the phase finished. Failures throw instead of returning. + public required AgentPhaseOutcome Outcome { get; init; } + + /// + /// Resulting commit: the work-branch tip after work/rework, the merge + /// commit after merge. + /// + public string? ResultingCommitSha { get; init; } + + /// Agent stdout for post-phase processing (question parsing, PR text). + public string? AgentStdout { get; init; } + + /// Findings the phase produced. Empty for work/rework/merge. + public IReadOnlyList Findings { get; init; } = []; + + /// Usage the phase produced. Null: phases record cost directly + /// to the cost store instead of returning it. + public WorkItemUsageSummary? Usage { get; init; } + + /// File name of the captured agent stream, when stream capture + /// was enabled for the turn. Null otherwise. + public string? AgentStreamFileName { get; init; } +} + +/// +/// Executes one sandbox-backed agent phase of one work item. The seam +/// between pipeline orchestration and phase machinery: tests substitute a +/// double here to exercise orchestration without a sandbox provider, and +/// alternative implementations can relocate phase execution without +/// touching the pipeline state machine. +/// +public interface IAgentPhaseExecutor +{ + /// + /// Runs to completion. Returns the phase + /// result; throws on phase failure. cancels the + /// phase; triggers the + /// checkpoint-and-preserve path. + /// + Task ExecuteAsync( + AgentPhaseRequest request, + CancellationToken ct, + CancellationToken hostShutdownToken); +} diff --git a/src/CodeyBox.Orchestrator/PipelineRunner.AgentPhase.cs b/src/CodeyBox.Orchestrator/PipelineRunner.AgentPhase.cs new file mode 100644 index 00000000..3be0c898 --- /dev/null +++ b/src/CodeyBox.Orchestrator/PipelineRunner.AgentPhase.cs @@ -0,0 +1,137 @@ +using CodeyBox.Core; + +namespace CodeyBox.Orchestrator; + +public sealed partial class PipelineRunner +{ + private readonly IAgentPhaseExecutor _phaseExecutor; + + /// + /// The phase-execution seam. Production runs phases in-process through + /// ; tests substitute a double to drive + /// orchestration without a sandbox provider. + /// + internal IAgentPhaseExecutor PhaseExecutor => _phaseExecutor; + + private static RequiredBuildPolicy ToRequiredBuildPolicy(AgentPhaseBuildPolicy policy) => + policy switch + { + AgentPhaseBuildPolicy.Terminal => RequiredBuildPolicy.Terminal, + AgentPhaseBuildPolicy.DeferToAuditLoop => RequiredBuildPolicy.DeferToAuditLoop, + _ => throw new ArgumentOutOfRangeException(nameof(policy), policy, "Unknown agent-phase build policy."), + }; + + private static ReworkNoDiffHandling ToReworkNoDiffHandling(AgentPhaseReworkNoDiffHandling handling) => + handling switch + { + AgentPhaseReworkNoDiffHandling.TerminalError => ReworkNoDiffHandling.TerminalError, + AgentPhaseReworkNoDiffHandling.AuditEmptyRework => ReworkNoDiffHandling.AuditEmptyRework, + _ => throw new ArgumentOutOfRangeException(nameof(handling), handling, "Unknown agent-phase rework no-diff handling."), + }; + + private async Task ExecuteWorkReworkPhaseAsync( + AgentPhaseRequest request, + bool isInitial, + CancellationToken ct, + CancellationToken hostShutdownToken) + { + if (request.Phase is not (AgentPhaseKind.Work or AgentPhaseKind.Rework)) + throw new ArgumentException( + $"Work/rework execution requires phase Work or Rework, got {request.Phase}.", + nameof(request)); + var prompt = request.Prompt + ?? throw new ArgumentException("Prompt is required for work/rework phases.", nameof(request)); + var buildPolicy = request.BuildPolicy + ?? throw new ArgumentException("BuildPolicy is required for work/rework phases.", nameof(request)); + Validation.ValidateBranchName(request.BaseBranch, nameof(request)); + Validation.ValidateBranchName(request.Branch, nameof(request)); + + var (stdout, streamFileName) = await RunAgentPhaseAsync( + request.Item, + request.Runner, + request.RepositoryId, + request.BaseBranch, + request.Branch, + prompt, + isInitial, + request.NetworkProfile, + request.SandboxFlavor, + request.Project, + ct, + hostShutdownToken, + ToRequiredBuildPolicy(buildPolicy), + request.Iteration, + request.AuditorsForPreemptiveSelfReview, + ToReworkNoDiffHandling(request.ReworkNoDiffHandling), + request.ResumePreTurnCommitSha, + request.SuppressNoChangesBreaker); + + var resultingSha = await _gitHost.ResolveCommitAsync(request.RepositoryId, request.Branch, ct); + return new AgentPhaseResult + { + Phase = request.Phase, + Outcome = AgentPhaseOutcome.Completed, + ResultingCommitSha = resultingSha, + AgentStdout = stdout, + AgentStreamFileName = streamFileName, + }; + } + + private async Task ExecuteMergePhaseAsync( + AgentPhaseRequest request, + CancellationToken ct, + CancellationToken hostShutdownToken) + { + if (request.Phase != AgentPhaseKind.Merge) + throw new ArgumentException( + $"Merge execution requires phase Merge, got {request.Phase}.", + nameof(request)); + Validation.ValidateBranchName(request.BaseBranch, nameof(request)); + Validation.ValidateBranchName(request.Branch, nameof(request)); + + var (mergeSha, stdout) = await RunAgentMergePhaseAsync( + request.Item, + request.Runner, + request.RepositoryId, + request.BaseBranch, + request.Branch, + request.NetworkProfile, + request.Project, + ct, + hostShutdownToken); + return new AgentPhaseResult + { + Phase = request.Phase, + Outcome = AgentPhaseOutcome.Completed, + ResultingCommitSha = mergeSha, + AgentStdout = stdout, + }; + } + + /// + /// In-process : exactly the current + /// behaviour, invoked behind the seam. Work maps to the initial + /// phase, rework to the non-initial phase; merge keeps its own path. + /// + private sealed class AgentPhaseExecutor(PipelineRunner runner) : IAgentPhaseExecutor + { + public Task ExecuteAsync( + AgentPhaseRequest request, + CancellationToken ct, + CancellationToken hostShutdownToken) + { + ArgumentNullException.ThrowIfNull(request); + return request.Phase switch + { + AgentPhaseKind.Work => runner.ExecuteWorkReworkPhaseAsync( + request, isInitial: true, ct, hostShutdownToken), + AgentPhaseKind.Rework => runner.ExecuteWorkReworkPhaseAsync( + request, isInitial: false, ct, hostShutdownToken), + AgentPhaseKind.Merge => runner.ExecuteMergePhaseAsync( + request, ct, hostShutdownToken), + _ => throw new ArgumentOutOfRangeException( + nameof(request), request.Phase, "Unknown agent phase."), + }; + } + } +} diff --git a/src/CodeyBox.Orchestrator/PipelineRunner.cs b/src/CodeyBox.Orchestrator/PipelineRunner.cs index 521242b1..e1156dd1 100644 --- a/src/CodeyBox.Orchestrator/PipelineRunner.cs +++ b/src/CodeyBox.Orchestrator/PipelineRunner.cs @@ -341,7 +341,11 @@ public PipelineRunner( // Optional best-effort exporter that propagates a completed item's test // cases to JobTrack. Null disables propagation; when wired it self-gates // on each project's JobTrackExport.Enabled opt-in. - IJobTrackTestCaseExporter? jobTrackExporter = null) + IJobTrackTestCaseExporter? jobTrackExporter = null, + // Phase-execution seam. Null selects the in-process implementation + // (current behaviour). Tests inject a double to drive orchestration + // without a sandbox provider. + IAgentPhaseExecutor? phaseExecutor = null) { _sandboxes = sandboxes; _gitHost = gitHost; @@ -417,6 +421,7 @@ public PipelineRunner( _testCaseStore = testCaseStore; _e2eReplayGate = e2eReplayGate; _jobTrackExporter = jobTrackExporter; + _phaseExecutor = phaseExecutor ?? new AgentPhaseExecutor(this); _mergeScopeResolver = mergeScopeResolver ?? NullMergeScopeResolver.Instance; _availability = availability; // Prefer the DI-injected handler when supplied: keeps the registry @@ -2670,23 +2675,31 @@ await _store.RecordIterationDispatchAsync( { workAgentStdout = await InvokeAgentWithQuotaFallbackAsync(item, project, "work", iteration: null, async (runner, trialItem, attemptCt) => - await RunWithStuckProbeAsync(trialItem, project, runner.Kind, "work", workPhase, ct, phaseCt => - RunAgentPhaseAsync(trialItem, runner, repoId, baseBranch, workBranch, - BuildInitialWorkPrompt( + await RunWithStuckProbeAsync(trialItem, project, runner.Kind, "work", workPhase, ct, async phaseCt => + { + var phaseResult = await PhaseExecutor.ExecuteAsync(new AgentPhaseRequest + { + Item = trialItem, + Project = project, + Phase = AgentPhaseKind.Work, + RepositoryId = repoId, + BaseBranch = baseBranch, + Branch = workBranch, + Runner = runner, + Prompt = BuildInitialWorkPrompt( trialItem.Prompt, project.AllowAgentQuestions, auditors, selfReviewChecklistEnabled, ApprovedPlanForImplementation(trialItem, planningLifecycleRequiredAtEntry)), - isInitial: true, - networkProfile: sandboxTarget.NetworkProfile, - sandboxFlavor: sandboxTarget.Flavor, - project: project, - phaseCt, - hostShutdownToken, - buildFailurePolicy: RequiredBuildPolicy.Terminal, - auditorsForPreemptiveSelfReview: auditors, - resumePreTurnCommitSha: resumePreTurnCommitSha), + NetworkProfile = sandboxTarget.NetworkProfile, + SandboxFlavor = sandboxTarget.Flavor, + BuildPolicy = AgentPhaseBuildPolicy.Terminal, + AuditorsForPreemptiveSelfReview = auditors, + ResumePreTurnCommitSha = resumePreTurnCommitSha, + }, phaseCt, hostShutdownToken); + return phaseResult.AgentStdout; + }, workToken: attemptCt), ct, phaseCancellation: workPhase, @@ -2738,22 +2751,31 @@ await PublishIterationCompletedAsync(item, project, IterationPhase.Work, AuditPr reworkStdout = await InvokeAgentWithQuotaFallbackAsync(item, project, "rework", resumeIteration, async (runner, trialItem, attemptCt) => await RunWithStuckProbeAsync(trialItem, project, runner.Kind, "rework", reworkPhase, ct, - phaseCt => RunAgentPhaseAsync(trialItem, runner, repoId, baseBranch, workBranch, - trialItem.PreemptCheckpoint is { } checkpointRef - ? BuildInterruptedReworkResumePrompt(trialItem.Prompt, checkpointRef) - : trialItem.Prompt, - isInitial: false, - networkProfile: sandboxTarget.NetworkProfile, - sandboxFlavor: sandboxTarget.Flavor, - project: project, - phaseCt, - hostShutdownToken, - // The audit loop runs immediately after this resume-rework - // path, so a non-compiling tree is re-detected by the audit - // build gate and folded into the iteration's findings. - buildFailurePolicy: RequiredBuildPolicy.DeferToAuditLoop, - iteration: resumeIteration, - resumePreTurnCommitSha: resumePreTurnCommitSha), + async phaseCt => + { + var phaseResult = await PhaseExecutor.ExecuteAsync(new AgentPhaseRequest + { + Item = trialItem, + Project = project, + Phase = AgentPhaseKind.Rework, + RepositoryId = repoId, + BaseBranch = baseBranch, + Branch = workBranch, + Runner = runner, + Prompt = trialItem.PreemptCheckpoint is { } checkpointRef + ? BuildInterruptedReworkResumePrompt(trialItem.Prompt, checkpointRef) + : trialItem.Prompt, + NetworkProfile = sandboxTarget.NetworkProfile, + SandboxFlavor = sandboxTarget.Flavor, + // The audit loop runs immediately after this resume-rework + // path, so a non-compiling tree is re-detected by the audit + // build gate and folded into the iteration's findings. + BuildPolicy = AgentPhaseBuildPolicy.DeferToAuditLoop, + Iteration = resumeIteration, + ResumePreTurnCommitSha = resumePreTurnCommitSha, + }, phaseCt, hostShutdownToken); + return phaseResult.AgentStdout; + }, workToken: attemptCt), ct, phaseCancellation: reworkPhase, @@ -2924,12 +2946,23 @@ await EnsureCurrentRealAuditPassBeforeMergeAsync( { return await InvokeAgentWithQuotaFallbackAsync(item, project, "merge", iteration: null, async (runner, trialItem, attemptCt) => - await RunWithStuckProbeAsync(trialItem, project, runner.Kind, "merge", mergePhase, phaseCt, mergeCt => - RunAgentMergePhaseAsync(trialItem, runner, repoId, baseBranch, workBranch, - networkProfile: project.NetworkProfiles.Merge, - project: project, - mergeCt, - hostShutdownToken), + await RunWithStuckProbeAsync(trialItem, project, runner.Kind, "merge", mergePhase, phaseCt, async mergeCt => + { + var phaseResult = await PhaseExecutor.ExecuteAsync(new AgentPhaseRequest + { + Item = trialItem, + Project = project, + Phase = AgentPhaseKind.Merge, + RepositoryId = repoId, + BaseBranch = baseBranch, + Branch = workBranch, + Runner = runner, + NetworkProfile = project.NetworkProfiles.Merge, + }, mergeCt, hostShutdownToken); + return (phaseResult.ResultingCommitSha + ?? throw new InvalidOperationException("Merge phase completed without a merge commit."), + phaseResult.AgentStdout); + }, workToken: attemptCt), phaseCt, phaseCancellation: mergePhase, @@ -4922,9 +4955,10 @@ private async Task ResolveAgentTurnPreTurnCommitAsync( /// branch is created from . On rework calls /// the branch is checked out as-is (with the work-phase commits already /// on it) and the agent stacks new commits on top. - /// Returns the agent's stdout for post-phase processing (e.g. question parsing). + /// Returns the agent's stdout for post-phase processing (e.g. question parsing) + /// plus the captured agent-stream file name, if stream capture was enabled. /// - private async Task RunAgentPhaseAsync( + private async Task<(string? Stdout, string? StreamFileName)> RunAgentPhaseAsync( WorkItem item, IAgentRunner runner, string repoId, @@ -5312,6 +5346,7 @@ await MigrateAndRemoveLegacyScratchpadArchiveAsync( var streamCapture = (_agentStreams is not null && _agentStreams.Options.Enabled) ? await BeginAgentStreamCaptureAsync(item.Id, agentPhase, iteration ?? 1, ct) : null; + var capturedStreamFileName = streamCapture?.FileName; var stdoutCallback = BuildStdoutCallback(item.Id, agentPhase, streamCapture); var supervision = await StartAgentSupervisionSessionAsync( item.Id, @@ -5976,7 +6011,7 @@ agentResult with await _requiredBuildGate.EnforceForWorkPhaseAsync(item, project, repoId, baseBranch, branch, agentPhase, buildFailurePolicy, ct); phaseSucceeded = true; - return agentResult.Stdout; + return (agentResult.Stdout, capturedStreamFileName); } var buildOutcome = RequiredBuildWorkPhaseOutcome.PassedOrSkipped; @@ -5987,7 +6022,7 @@ agentResult with } if (buildOutcome == RequiredBuildWorkPhaseOutcome.DeferredFailure) - return agentResult.Stdout; + return (agentResult.Stdout, capturedStreamFileName); // Feed the no-changes circuit breaker: a clean-exit-but-no-diff // outcome is the silent-failure signature an agent exhibits when @@ -6102,7 +6137,7 @@ await TryRunPreemptiveSelfReviewTurnAsync( await _requiredBuildGate.EnforceForWorkPhaseAsync(item, project, repoId, baseBranch, branch, agentPhase, buildFailurePolicy, ct); phaseSucceeded = true; - return agentResult.Stdout; + return (agentResult.Stdout, capturedStreamFileName); } catch (AgentResumePreparationUnavailableException ex) { @@ -8602,20 +8637,29 @@ await _webhooks.PublishAsync(new WebhookEvent await InvokeAgentWithQuotaFallbackAsync(item, project, "rework", iteration: null, async (workerRunner, trialItem, attemptCt) => await RunWithStuckProbeAsync(trialItem, project, workerRunner.Kind, "rework", reworkPhase, ct, - phaseCt => RunAgentPhaseAsync(trialItem, workerRunner, repoId, baseBranch, workBranch, - reworkPrompt, isInitial: false, - networkProfile: sandboxTarget.NetworkProfile, - sandboxFlavor: sandboxTarget.Flavor, - project: project, - phaseCt, - hostShutdownToken, - // Post-act rework is followed by another check-verdict iteration, - // NOT a build-gated audit iteration. A non-compiling tree here will - // not be re-surfaced by any subsequent gate, so a build failure - // produced by this rework must terminal-fail the item rather than - // silently slip toward the merge / merged path. - buildFailurePolicy: RequiredBuildPolicy.Terminal, - iteration: null), + async phaseCt => + { + var phaseResult = await PhaseExecutor.ExecuteAsync(new AgentPhaseRequest + { + Item = trialItem, + Project = project, + Phase = AgentPhaseKind.Rework, + RepositoryId = repoId, + BaseBranch = baseBranch, + Branch = workBranch, + Runner = workerRunner, + Prompt = reworkPrompt, + NetworkProfile = sandboxTarget.NetworkProfile, + SandboxFlavor = sandboxTarget.Flavor, + // Post-act rework is followed by another check-verdict iteration, + // NOT a build-gated audit iteration. A non-compiling tree here will + // not be re-surfaced by any subsequent gate, so a build failure + // produced by this rework must terminal-fail the item rather than + // silently slip toward the merge / merged path. + BuildPolicy = AgentPhaseBuildPolicy.Terminal, + }, phaseCt, hostShutdownToken); + return phaseResult.AgentStdout; + }, workToken: attemptCt), ct, phaseCancellation: reworkPhase, @@ -11454,27 +11498,37 @@ await _store.RecordIterationDispatchAsync( return await InvokeAgentWithQuotaFallbackAsync(item, project, "rework", iteration: reworkIterationNumber, async (workerRunner, trialItem, attemptCt) => await RunWithStuckProbeAsync(trialItem, project, workerRunner.Kind, "rework", reworkPhase, ct, - phaseCt => RunAgentPhaseAsync(trialItem, workerRunner, repoId, baseBranch, workBranch, - prompt, isInitial: false, - networkProfile: sandboxTarget.NetworkProfile, - sandboxFlavor: sandboxTarget.Flavor, - project: project, - phaseCt, - hostShutdownToken, - // Audit-driven rework: the next iteration of the audit/rework loop - // re-runs the build gate via RunForAuditAsync, which surfaces the - // failure as a blocking finding. Terminal-failing here would defeat - // the loop's purpose of converging on a fix within the audit budget. - buildFailurePolicy: RequiredBuildPolicy.DeferToAuditLoop, - iteration: reworkIterationNumber, - reworkNoDiffHandling: ReworkNoDiffHandling.AuditEmptyRework, - // With zero blocking findings there is nothing for - // the agent to change, so an empty diff is the - // correct outcome — not a silent-failure signal for - // the no-changes circuit breaker. The same holds - // when the driving verdict never completed: its - // findings may be partial. - suppressNoChangesBreaker: !auditHasBlockingFindings), + async phaseCt => + { + var phaseResult = await PhaseExecutor.ExecuteAsync(new AgentPhaseRequest + { + Item = trialItem, + Project = project, + Phase = AgentPhaseKind.Rework, + RepositoryId = repoId, + BaseBranch = baseBranch, + Branch = workBranch, + Runner = workerRunner, + Prompt = prompt, + NetworkProfile = sandboxTarget.NetworkProfile, + SandboxFlavor = sandboxTarget.Flavor, + // Audit-driven rework: the next iteration of the audit/rework loop + // re-runs the build gate via RunForAuditAsync, which surfaces the + // failure as a blocking finding. Terminal-failing here would defeat + // the loop's purpose of converging on a fix within the audit budget. + BuildPolicy = AgentPhaseBuildPolicy.DeferToAuditLoop, + Iteration = reworkIterationNumber, + ReworkNoDiffHandling = AgentPhaseReworkNoDiffHandling.AuditEmptyRework, + // With zero blocking findings there is nothing for + // the agent to change, so an empty diff is the + // correct outcome — not a silent-failure signal for + // the no-changes circuit breaker. The same holds + // when the driving verdict never completed: its + // findings may be partial. + SuppressNoChangesBreaker = !auditHasBlockingFindings, + }, phaseCt, hostShutdownToken); + return phaseResult.AgentStdout; + }, workToken: attemptCt), ct, phaseCancellation: reworkPhase, diff --git a/tests/CodeyBox.Tests/AgentPhaseExecutorTests.cs b/tests/CodeyBox.Tests/AgentPhaseExecutorTests.cs new file mode 100644 index 00000000..27290d6c --- /dev/null +++ b/tests/CodeyBox.Tests/AgentPhaseExecutorTests.cs @@ -0,0 +1,583 @@ +using Microsoft.Extensions.Logging.Abstractions; +using CodeyBox.Agents; +using CodeyBox.Core; +using CodeyBox.Git; +using CodeyBox.Orchestrator; +using CodeyBox.Projects; +using CodeyBox.Sandbox; +using CodeyBox.Webhooks; + +namespace CodeyBox.Tests; + +/// +/// Field-by-field round-trip of the phase-execution seam request object: every +/// field the old positional phase signatures carried must survive the move. +/// No sandbox, git, or store involved. +/// +public sealed class AgentPhaseRequestTests +{ + [Theory] + [InlineData(AgentPhaseKind.Work)] + [InlineData(AgentPhaseKind.Rework)] + [InlineData(AgentPhaseKind.Merge)] + public void Request_RoundTripsEveryField(AgentPhaseKind phase) + { + var item = new WorkItem + { + Id = WorkItemId.New(), + ProjectId = new ProjectId("test-project"), + Title = "Seam item", + Prompt = "do the thing", + WorkBranch = "feature/seam-roundtrip", + ModelId = "model-7", + ReasoningMode = "high", + AgentTurnRecoveryLease = new SandboxRecoveryLease("prov", "sandbox-1", "tok-1"), + }; + var project = new Project + { + Id = new ProjectId("test-project"), + DisplayName = "Test Project", + RepositoryUrl = "https://example.invalid/seed.git", + DefaultBaseBranch = "main", + DefaultAgent = AgentKind.Claude, + }; + var runner = new StubAgentRunner(new AgentKind("stub-kind")); + var auditor = new StubAuditor(); + + var request = new AgentPhaseRequest + { + Item = item, + Project = project, + Phase = phase, + RepositoryId = "repo-123", + BaseBranch = "main", + Branch = "feature/seam-roundtrip", + Runner = runner, + Prompt = "do the thing", + NetworkProfile = "test-net", + SandboxFlavor = SandboxProfileFlavor.Graphical, + BuildPolicy = AgentPhaseBuildPolicy.DeferToAuditLoop, + Iteration = 3, + AuditorsForPreemptiveSelfReview = [auditor], + ReworkNoDiffHandling = AgentPhaseReworkNoDiffHandling.AuditEmptyRework, + ResumePreTurnCommitSha = "0123456789abcdef0123456789abcdef01234567", + SuppressNoChangesBreaker = true, + }; + + Assert.Same(item, request.Item); + Assert.Same(project, request.Project); + Assert.Equal(phase, request.Phase); + Assert.Equal("repo-123", request.RepositoryId); + Assert.Equal("main", request.BaseBranch); + Assert.Equal("feature/seam-roundtrip", request.Branch); + Assert.Same(runner, request.Runner); + Assert.Equal("do the thing", request.Prompt); + Assert.Equal("test-net", request.NetworkProfile); + Assert.Equal(SandboxProfileFlavor.Graphical, request.SandboxFlavor); + Assert.Equal(AgentPhaseBuildPolicy.DeferToAuditLoop, request.BuildPolicy); + Assert.Equal(3, request.Iteration); + Assert.Same(auditor, Assert.Single(request.AuditorsForPreemptiveSelfReview!)); + Assert.Equal(AgentPhaseReworkNoDiffHandling.AuditEmptyRework, request.ReworkNoDiffHandling); + Assert.Equal("0123456789abcdef0123456789abcdef01234567", request.ResumePreTurnCommitSha); + Assert.True(request.SuppressNoChangesBreaker); + + // Resume state travels on the item itself. + Assert.Equal("tok-1", request.Item.AgentTurnRecoveryLease!.Token); + + // The declared route is derived from the runner + item, so it can + // never contradict the invocation the executor will perform. + Assert.Equal(new AgentKind("stub-kind"), request.AgentRoute.Kind); + Assert.Equal("model-7", request.AgentRoute.ModelId); + Assert.Equal("high", request.AgentRoute.ReasoningMode); + } + + [Fact] + public void Request_DefaultsMatchLegacyCallConventions() + { + var item = new WorkItem + { + Id = WorkItemId.New(), + ProjectId = new ProjectId("test-project"), + Title = "Seam item", + Prompt = "do the thing", + WorkBranch = "feature/seam-defaults", + }; + var project = new Project + { + Id = new ProjectId("test-project"), + DisplayName = "Test Project", + RepositoryUrl = "https://example.invalid/seed.git", + }; + var request = new AgentPhaseRequest + { + Item = item, + Project = project, + Phase = AgentPhaseKind.Merge, + RepositoryId = "repo-123", + BaseBranch = "main", + Branch = "feature/seam-defaults", + Runner = new StubAgentRunner(AgentKind.Claude), + }; + + Assert.Equal(SandboxProfileFlavor.Headless, request.SandboxFlavor); + Assert.Equal(AgentPhaseReworkNoDiffHandling.TerminalError, request.ReworkNoDiffHandling); + Assert.Null(request.Prompt); + Assert.Null(request.BuildPolicy); + Assert.Null(request.Iteration); + Assert.Null(request.AuditorsForPreemptiveSelfReview); + Assert.Null(request.ResumePreTurnCommitSha); + Assert.False(request.SuppressNoChangesBreaker); + + var result = new AgentPhaseResult + { + Phase = AgentPhaseKind.Merge, + Outcome = AgentPhaseOutcome.Completed, + }; + Assert.Empty(result.Findings); + Assert.Null(result.Usage); + Assert.Null(result.ResultingCommitSha); + Assert.Null(result.AgentStdout); + Assert.Null(result.AgentStreamFileName); + } + + private sealed class StubAgentRunner(AgentKind kind) : IAgentRunner + { + public AgentKind Kind { get; } = kind; + + public Task RunAsync( + ISandbox sandbox, string workingDirectory, string prompt, AgentCredential? credential, + string? modelId = null, string? reasoningMode = null, CancellationToken ct = default, + Action? stdoutChunkCallback = null, bool captureStructuredStream = false) + => throw new NotSupportedException("Round-trip test never runs the agent."); + } + + private sealed class StubAuditor : IAuditor + { + public string Name => "stub-auditor"; + public string Kind => "tool"; + public AuditCapabilities Required => AuditCapabilities.None; + + public Task RunAsync( + ISandbox sandbox, string workingDirectory, AuditContext context, CancellationToken ct = default) + => throw new NotSupportedException("Round-trip test never runs auditors."); + } +} + +/// +/// The phase-execution seam exercised end to end: substitution with a test +/// double (no sandbox provider), parity of work/rework/merge through the +/// interface against the real in-process implementation, and resume-state +/// honouring through the interface. +/// +[Collection("Pipeline integration")] +public sealed class AgentPhaseExecutorTests : IDisposable +{ + private readonly string _workspace; + + public AgentPhaseExecutorTests() + => _workspace = Directory.CreateTempSubdirectory("codeybox-agent-phase-").FullName; + + public void Dispose() + { + try + { + Directory.Delete(_workspace, recursive: true); + } + catch + { + // Best-effort cleanup of the temp workspace. + } + } + + [Fact] + public async Task PipelineRuns_WorkAndMergeThroughDouble_WithNoSandboxProvider() + { + var seed = await TestSupport.CreateSeedRepoAsync(_workspace); + var gitRoot = Path.Combine(_workspace, "repos-" + Guid.NewGuid().ToString("N")[..8]); + var stateDb = Path.Combine(_workspace, "state-" + Guid.NewGuid().ToString("N")[..8] + ".db"); + + using var store = new SqliteWorkItemStore(stateDb); + var gitHost = new LocalGitHost( + new LocalGitHostOptions { RootDirectory = gitRoot }, + NullLogger.Instance); + var sandboxes = new ThrowingSandboxProvider(); + var agent = new ScriptedAgent([MergeStrategy.RealMerge]); + var webhooks = new NullWebhookDispatcher(); + var project = SeamProject(seed); + var projects = new InMemoryProjectRepository(project); + var composer = new ProjectAuditorComposer(new ScriptedAuditorCatalog([])); + var terminalTransitions = TestSupport.CreateTerminalTransition(store, webhooks, projects); + var fake = new GitBackedFakeExecutor(gitHost, _workspace); + + var pipeline = new PipelineRunner( + sandboxes, gitHost, new AgentRegistry([agent]), new StaticCredentialProvider(), + new InMemoryPullRequestService(), projects, new TestUpstreamFactory(), composer, + store, webhooks, + new PipelineOptions { SandboxImageReference = "ignored", AgentAllowedHosts = [] }, + NullLogger.Instance, + requiredBuildVerifier: TestRequiredBuildVerifier.NotApplicable, + terminalTransitions: terminalTransitions, + terminalRevisionBuilder: terminalTransitions, + phaseExecutor: fake); + + var item = new WorkItem + { + Id = WorkItemId.New(), + ProjectId = project.Id, + Title = "Seam double item", + Prompt = "do the thing", + WorkBranch = "feature/seam-double", + }; + await store.CreateAsync(item); + await pipeline.RunAsync(item, CancellationToken.None); + + var final = await store.GetAsync(item.Id); + Assert.True(final!.State == WorkItemState.Done, + $"Expected Done but was {final!.State} (kind={final.FailureKind}): {final.LastError}"); + + // Both sandbox-backed phases went through the double, in order. + Assert.Equal( + new[] { AgentPhaseKind.Work, AgentPhaseKind.Merge }, + fake.Requests.Select(r => r.Phase).ToList()); + var workRequest = fake.Requests[0]; + Assert.Equal(item.Id, workRequest.Item.Id); + Assert.Equal("test-project", workRequest.Project.Id.Value); + Assert.Equal("main", workRequest.BaseBranch); + Assert.Equal("feature/seam-double", workRequest.Branch); + Assert.False(string.IsNullOrWhiteSpace(workRequest.Prompt)); + Assert.Equal(AgentKind.Claude, workRequest.AgentRoute.Kind); + var mergeRequest = fake.Requests[1]; + Assert.Equal(item.Id, mergeRequest.Item.Id); + + // No sandbox was ever provisioned and the registry agent never ran: + // the double owns phase execution. + Assert.Equal(0, sandboxes.CreateCalls); + Assert.Empty(agent.WorkPrompts); + + // The fake merge produced a real two-parent merge commit on main. + var bare = gitHost.GetRepoPath(fake.LastRepositoryId); + var (_, parents, _) = await TestSupport.RunGit(bare, "rev-list", "--parents", "-n", "1", fake.LastMergeSha); + Assert.Equal(3, parents.Trim().Split(' ', StringSplitOptions.RemoveEmptyEntries).Length); + } + + [Fact] + public async Task WorkReworkMerge_ThroughInterface_ProduceSameOutcomeShaAndState() + { + var seed = await TestSupport.CreateSeedRepoAsync(_workspace); + using var setup = TestSupport.BuildPipeline( + _workspace, seed, requiredBuildVerifier: TestRequiredBuildVerifier.NotApplicable); + var project = SeamProject(seed); + var agent = setup.Agent; + agent.WorkPlan.Enqueue(new FileWrite("output.txt", "v1")); + agent.WorkPlan.Enqueue(new FileWrite("rework.txt", "v2")); + agent.ResultStdout = "phase-stdout"; + + var item = new WorkItem + { + Id = WorkItemId.New(), + ProjectId = project.Id, + Title = "Seam parity item", + Prompt = "write output.txt", + WorkBranch = "feature/seam-parity", + }; + await setup.Store.CreateAsync(item); + var repoId = await setup.GitHost.EnsureRepositoryAsync(item.Id, seed, "main"); + + var workResult = await setup.Pipeline.PhaseExecutor.ExecuteAsync(new AgentPhaseRequest + { + Item = item, + Project = project, + Phase = AgentPhaseKind.Work, + RepositoryId = repoId, + BaseBranch = "main", + Branch = item.WorkBranch, + Runner = agent, + Prompt = "write output.txt", + BuildPolicy = AgentPhaseBuildPolicy.Terminal, + }, CancellationToken.None, CancellationToken.None); + + Assert.Equal(AgentPhaseKind.Work, workResult.Phase); + Assert.Equal(AgentPhaseOutcome.Completed, workResult.Outcome); + Assert.Equal("phase-stdout", workResult.AgentStdout); + var workTip = await setup.GitHost.ResolveCommitAsync(repoId, item.WorkBranch); + Assert.Equal(workTip, workResult.ResultingCommitSha); + Assert.Empty(workResult.Findings); + Assert.Null(workResult.AgentStreamFileName); + + var reworkResult = await setup.Pipeline.PhaseExecutor.ExecuteAsync(new AgentPhaseRequest + { + Item = (await setup.Store.GetAsync(item.Id))!, + Project = project, + Phase = AgentPhaseKind.Rework, + RepositoryId = repoId, + BaseBranch = "main", + Branch = item.WorkBranch, + Runner = agent, + Prompt = "address findings", + BuildPolicy = AgentPhaseBuildPolicy.DeferToAuditLoop, + Iteration = 1, + ReworkNoDiffHandling = AgentPhaseReworkNoDiffHandling.AuditEmptyRework, + SuppressNoChangesBreaker = true, + }, CancellationToken.None, CancellationToken.None); + + Assert.Equal(AgentPhaseKind.Rework, reworkResult.Phase); + Assert.Equal(AgentPhaseOutcome.Completed, reworkResult.Outcome); + Assert.Equal("phase-stdout", reworkResult.AgentStdout); + var reworkTip = await setup.GitHost.ResolveCommitAsync(repoId, item.WorkBranch); + Assert.Equal(reworkTip, reworkResult.ResultingCommitSha); + Assert.NotEqual(workTip, reworkTip); + + var bare = setup.GitHost.GetRepoPath(repoId); + var (_, tree, _) = await TestSupport.RunGit(bare, "ls-tree", "-r", item.WorkBranch, "--name-only"); + Assert.Contains("output.txt", tree); + Assert.Contains("rework.txt", tree); + + // Clean merge is host-side: the merge phase through the interface + // produces the merge commit without invoking any agent. + var mergeResult = await setup.Pipeline.PhaseExecutor.ExecuteAsync(new AgentPhaseRequest + { + Item = (await setup.Store.GetAsync(item.Id))!, + Project = project, + Phase = AgentPhaseKind.Merge, + RepositoryId = repoId, + BaseBranch = "main", + Branch = item.WorkBranch, + Runner = agent, + }, CancellationToken.None, CancellationToken.None); + + Assert.Equal(AgentPhaseKind.Merge, mergeResult.Phase); + Assert.Equal(AgentPhaseOutcome.Completed, mergeResult.Outcome); + Assert.Null(mergeResult.AgentStdout); + var baseTip = await setup.GitHost.ResolveCommitAsync(repoId, "main"); + Assert.Equal(baseTip, mergeResult.ResultingCommitSha); + var (_, mergeParents, _) = await TestSupport.RunGit(bare, "rev-list", "--parents", "-n", "1", baseTip); + Assert.Equal(3, mergeParents.Trim().Split(' ', StringSplitOptions.RemoveEmptyEntries).Length); + + // Only work + rework reached the agent; the clean merge did not. + Assert.Equal(2, agent.WorkPrompts.Count); + } + + [Fact] + public async Task ResumeState_PreemptCheckpoint_HonouredThroughInterface() + { + var seed = await TestSupport.CreateSeedRepoAsync(_workspace); + using var setup = TestSupport.BuildPipeline( + _workspace, seed, requiredBuildVerifier: TestRequiredBuildVerifier.NotApplicable); + var project = SeamProject(seed); + var agent = setup.Agent; + + var item = new WorkItem + { + Id = WorkItemId.New(), + ProjectId = project.Id, + Title = "Seam resume item", + Prompt = "write partial work", + WorkBranch = "feature/seam-resume", + }; + await setup.Store.CreateAsync(item); + // RunAsync transitions the item to Working before dispatching the + // phase; checkpoint creation requires it, so mirror that here. + item = item with { State = WorkItemState.Working }; + await setup.Store.UpdateAsync(item); + var repoId = await setup.GitHost.EnsureRepositoryAsync(item.Id, seed, "main"); + var baseTip = await setup.GitHost.ResolveCommitAsync(repoId, "main"); + + var blocking = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + agent.BeforeWorkAsync = async (sandbox, workingDirectory, ct) => + { + var write = await sandbox.ExecAsync(new SandboxExec + { + Argv = ["sh", "-c", "cat > \"$0\"", $"{workingDirectory}/partial.txt"], + Stdin = "partial", + }, ct); + if (!write.Success) + throw new InvalidOperationException("setup write failed: " + write.Stderr); + blocking.SetResult(); + await Task.Delay(Timeout.Infinite, ct); + }; + + using var hostShutdown = new CancellationTokenSource(); + var interrupted = setup.Pipeline.PhaseExecutor.ExecuteAsync(new AgentPhaseRequest + { + Item = item, + Project = project, + Phase = AgentPhaseKind.Work, + RepositoryId = repoId, + BaseBranch = "main", + Branch = item.WorkBranch, + Runner = agent, + Prompt = "write partial work", + BuildPolicy = AgentPhaseBuildPolicy.Terminal, + }, CancellationToken.None, hostShutdown.Token); + + await blocking.Task.WaitAsync(TimeSpan.FromSeconds(60)); + await hostShutdown.CancelAsync(); + await Assert.ThrowsAnyAsync(() => interrupted); + + var checkpointed = await setup.Store.GetAsync(item.Id); + Assert.False(string.IsNullOrWhiteSpace(checkpointed!.PreemptCheckpoint)); + Assert.NotNull(checkpointed.AgentTurnResumeCheckpoint); + + agent.BeforeWorkAsync = null; + agent.WorkPlan.Enqueue(new FileWrite("resumed.txt", "done")); + agent.ResultStdout = "resumed-stdout"; + + var resumed = await setup.Pipeline.PhaseExecutor.ExecuteAsync(new AgentPhaseRequest + { + Item = checkpointed, + Project = project, + Phase = AgentPhaseKind.Work, + RepositoryId = repoId, + BaseBranch = "main", + Branch = item.WorkBranch, + Runner = agent, + Prompt = "write partial work", + BuildPolicy = AgentPhaseBuildPolicy.Terminal, + ResumePreTurnCommitSha = baseTip, + }, CancellationToken.None, CancellationToken.None); + + Assert.Equal(AgentPhaseOutcome.Completed, resumed.Outcome); + Assert.Equal("resumed-stdout", resumed.AgentStdout); + var tip = await setup.GitHost.ResolveCommitAsync(repoId, item.WorkBranch); + Assert.Equal(tip, resumed.ResultingCommitSha); + + // The resumed turn restored the checkpointed tree (partial.txt) and + // stacked its own commit (resumed.txt) on top. + var bare = setup.GitHost.GetRepoPath(repoId); + var (_, tree, _) = await TestSupport.RunGit(bare, "ls-tree", "-r", item.WorkBranch, "--name-only"); + Assert.Contains("partial.txt", tree); + Assert.Contains("resumed.txt", tree); + + // The durable turn state was cleared once the resumed tree landed. + var cleared = await setup.Store.GetAsync(item.Id); + Assert.True(string.IsNullOrWhiteSpace(cleared!.PreemptCheckpoint)); + Assert.Null(cleared.AgentTurnResumeCheckpoint); + } + + private static Project SeamProject(string seedRepoUrl) => new() + { + Id = new ProjectId("test-project"), + DisplayName = "Test Project", + RepositoryUrl = seedRepoUrl, + DefaultBaseBranch = "main", + DefaultAgent = AgentKind.Claude, + Audit = new ProjectAudit + { + MaxIterations = 1, + }, + }; + + private sealed class ThrowingSandboxProvider : ISandboxProvider + { + public int CreateCalls { get; private set; } + public string Name => "throwing-test-provider"; + + public Task> ListAllManagedAsync(CancellationToken ct) + => Task.FromResult>([]); + + public Task DisposeLeakedAsync(string name, CancellationToken ct) + => Task.CompletedTask; + + public Task CreateAsync(SandboxSpec spec, CancellationToken ct = default) + { + CreateCalls++; + throw new InvalidOperationException( + "No sandbox provider registered: phases must run through the injected executor double."); + } + } + + /// + /// Executor double that performs minimal real git through the real git + /// host (temp clones, no sandbox) so downstream pipeline steps observe a + /// consistent repository. + /// + private sealed class GitBackedFakeExecutor(LocalGitHost gitHost, string workspace) : IAgentPhaseExecutor + { + public List Requests { get; } = []; + public string LastRepositoryId { get; private set; } = string.Empty; + public string LastMergeSha { get; private set; } = string.Empty; + + public async Task ExecuteAsync( + AgentPhaseRequest request, CancellationToken ct, CancellationToken hostShutdownToken) + { + Requests.Add(request); + LastRepositoryId = request.RepositoryId; + return request.Phase switch + { + AgentPhaseKind.Work => await ExecuteWorkAsync(request, ct), + AgentPhaseKind.Merge => await ExecuteMergeAsync(request, ct), + _ => throw new InvalidOperationException($"Unexpected phase {request.Phase} in double test."), + }; + } + + private async Task ExecuteWorkAsync(AgentPhaseRequest request, CancellationToken ct) + { + var clone = Directory.CreateTempSubdirectory("codeybox-fake-work-").FullName; + try + { + var bare = gitHost.GetRepoPath(request.RepositoryId); + await TestSupport.RunGit(workspace, "clone", bare, clone); + await TestSupport.RunGit(clone, "config", "user.email", "fake@test.invalid"); + await TestSupport.RunGit(clone, "config", "user.name", "Fake"); + await TestSupport.RunGit(clone, "checkout", "-b", request.Branch, $"origin/{request.BaseBranch}"); + await TestSupport.RunGit(clone, "commit", "--allow-empty", "-m", "fake work"); + await TestSupport.RunGit(clone, "push", "origin", $"HEAD:{request.Branch}"); + var sha = await gitHost.ResolveCommitAsync(request.RepositoryId, request.Branch, ct); + return new AgentPhaseResult + { + Phase = request.Phase, + Outcome = AgentPhaseOutcome.Completed, + ResultingCommitSha = sha, + AgentStdout = "fake-work-stdout", + }; + } + finally + { + try + { + Directory.Delete(clone, recursive: true); + } + catch + { + // Best-effort cleanup of the temp clone. + } + } + } + + private async Task ExecuteMergeAsync(AgentPhaseRequest request, CancellationToken ct) + { + var clone = Directory.CreateTempSubdirectory("codeybox-fake-merge-").FullName; + try + { + var bare = gitHost.GetRepoPath(request.RepositoryId); + await TestSupport.RunGit(workspace, "clone", bare, clone); + await TestSupport.RunGit(clone, "config", "user.email", "fake@test.invalid"); + await TestSupport.RunGit(clone, "config", "user.name", "Fake"); + await TestSupport.RunGit(clone, "checkout", request.BaseBranch); + await TestSupport.RunGit( + clone, "merge", "--no-ff", "-m", $"fake merge {request.Branch}", $"origin/{request.Branch}"); + await TestSupport.RunGit(clone, "push", "origin", $"HEAD:{request.BaseBranch}"); + var sha = await gitHost.ResolveCommitAsync(request.RepositoryId, request.BaseBranch, ct); + LastMergeSha = sha; + return new AgentPhaseResult + { + Phase = request.Phase, + Outcome = AgentPhaseOutcome.Completed, + ResultingCommitSha = sha, + AgentStdout = "fake-merge-stdout", + }; + } + finally + { + try + { + Directory.Delete(clone, recursive: true); + } + catch + { + // Best-effort cleanup of the temp clone. + } + } + } + } +} diff --git a/tests/CodeyBox.Tests/WorkCompleteRecoveryTests.cs b/tests/CodeyBox.Tests/WorkCompleteRecoveryTests.cs index 598b462a..98c3831e 100644 --- a/tests/CodeyBox.Tests/WorkCompleteRecoveryTests.cs +++ b/tests/CodeyBox.Tests/WorkCompleteRecoveryTests.cs @@ -387,8 +387,11 @@ public async Task Retry_StaleWorkerHeldItem_FencesAndRetries() Title = "wedged work", Prompt = "p", State = WorkItemState.Working, - StartedAt = DateTimeOffset.UtcNow.AddHours(-2), - UpdatedAt = DateTimeOffset.UtcNow.AddHours(-2), + // Frozen past the default ItemStaleTimeout (150m since the + // audit-budget ordering change, previously 75m) so the retry + // fence treats the worker-held item as stale. + StartedAt = DateTimeOffset.UtcNow.AddHours(-3), + UpdatedAt = DateTimeOffset.UtcNow.AddHours(-3), }; await factory.Store.CreateAsync(item); @@ -398,7 +401,7 @@ await registry.RegisterAsync(new WorkerRegistration WorkerId = "wedged-http-worker", HostName = "host", ProcessId = 4242, - StartedAt = DateTimeOffset.UtcNow.AddHours(-2), + StartedAt = DateTimeOffset.UtcNow.AddHours(-3), LastHeartbeatAt = DateTimeOffset.UtcNow, CurrentWorkItemId = item.Id.ToString(), });