diff --git a/.changeset/bound-outcome-plan-drafting.md b/.changeset/bound-outcome-plan-drafting.md new file mode 100644 index 000000000..96ab6051a --- /dev/null +++ b/.changeset/bound-outcome-plan-drafting.md @@ -0,0 +1,5 @@ +--- +"agentweaver": patch +--- + +Fail coordinator runs when outcome-plan drafting cannot finish provider startup, session creation, or the model turn within a bounded deadline, and keep the displayed elapsed time advancing from the durable run start. diff --git a/apps/Agentweaver.Api/Coordinator/CoordinatorMessages.cs b/apps/Agentweaver.Api/Coordinator/CoordinatorMessages.cs index c77eaf0c3..fd62cc02b 100644 --- a/apps/Agentweaver.Api/Coordinator/CoordinatorMessages.cs +++ b/apps/Agentweaver.Api/Coordinator/CoordinatorMessages.cs @@ -55,3 +55,16 @@ public sealed record CoordinatorOutcomeSpecDecision( /// Terminal workflow output for a coordinator run. public sealed record CoordinatorOutcome(string RunId, int SpecId, string Status); + +/// +/// Raised when outcome-spec drafting exceeds its coordinator-level wall-clock bound. The bound +/// covers provider setup and session creation as well as the model turn, which have looser runtime +/// defaults and can otherwise leave the durable coordinator run in drafting indefinitely. +/// +public sealed class CoordinatorOutcomeSpecDraftTimeoutException( + string runId, + TimeSpan timeout, + Exception? innerException = null) + : TimeoutException( + $"Coordinator outcome-spec drafting for run '{runId}' exceeded {timeout.TotalSeconds:n0} seconds.", + innerException); diff --git a/apps/Agentweaver.Api/Coordinator/CoordinatorRunService.cs b/apps/Agentweaver.Api/Coordinator/CoordinatorRunService.cs index 609488bee..749b9f73d 100644 --- a/apps/Agentweaver.Api/Coordinator/CoordinatorRunService.cs +++ b/apps/Agentweaver.Api/Coordinator/CoordinatorRunService.cs @@ -846,6 +846,14 @@ private void StartWatching( "User must re-link their GitHub account.", runId); await FailRunSafeAsync(runId, entry, GitHubCopilotUnauthorizedException.AuthRequiredErrorCode).ConfigureAwait(false); } + catch (Exception ex) when (ContainsOutcomeSpecDraftTimeout(ex)) + { + _logger.LogError( + ex, + "Coordinator run {RunId} exceeded the outcome-spec drafting deadline; transitioning to Failed", + runId); + await FailRunSafeAsync(runId, entry, "outcome_spec_draft_timeout").ConfigureAwait(false); + } catch (Exception ex) { _logger.LogError(ex, "Coordinator watch loop failed for run {RunId}; transitioning to Failed", runId); @@ -854,6 +862,10 @@ private void StartWatching( }, _appStopping); } + private static bool ContainsOutcomeSpecDraftTimeout(Exception? exception) => + exception is CoordinatorOutcomeSpecDraftTimeoutException + || (exception?.InnerException is not null && ContainsOutcomeSpecDraftTimeout(exception.InnerException)); + private async Task WatchAsync( string runId, StreamingRun streamingRun, RunStreamEntry entry, string ownerUser, CancellationToken ct) { @@ -861,6 +873,21 @@ private async Task WatchAsync( { switch (evt) { + case ExecutorFailedEvent failed: + var isDraftTimeout = ContainsOutcomeSpecDraftTimeout(failed.Data); + var providerFailure = isDraftTimeout ? null : FindProviderFailure(failed.Data); + var reason = isDraftTimeout + ? "outcome_spec_draft_timeout" + : providerFailure?.ErrorCode ?? $"coordinator_executor_failed:{failed.ExecutorId}"; + _logger.LogError( + failed.Data, + "Coordinator executor {ExecutorId} failed for run {RunId}; transitioning to Failed with {Reason}", + failed.ExecutorId, + runId, + reason); + await FailRunSafeAsync(runId, entry, reason, providerFailure).ConfigureAwait(false); + return; + case RequestInfoEvent rie: // Suspended at the await-confirmation gate. The draft executor already emitted // coordinator.outcome_spec and marked the entry awaiting-review. @@ -1680,6 +1707,13 @@ private async Task IsDelegatedPlanAsync(string runId) } private async Task FailRunSafeAsync(string runId, RunStreamEntry entry, string reason = "watch_loop_error") + => await FailRunSafeAsync(runId, entry, reason, providerFailure: null).ConfigureAwait(false); + + private async Task FailRunSafeAsync( + string runId, + RunStreamEntry entry, + string reason, + AgentProviderException? providerFailure) { try { @@ -1697,7 +1731,24 @@ private async Task FailRunSafeAsync(string runId, RunStreamEntry entry, string r runId); return; } - entry.RecordNext(EventTypes.RunFailed, new { reason }); + if (!entry.HasEventType(EventTypes.RunFailed)) + { + if (providerFailure is null) + { + entry.RecordNext(EventTypes.RunFailed, new { reason }); + } + else + { + entry.RecordNext(EventTypes.RunFailed, new + { + reason = providerFailure.ErrorCode, + errorCode = providerFailure.ErrorCode, + message = providerFailure.UserMessage, + category = providerFailure.FailureKind.ToString(), + retryable = providerFailure.IsRetryable, + }); + } + } _streamStore.Complete(runId); _ = _runWorkflowFactory.PersistRunEventsAsync(runId); } @@ -1715,6 +1766,17 @@ private async Task FailRunSafeAsync(string runId, RunStreamEntry entry, string r } } + private static AgentProviderException? FindProviderFailure(Exception? exception) + { + for (var current = exception; current is not null; current = current.InnerException) + { + if (current is AgentProviderException providerFailure) + return providerFailure; + } + + return null; + } + /// /// Releases the AgentHost pod for when running pod-per-run. Best-effort: /// logs and swallows exceptions so a release failure never disrupts run finalization. No-op when diff --git a/apps/Agentweaver.Api/Coordinator/CoordinatorWorkflowFactory.cs b/apps/Agentweaver.Api/Coordinator/CoordinatorWorkflowFactory.cs index 47579f43a..296f32fea 100644 --- a/apps/Agentweaver.Api/Coordinator/CoordinatorWorkflowFactory.cs +++ b/apps/Agentweaver.Api/Coordinator/CoordinatorWorkflowFactory.cs @@ -31,6 +31,7 @@ public sealed class CoordinatorWorkflowFactory private const string InputStateKey = "coordinator-input"; private const string InputStateScope = "run-context"; private const string CoordinatorAgentName = "Coordinator"; + private const int DefaultOutcomeSpecDraftTimeoutSeconds = 120; private const string FallbackCharter = "You are the Coordinator, the built-in orchestration agent. Restate the human's goal as a " + @@ -46,6 +47,7 @@ public sealed class CoordinatorWorkflowFactory private readonly string _checkpointDir; private readonly ICheckpointStoreFactory _checkpointStoreFactory; private readonly CoordinatorOrchestratorExecutor _orchestrator; + private readonly TimeSpan _outcomeSpecDraftTimeout; public CoordinatorWorkflowFactory( IWorkflowAgentFactory agentFactory, @@ -64,6 +66,14 @@ public CoordinatorWorkflowFactory( _loggerFactory = loggerFactory; _logger = loggerFactory.CreateLogger(); + var outcomeSpecDraftTimeoutSeconds = configuration.GetValue( + "Coordinator:OutcomeSpecDraftTimeoutSeconds", + DefaultOutcomeSpecDraftTimeoutSeconds); + if (outcomeSpecDraftTimeoutSeconds <= 0) + throw new InvalidOperationException( + "Coordinator:OutcomeSpecDraftTimeoutSeconds must be greater than zero."); + _outcomeSpecDraftTimeout = TimeSpan.FromSeconds(outcomeSpecDraftTimeoutSeconds); + _checkpointDir = configuration["Coordinator:Checkpoints:Path"] ?? Path.Combine(AppPaths.DataDirectory, "coordinator-checkpoints"); _checkpointStoreFactory = checkpointStoreFactory ?? new FileCheckpointStoreFactory(); @@ -338,7 +348,7 @@ private async Task DraftAndPersistAsync( var memoryContext = await CompileMemoryContextAsync(input.ProjectId, ct).ConfigureAwait(false); var charter = BuiltInCharterResolver.Resolve(input.RepositoryPath, "coordinator") ?? FallbackCharter; - var draft = await _drafter.DraftAsync(input, charter, memoryContext, ct).ConfigureAwait(false); + var draft = await DraftWithTimeoutAsync(input, charter, memoryContext, ct).ConfigureAwait(false); var (specId, status) = await PersistDraftAsync(input, draft, ct).ConfigureAwait(false); @@ -362,6 +372,97 @@ private async Task DraftAndPersistAsync( draft.DesiredOutcome, draft.Scope, draft.Assumptions, draft.ClarifyingQuestions, status); } + private async Task DraftWithTimeoutAsync( + CoordinatorDraftInput input, + string charter, + string? memoryContext, + CancellationToken ct) + { + var draftCts = CancellationTokenSource.CreateLinkedTokenSource(ct); + var cleanupDeferred = false; + try + { + var draftTask = _drafter.DraftAsync(input, charter, memoryContext, draftCts.Token); + var timeoutTask = Task.Delay(_outcomeSpecDraftTimeout, ct); + + if (await Task.WhenAny(draftTask, timeoutTask).ConfigureAwait(false) == draftTask) + return await draftTask.ConfigureAwait(false); + + ct.ThrowIfCancellationRequested(); + var cancellationTask = draftCts.CancelAsync(); + cleanupDeferred = true; + _ = ObserveTimedOutDraftAsync(draftTask, cancellationTask, draftCts, input.RunId); + + _logger.LogError( + "Outcome-spec drafting timed out after {TimeoutSeconds}s for coordinator run {RunId}", + _outcomeSpecDraftTimeout.TotalSeconds, + input.RunId); + throw new CoordinatorOutcomeSpecDraftTimeoutException( + input.RunId, + _outcomeSpecDraftTimeout); + } + finally + { + if (!cleanupDeferred) + draftCts.Dispose(); + } + } + + private async Task ObserveTimedOutDraftAsync( + Task draftTask, + Task cancellationTask, + CancellationTokenSource draftCts, + string runId) + { + try + { + await Task.WhenAll( + ObserveTimedOutDraftCompletionAsync(draftTask, draftCts, runId), + ObserveDraftCancellationAsync(cancellationTask, runId)).ConfigureAwait(false); + } + finally + { + draftCts.Dispose(); + } + } + + private async Task ObserveTimedOutDraftCompletionAsync( + Task draftTask, + CancellationTokenSource draftCts, + string runId) + { + try + { + await draftTask.ConfigureAwait(false); + } + catch (OperationCanceledException) when (draftCts.IsCancellationRequested) + { + // Expected after the deadline requests cancellation. + } + catch (Exception ex) + { + _logger.LogWarning( + ex, + "Outcome-spec drafter faulted after its timeout for coordinator run {RunId}", + runId); + } + } + + private async Task ObserveDraftCancellationAsync(Task cancellationTask, string runId) + { + try + { + await cancellationTask.ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.LogWarning( + ex, + "Outcome-spec draft cancellation callback failed for coordinator run {RunId}", + runId); + } + } + /// /// Loads the current persisted outcome spec for a run and projects it into an /// to seed a revision's invariant-preservation context (issue diff --git a/apps/web/src/__tests__/CoordinatorRunPage.test.tsx b/apps/web/src/__tests__/CoordinatorRunPage.test.tsx index 62bcc35f8..555cc96df 100644 --- a/apps/web/src/__tests__/CoordinatorRunPage.test.tsx +++ b/apps/web/src/__tests__/CoordinatorRunPage.test.tsx @@ -1134,6 +1134,23 @@ describe('CoordinatorRunPage — graph during outcome-plan drafting', () => { expect(screen.queryByLabelText('Select Outcome plan: Pending')).toBeNull(); }); + it('uses the durable run start while drafting so elapsed time does not remain at zero', async () => { + vi.setSystemTime(new Date('2026-07-07T00:02:05.000Z')); + vi.mocked(apiClient.getRun).mockResolvedValue({ + run_id: 'coord-run-1', + status: 'in_progress', + coordinator_status: 'drafting', + started_at: '2026-07-07T00:01:00.000Z', + ended_at: null, + } as never); + vi.mocked(apiClient.getRunGraph).mockResolvedValue(COORDINATOR_GRAPH_DRAFTING_DESCRIPTOR); + + render(); + + const progress = await screen.findByTestId('run-progress-chips', undefined, { timeout: 4000 }); + expect(progress.textContent).toContain('1m 5s elapsed'); + }); + it('hides the assembly pipeline stages and shows a caption while drafting the spec', async () => { // Drafting state: coordinator + planned assembly stages, no subtasks, no confirmed spec. vi.mocked(apiClient.getRunGraph).mockResolvedValue(COORDINATOR_GRAPH_DRAFTING_DESCRIPTOR); diff --git a/apps/web/src/pages/CoordinatorRunPage.tsx b/apps/web/src/pages/CoordinatorRunPage.tsx index fee26b548..f56488e09 100644 --- a/apps/web/src/pages/CoordinatorRunPage.tsx +++ b/apps/web/src/pages/CoordinatorRunPage.tsx @@ -361,6 +361,12 @@ function readEventTimestamp(p: Record): string | undefined { return readStr(p, ['timestamp_utc', 'timestampUtc', 'updated_at', 'updatedAt', 'timestamp']); } +function parseTimestamp(value: string | null | undefined): number | undefined { + if (!value) return undefined; + const parsed = Date.parse(value); + return Number.isFinite(parsed) ? parsed : undefined; +} + function readChildRunId(node: GraphDescriptor['nodes'][number]): string | undefined { return node.child_run_id ?? readStr(node.data ?? {}, ['child_run_id', 'childRunId']) @@ -2264,6 +2270,13 @@ export function CoordinatorRunPage() { const setRunLevelStatus = useCallback((status: RunStatus | undefined) => { setRunLevelStatusState({ runId: runId ?? '', status }); }, [runId]); + const [runTimingState, setRunTimingState] = useState<{ + runId: string; + startedAt: number | undefined; + endedAt: number | undefined; + }>({ runId: '', startedAt: undefined, endedAt: undefined }); + const runStartedAt = runTimingState.runId === (runId ?? '') ? runTimingState.startedAt : undefined; + const runEndedAt = runTimingState.runId === (runId ?? '') ? runTimingState.endedAt : undefined; const { events, @@ -2447,6 +2460,7 @@ export function CoordinatorRunPage() { setWorkPlanError(null); setNoWorkPlan(false); setRunLevelStatus(undefined); + setRunTimingState({ runId: runId ?? '', startedAt: undefined, endedAt: undefined }); setCoordStatusField(undefined); setCoordStatusReason(undefined); setCoordinatorSteerable(undefined); @@ -2513,6 +2527,11 @@ export function CoordinatorRunPage() { setCoordinatorSteerable(typeof detail?.coordinator_steerable === 'boolean' ? detail.coordinator_steerable : undefined); setWorkPlanStatus(wpStatus); setRunLevelStatus(detail?.status ?? undefined); + setRunTimingState({ + runId, + startedAt: parseTimestamp(detail?.started_at), + endedAt: parseTimestamp(detail?.ended_at), + }); if (wp) consecutiveWorkPlanNotReady = 0; // Seed the option toggles once from the run detail; subsequent user toggles own the state. if (!seededToggles.current && detail) { @@ -3508,12 +3527,14 @@ export function CoordinatorRunPage() { { pending: 0, waiting: 0, blocked: 0, failed: 0 }, ); const hasRunningSessionItem = flatSessionTree.some((node) => node.startedAt !== undefined && node.completedAt === undefined); - const elapsedNow = useTickingNow(hasRunningSessionItem); - const earliestStart = flatSessionTree.reduce( + const earliestNodeStart = flatSessionTree.reduce( (min, node) => (node.startedAt == null ? min : min == null ? node.startedAt : Math.min(min, node.startedAt)), undefined, ); - const elapsedLabel = earliestStart ? fmtTotal(elapsedNow - earliestStart) : '0s'; + const earliestStart = earliestNodeStart ?? runStartedAt; + const elapsedNow = useTickingNow(hasRunningSessionItem || (!viewState.terminal && earliestStart !== undefined)); + const elapsedEnd = viewState.terminal ? (runEndedAt ?? elapsedNow) : elapsedNow; + const elapsedLabel = earliestStart ? fmtTotal(Math.max(0, elapsedEnd - earliestStart)) : '0s'; const runStatusText = viewState.label; const taskCountsLabel = `${taskRows.length} tasks · ${taskStatusSummary.pending} pending · ${taskStatusSummary.waiting} waiting`; diff --git a/docs/deep-dive/coordinator-internals.md b/docs/deep-dive/coordinator-internals.md index 8e2de3cb0..ee4b18e9e 100644 --- a/docs/deep-dive/coordinator-internals.md +++ b/docs/deep-dive/coordinator-internals.md @@ -584,6 +584,20 @@ re-arms the correct service. For `dispatching` plans it honors the distributed ` lease first, skipping freshly owned plans and stealing only stale ones. Each candidate is isolated by try/catch so one corrupt plan does not stop the sweep. +### Bounded outcome-spec drafting + +Outcome-spec drafting has its own wall-clock bound because it includes provider setup and session +creation before the normal streaming-turn watchdog begins. `Coordinator:OutcomeSpecDraftTimeoutSeconds` +defaults to **120 seconds** and must be greater than zero. When it expires, the coordinator cancels +the draft without waiting for provider cancellation callbacks, transitions the durable run to +`failed`, and emits `run.failed` with reason `outcome_spec_draft_timeout` instead of leaving the run +indefinitely in `drafting`. The abandoned draft and cancellation work remain observed, and their +linked cancellation source is disposed after both settle. + +| Configuration key | Default | Effect | +|---|---:|---| +| `Coordinator:OutcomeSpecDraftTimeoutSeconds` | `120` | Maximum wall-clock time for provider setup, session creation, and the outcome-spec model turn before the coordinator fails with `outcome_spec_draft_timeout`; must be greater than zero. | + ### Bounded final-Scribe recovery At startup, recovery checks terminal coordinator runs for a missing final Scribe. It skips runs that diff --git a/packages/Agentweaver.AgentRuntime/CopilotAIAgent.cs b/packages/Agentweaver.AgentRuntime/CopilotAIAgent.cs index b47fa32bc..4bb005f45 100644 --- a/packages/Agentweaver.AgentRuntime/CopilotAIAgent.cs +++ b/packages/Agentweaver.AgentRuntime/CopilotAIAgent.cs @@ -806,7 +806,7 @@ internal static bool IsMissingCopilotAuth(Exception ex) return providerFailure; } - private void EmitProviderFailure(AgentProviderException providerFailure) + internal void EmitProviderFailure(AgentProviderException providerFailure) { Emit(EventTypes.RunFailed, new { diff --git a/tests/Agentweaver.Tests/Coordinator/CoordinatorOutcomeSpecTests.cs b/tests/Agentweaver.Tests/Coordinator/CoordinatorOutcomeSpecTests.cs index 20e4d9264..c31d4f0e7 100644 --- a/tests/Agentweaver.Tests/Coordinator/CoordinatorOutcomeSpecTests.cs +++ b/tests/Agentweaver.Tests/Coordinator/CoordinatorOutcomeSpecTests.cs @@ -5,6 +5,7 @@ using FluentAssertions; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; +using Agentweaver.AgentRuntime.Providers; using Agentweaver.Api.Contracts; using Agentweaver.Api.Coordinator; using Agentweaver.Api.Infrastructure; @@ -137,6 +138,205 @@ public async Task Start_UsesProjectOutcomeSpecGenerationModel() drafter.LastInput!.OutcomeSpecGenerationModel.Should().Be("gpt-5-mini"); } + [Fact] + public async Task Start_DraftExceedsDeadline_FailsRunAndEmitsTypedTerminal() + { + var projectId = await CreateProjectAsync(); + var drafter = _factory.Services.GetRequiredService() + .Should().BeOfType().Subject; + drafter.BlockUntilCancelled = true; + + var runId = await StartOrchestrationAsync( + projectId, + "A provider startup that never completes must not strand outcome planning"); + + RunResponse? run = null; + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10); + while (DateTime.UtcNow < deadline) + { + run = await GetRunAsync(_owner, runId); + if (run?.Status == "failed") + break; + await Task.Delay(50); + } + + run.Should().NotBeNull(); + run!.Status.Should().Be("failed"); + var cancellationDeadline = DateTime.UtcNow + TimeSpan.FromSeconds(5); + while (!drafter.CancellationObserved && DateTime.UtcNow < cancellationDeadline) + await Task.Delay(25); + drafter.CancellationObserved.Should().BeTrue( + "the coordinator deadline must still cancel provider setup/session work after failing the run"); + + var spec = await GetOutcomeSpecAsync(_owner, runId); + spec.Should().NotBeNull(); + spec!.Status.Should().Be("drafting", + "the persisted drafting row should remain diagnostic evidence rather than masquerade as a completed plan"); + + var eventsResponse = await _owner.GetAsync($"/api/runs/{runId}/events"); + eventsResponse.StatusCode.Should().Be(HttpStatusCode.OK); + var events = await eventsResponse.Content.ReadFromJsonAsync(); + var failedEvent = events.Should().NotBeNull().And.Subject + .Single(e => e.GetProperty("type").GetString() == EventTypes.RunFailed); + failedEvent.GetProperty("payload").GetProperty("reason").GetString() + .Should().Be("outcome_spec_draft_timeout"); + } + + [Fact] + public async Task Start_DraftCancellationCallbackBlocks_FailsPromptlyAtDeadlineAndCleansUp() + { + var projectId = await CreateProjectAsync(); + var drafter = _factory.Services.GetRequiredService() + .Should().BeOfType().Subject; + drafter.BlockCancellationCallback = true; + + var deadlineStopwatch = Stopwatch.StartNew(); + var runId = await StartOrchestrationAsync( + projectId, + "A blocked provider cancellation callback must not delay terminal failure"); + + try + { + await drafter.BlockedCancellationStarted.WaitAsync(TimeSpan.FromSeconds(5)); + var terminalizationStopwatch = Stopwatch.StartNew(); + + RunResponse? run = null; + var terminalDeadline = DateTime.UtcNow + TimeSpan.FromSeconds(2); + while (DateTime.UtcNow < terminalDeadline) + { + run = await GetRunAsync(_owner, runId); + if (run?.Status == "failed") + break; + await Task.Delay(25); + } + + run.Should().NotBeNull(); + run!.Status.Should().Be("failed", + "terminalization must not await a provider cancellation callback that is still blocked"); + terminalizationStopwatch.Elapsed.Should().BeLessThan(TimeSpan.FromSeconds(2)); + deadlineStopwatch.Elapsed.Should().BeLessThan(TimeSpan.FromSeconds(4), + "the test host configures a one-second drafting deadline"); + } + finally + { + drafter.ReleaseBlockedCancellation(); + } + + await drafter.BlockedDraftCompleted.WaitAsync(TimeSpan.FromSeconds(5)); + drafter.CancellationObserved.Should().BeTrue(); + + await using var scope = _factory.Services.CreateAsyncScope(); + var db = scope.ServiceProvider.GetRequiredService(); + List durableFailures = []; + var persistenceDeadline = DateTime.UtcNow + TimeSpan.FromSeconds(10); + while (DateTime.UtcNow < persistenceDeadline) + { + durableFailures = await db.RunEvents.AsNoTracking() + .Where(e => e.RunId == runId && e.EventType == EventTypes.RunFailed) + .OrderBy(e => e.Sequence) + .ToListAsync(); + if (durableFailures.Count > 0) + break; + await Task.Delay(50); + } + + var failedEvent = durableFailures.Should().ContainSingle( + "deadline cleanup must not emit a second terminal event") + .Subject; + var payload = JsonSerializer.Deserialize(failedEvent.PayloadJson); + payload.GetProperty("reason").GetString() + .Should().Be("outcome_spec_draft_timeout"); + } + + [Fact] + public async Task Start_CopilotProviderFailure_PreservesTypedDurableTerminalWithoutDuplicate() + { + var projectId = await CreateProjectAsync(); + var drafter = _factory.Services.GetRequiredService() + .Should().BeOfType().Subject; + drafter.ProviderFailureToThrow = new AgentProviderException( + ModelSource.GitHubCopilot, + AgentProviderFailureKind.ProviderUnavailable, + "github_copilot_models_unavailable", + "GitHub Copilot could not list available models.", + isRetryable: false); + + var runId = await StartOrchestrationAsync( + projectId, + "A typed Copilot provider failure must remain the coordinator terminal"); + + RunResponse? run = null; + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10); + while (DateTime.UtcNow < deadline) + { + run = await GetRunAsync(_owner, runId); + if (run?.Status == "failed") + break; + await Task.Delay(50); + } + + run.Should().NotBeNull(); + run!.Status.Should().Be("failed"); + run.Result.Should().Be("github_copilot_models_unavailable"); + + await using var scope = _factory.Services.CreateAsyncScope(); + var db = scope.ServiceProvider.GetRequiredService(); + List durableFailures = []; + deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10); + while (DateTime.UtcNow < deadline) + { + durableFailures = await db.RunEvents.AsNoTracking() + .Where(e => e.RunId == runId && e.EventType == EventTypes.RunFailed) + .OrderBy(e => e.Sequence) + .ToListAsync(); + if (durableFailures.Count > 0) + break; + await Task.Delay(50); + } + + var durableFailure = durableFailures.Should().ContainSingle( + "CopilotAIAgent already emitted the provider terminal before MAF surfaced ExecutorFailedEvent") + .Subject; + var payload = JsonSerializer.Deserialize(durableFailure.PayloadJson); + payload.GetProperty("errorCode").GetString().Should().Be("github_copilot_models_unavailable"); + payload.GetProperty("message").GetString().Should().Be("GitHub Copilot could not list available models."); + payload.GetProperty("category").GetString().Should().Be( + AgentProviderFailureKind.ProviderUnavailable.ToString()); + payload.GetProperty("retryable").GetBoolean().Should().BeFalse(); + } + + [Fact] + public async Task Start_DrafterThrowsTimeout_DoesNotMislabelCoordinatorDeadline() + { + var projectId = await CreateProjectAsync(); + var drafter = _factory.Services.GetRequiredService() + .Should().BeOfType().Subject; + drafter.ExceptionToThrow = new TimeoutException("simulated provider timeout"); + + var runId = await StartOrchestrationAsync( + projectId, + "An immediate provider timeout must keep its executor-failure classification"); + + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10); + RunResponse? run = null; + while (DateTime.UtcNow < deadline) + { + run = await GetRunAsync(_owner, runId); + if (run?.Status == "failed") + break; + await Task.Delay(50); + } + + run.Should().NotBeNull(); + run!.Status.Should().Be("failed"); + + var events = await _owner.GetFromJsonAsync($"/api/runs/{runId}/events"); + var failedEvent = events.Should().NotBeNull().And.Subject + .Single(e => e.GetProperty("type").GetString() == EventTypes.RunFailed); + failedEvent.GetProperty("payload").GetProperty("reason").GetString() + .Should().Be("coordinator_executor_failed:coordinator-draft"); + } + [Fact] public async Task Start_DefineOutcomeMode_DraftsSpecAndSuspendsAtGate() { diff --git a/tests/Agentweaver.Tests/Helpers/CoordinatorWebApplicationFactory.cs b/tests/Agentweaver.Tests/Helpers/CoordinatorWebApplicationFactory.cs index 567fba417..c3d9e59ec 100644 --- a/tests/Agentweaver.Tests/Helpers/CoordinatorWebApplicationFactory.cs +++ b/tests/Agentweaver.Tests/Helpers/CoordinatorWebApplicationFactory.cs @@ -4,6 +4,7 @@ using Microsoft.Extensions.DependencyInjection; using Agentweaver.Api.Auth; using Agentweaver.Api.Git; +using Agentweaver.Api.Infrastructure; using Agentweaver.Domain; namespace Agentweaver.Tests.Helpers; @@ -122,6 +123,7 @@ protected override void ConfigureWebHost(IWebHostBuilder builder) ["Providers:MicrosoftFoundry:Deployment"] = "gpt-4o", ["RunBounds:MaxSteps"] = "50", ["RunBounds:MaxMinutes"] = "10", + ["Coordinator:OutcomeSpecDraftTimeoutSeconds"] = "1", // Phase 1 + decompose/persist suite: keep child dispatch off so the confirm/decline // lifecycle and the work-plan contract stay deterministic in this hermetic host // (non-git workspaces + signed-out tokens cannot spawn real child runs). The @@ -137,8 +139,8 @@ protected override void ConfigureWebHost(IWebHostBuilder builder) // drafting step never makes a live model call. The boilerplate spec lives in the test // project, not production (production fails the run when the model is unavailable). RemoveService(services); - services.AddSingleton( - new FakeCoordinatorSpecDrafter()); + services.AddSingleton(sp => + new FakeCoordinatorSpecDrafter(sp.GetRequiredService())); // Replace the production Copilot-backed reply classifier with a deterministic, hermetic // fake so the confirm/revise routing at the outcome-spec gate never makes a live model diff --git a/tests/Agentweaver.Tests/Helpers/FakeCoordinatorSpecDrafter.cs b/tests/Agentweaver.Tests/Helpers/FakeCoordinatorSpecDrafter.cs index 4a2cca94b..f90c16fd8 100644 --- a/tests/Agentweaver.Tests/Helpers/FakeCoordinatorSpecDrafter.cs +++ b/tests/Agentweaver.Tests/Helpers/FakeCoordinatorSpecDrafter.cs @@ -1,4 +1,12 @@ +using Agentweaver.AgentRuntime; +using Agentweaver.AgentRuntime.Providers; +using Agentweaver.AgentTools; using Agentweaver.Api.Coordinator; +using Agentweaver.Api.Infrastructure; +using Agentweaver.Api.Runs; +using Agentweaver.SandboxExec; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.Logging.Abstractions; namespace Agentweaver.Tests.Helpers; @@ -12,12 +20,74 @@ namespace Agentweaver.Tests.Helpers; /// public sealed class FakeCoordinatorSpecDrafter : ICoordinatorSpecDrafter { + private readonly RunStreamStore _streamStore; + private readonly TaskCompletionSource _blockedCancellationStarted = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _releaseBlockedCancellation = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _blockedDraftCompleted = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + public FakeCoordinatorSpecDrafter(RunStreamStore streamStore) => _streamStore = streamStore; + public CoordinatorDraftInput? LastInput { get; private set; } + public bool BlockUntilCancelled { get; set; } + public bool BlockCancellationCallback { get; set; } + public bool CancellationObserved { get; private set; } + public Exception? ExceptionToThrow { get; set; } + public AgentProviderException? ProviderFailureToThrow { get; set; } + public Task BlockedCancellationStarted => _blockedCancellationStarted.Task; + public Task BlockedDraftCompleted => _blockedDraftCompleted.Task; - public Task DraftAsync( + public void ReleaseBlockedCancellation() => _releaseBlockedCancellation.TrySetResult(true); + + public async Task DraftAsync( CoordinatorDraftInput input, string charter, string? memoryContext, CancellationToken ct) { LastInput = input; + if (ProviderFailureToThrow is { } providerFailure) + { + await EmitProviderFailureAsync(input.RunId, providerFailure); + throw providerFailure; + } + if (ExceptionToThrow is not null) + throw ExceptionToThrow; + if (BlockCancellationCallback) + { + var cancellationDelivered = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + var blockingRegistration = ct.Register(() => + { + _blockedCancellationStarted.TrySetResult(true); + _releaseBlockedCancellation.Task.GetAwaiter().GetResult(); + }); + var deliveryRegistration = ct.Register(() => cancellationDelivered.TrySetResult(true)); + try + { + await cancellationDelivered.Task.ConfigureAwait(false); + CancellationObserved = true; + throw new OperationCanceledException(ct); + } + finally + { + deliveryRegistration.Dispose(); + blockingRegistration.Dispose(); + _blockedDraftCompleted.TrySetResult(true); + } + } + if (BlockUntilCancelled) + { + try + { + await Task.Delay(Timeout.InfiniteTimeSpan, ct); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + CancellationObserved = true; + throw; + } + } + var goal = input.Goal.Trim(); var hasContext = !string.IsNullOrWhiteSpace(memoryContext); @@ -46,6 +116,24 @@ public Task DraftAsync( : null) : "Revision requested: " + input.ReviseFeedback.Trim(); - return Task.FromResult(new OutcomeSpecDraft(desired, scope, assumptions, questions)); + return new OutcomeSpecDraft(desired, scope, assumptions, questions); + } + + private async Task EmitProviderFailureAsync(string runId, AgentProviderException providerFailure) + { + var entry = _streamStore.Get(runId) + ?? throw new InvalidOperationException($"Missing run stream for coordinator run {runId}."); + await using var clientFactory = new GitHubCopilotClientFactory( + new ConfigurationBuilder().Build(), + new FixedGitHubCopilotCapabilityCredentialProvider()); + await using var agent = new CopilotAIAgent( + clientFactory, + SandboxExecutorFactory.CreatePassthrough(), + new StubPolicyStore(), + new InMemoryShellApprovalStore(), + new InMemoryToolApprovalGate(), + NullLogger.Instance); + agent.SetTurnStreamWriter(new RecordingChannelWriter(entry)); + agent.EmitProviderFailure(providerFailure); } }