From 85357b9f9413bccd30066e26ad18005b02b7a149 Mon Sep 17 00:00:00 2001 From: "Mars.P" Date: Mon, 19 Oct 2026 08:36:47 +0800 Subject: [PATCH] Carry a provider failure out of a container body A provider failure inside a flow.map branch or a flow.loop pass that ended the container was written on the body node's own node.failed record and nowhere the run's remedy could find it. The run's failure is the one the node that ENDED the run recorded; for such a run that node is the container, and the container's node.failed carried only its sentence, so the run page said "Node 'map' failed." with no card, no credential and no repair. The container's record now carries the failure of the cell whose unhandled failure ended it: that node's own recorded failure, else the agent run that last worked in that very cell, read through a new ICellFailureReader in the order and with the parsers RunFailureReader uses for the node that ended a run. A nested container reads the record the level below wrote, so the failure travels as far as the nesting goes and the run's remedy is the one a top-level node of the same failure gets. Fail-fast and collect policies, retries, partial results and the sentences containers write are untouched; only what the record carries changes. The cell is chosen deterministically. A terminate map whose branches fail concurrently answers for the first failing branch by element order, which is the branch its sentence already names, never whichever finished first. A branch replayed from the ledger while a sibling was parked is still the cell that ended the map and is read back from its own record. Nothing else is read: a provider failure the body handled with an error edge, one a try caught, one a sibling branch met, or one a continue policy skipped leaves nothing on a container that later fails for another reason, and a container that fails on its own (a tripped budget, a binding that is not an array) carries none. A map-branch rerun replays a failed sibling into its fork, and the agent run that worked in that branch belongs to the original run, so the fork could never find it: its map failed again on the replayed branch and answered with no remedy. The rerun seeder now re-emits a failed cell with the failure the original run recorded for it (the cell's own record, else its agent run, through the same reader), so the fork's map answers for the replayed branch as the original's did. A failed cell that recorded no failure is re-emitted exactly as before. flow.iterate is a single-expression node, not a container, and flow.subworkflow runs a child run: a failure inside a child run is not carried to the parent node by this change. Additive: a container record gains the failure key a node record already carries, no migration, no API or wire change. A row written before this reads exactly as it did, and a reader that predates the key ignores it. --- .../Workflows/Engine/WorkflowEngine.cs | 69 ++- .../Workflows/Failures/CellFailureReader.cs | 51 +++ .../Workflows/Failures/ICellFailureReader.cs | 20 + .../Workflows/Failures/RunFailureReader.cs | 4 +- .../Workflows/Rerun/RerunCellSeeder.cs | 26 +- ...ProviderFailureInAContainerHttpE2ETests.cs | 234 +++++++++++ .../Credentials/CellFailureReaderFlowTests.cs | 265 ++++++++++++ ...ureThroughTheEngineFlowTests.Containers.cs | 393 ++++++++++++++++++ ...rFailureThroughTheEngineFlowTests.Rerun.cs | 94 +++++ ...roviderFailureThroughTheEngineFlowTests.cs | 2 +- backend/tests/TESTING.md | 13 + 11 files changed, 1149 insertions(+), 22 deletions(-) create mode 100644 backend/src/CodeSpace.Core/Services/Workflows/Failures/CellFailureReader.cs create mode 100644 backend/src/CodeSpace.Core/Services/Workflows/Failures/ICellFailureReader.cs create mode 100644 backend/tests/CodeSpace.E2ETests/Credentials/ProviderFailureInAContainerHttpE2ETests.cs create mode 100644 backend/tests/CodeSpace.IntegrationTests/Credentials/CellFailureReaderFlowTests.cs create mode 100644 backend/tests/CodeSpace.IntegrationTests/Credentials/ProviderFailureThroughTheEngineFlowTests.Containers.cs create mode 100644 backend/tests/CodeSpace.IntegrationTests/Credentials/ProviderFailureThroughTheEngineFlowTests.Rerun.cs diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Engine/WorkflowEngine.cs b/backend/src/CodeSpace.Core/Services/Workflows/Engine/WorkflowEngine.cs index 5c24504e7..64da768d4 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/Engine/WorkflowEngine.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/Engine/WorkflowEngine.cs @@ -12,6 +12,7 @@ using CodeSpace.Core.Services.Variables; using CodeSpace.Core.Services.Workflows; using CodeSpace.Core.Services.Workflows.Artifacts; +using CodeSpace.Core.Services.Workflows.Failures; using CodeSpace.Core.Services.Workflows.Lifecycle; using CodeSpace.Core.Services.Workflows.Nodes; using CodeSpace.Core.Services.Workflows.Budget; @@ -21,6 +22,7 @@ using CodeSpace.Messages.Constants; using CodeSpace.Messages.Dtos.Workflows; using CodeSpace.Messages.Enums; +using CodeSpace.Messages.Failures; using CodeSpace.Messages.Workflows; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Logging; @@ -68,6 +70,7 @@ public sealed class WorkflowEngine : IWorkflowEngine, IScopedDependency private readonly IAgentRunService _agentRunService; private readonly IAgentDefinitionResolver _agentDefinitionResolver; private readonly IRunCancellationRegistry _cancellationRegistry; + private readonly ICellFailureReader _cellFailureReader; // The run's lifetime scope — BeginLifetimeScope() per parallel node gives each its own DbContext + // record logger so concurrent nodes never share an EF change-tracker (see RunNodeInChildScopeAsync). private readonly ILifetimeScope _lifetimeScope; @@ -92,7 +95,7 @@ public sealed class WorkflowEngine : IWorkflowEngine, IScopedDependency internal const int DefaultMaxParallelism = 8; internal const int MaxParallelismCeiling = 64; - public WorkflowEngine(CodeSpaceDbContext db, INodeRegistry nodeRegistry, IVariableService variableService, IRunRecordLogger recordLogger, IPayloadRedactor redactor, IWorkflowSensitivePayloadStore sensitivePayloadStore, IRunDataCompletenessWriter completenessWriter, IArtifactStore artifactStore, IArtifactOffloader offloader, ILogger logger, ILoggerFactory loggerFactory, ICodeSpaceBackgroundJobClient backgroundJobClient, ISubworkflowService subworkflowService, IWorkflowResumeService resumeService, IAgentRunService agentRunService, IAgentDefinitionResolver agentDefinitionResolver, IRunCancellationRegistry cancellationRegistry, ILifetimeScope lifetimeScope, Completion.ICompletionTerminalAuthority terminalAuthority) + public WorkflowEngine(CodeSpaceDbContext db, INodeRegistry nodeRegistry, IVariableService variableService, IRunRecordLogger recordLogger, IPayloadRedactor redactor, IWorkflowSensitivePayloadStore sensitivePayloadStore, IRunDataCompletenessWriter completenessWriter, IArtifactStore artifactStore, IArtifactOffloader offloader, ILogger logger, ILoggerFactory loggerFactory, ICodeSpaceBackgroundJobClient backgroundJobClient, ISubworkflowService subworkflowService, IWorkflowResumeService resumeService, IAgentRunService agentRunService, IAgentDefinitionResolver agentDefinitionResolver, IRunCancellationRegistry cancellationRegistry, ICellFailureReader cellFailureReader, ILifetimeScope lifetimeScope, Completion.ICompletionTerminalAuthority terminalAuthority) { _terminalAuthority = terminalAuthority; _db = db; @@ -112,6 +115,7 @@ public WorkflowEngine(CodeSpaceDbContext db, INodeRegistry nodeRegistry, IVariab _agentRunService = agentRunService; _agentDefinitionResolver = agentDefinitionResolver; _cancellationRegistry = cancellationRegistry; + _cellFailureReader = cellFailureReader; _lifetimeScope = lifetimeScope; _maxParallelism = ParseMaxParallelism(Environment.GetEnvironmentVariable(MaxParallelismEnvVar)); } @@ -1496,12 +1500,12 @@ private void AdvanceAfterNodeSettled(NodeDefinition node, NodeStatus status, Wal /// node with its own error edge is "handled" regardless of policy. Returns true ONLY /// for the continue-abandon case; otherwise releases the ready downstream and returns false. /// - private bool AdvanceBodyNodeOrAbandon(NodeDefinition node, NodeStatus status, WalkerState bodyState, LoopErrorHandling errorHandling) + private bool AdvanceBodyNodeOrAbandon(NodeDefinition node, NodeStatus status, WalkerState bodyState, LoopErrorHandling errorHandling, string iterationKey) { if (status == NodeStatus.Failure && !HasErrorEdge(node, bodyState)) { if (errorHandling == LoopErrorHandling.Terminate) - throw new NodeFailureException($"Node '{node.Id}' failed."); + throw new NodeFailureException($"Node '{node.Id}' failed.", new FailedCell(node.Id, iterationKey)); return true; // continue: abandon the rest of this pass } @@ -1673,11 +1677,24 @@ private async Task ExecuteLoopAsync(WorkflowRun run, WorkflowDefinit // A body failure (no error edge) or a tripped budget — expose it as the loop's `error` // output so the loop's own error edge can catch it, then write node.failed. scope.Nodes[loopNode.Id] = BuildErrorOutput(ex.Message, loopNode.Id); - await _recordLogger.NodeFailedAsync(run.Id, loopNode.Id, nodeIterationKey, RedactForPersistence(scope, ex.Message), DateTimeOffset.UtcNow - startedAt, cancellationToken).ConfigureAwait(false); + + var failure = await ReadEndingCellFailureAsync(run, ex, cancellationToken).ConfigureAwait(false); + + await _recordLogger.NodeFailedAsync(run.Id, loopNode.Id, nodeIterationKey, RedactForPersistence(scope, ex.Message), DateTimeOffset.UtcNow - startedAt, failure, cancellationToken).ConfigureAwait(false); return NodeStatus.Failure; } } + /// + /// The provider failure behind a container's failure, for the container's own node.failed record to carry beside its sentence: what + /// the cell whose unhandled failure ended it recorded (its own failure, else its agent run's), read as + /// reads any cell — so the run's remedy is the one a top-level node of the same failure gets. Null when the container failed on its own + /// (a tripped budget, a binding that is not an array) or the cell recorded no failure a person can repair. It reads only the cell that ended + /// the container, never a failure its body handled or a sibling branch met: the run's failure is the one the node that ENDED it recorded. + /// + private async Task ReadEndingCellFailureAsync(WorkflowRun run, NodeFailureException ended, CancellationToken cancellationToken) => + ended.FailedCell is { } cell ? await _cellFailureReader.ReadAsync(run.Id, run.TeamId, cell.NodeId, cell.IterationKey, cancellationToken).ConfigureAwait(false) : null; + /// /// Run a flow.try scope container: execute its body subgraph ONCE; if any body node fails /// unhandled, CATCH it — route the run down the container's catch handle (exposing the @@ -1794,7 +1811,10 @@ private async Task ExecuteMapAsync(WorkflowRun run, WorkflowDefiniti // A non-array binding, a tripped ceiling, or (terminate mode) a branch failure — expose it as // the map's `error` output so the map's own error edge can catch it, then write node.failed. scope.Nodes[mapNode.Id] = BuildErrorOutput(ex.Message, mapNode.Id); - await _recordLogger.NodeFailedAsync(run.Id, mapNode.Id, nodeIterationKey, RedactForPersistence(scope, ex.Message), DateTimeOffset.UtcNow - startedAt, cancellationToken).ConfigureAwait(false); + + var failure = await ReadEndingCellFailureAsync(run, ex, cancellationToken).ConfigureAwait(false); + + await _recordLogger.NodeFailedAsync(run.Id, mapNode.Id, nodeIterationKey, RedactForPersistence(scope, ex.Message), DateTimeOffset.UtcNow - startedAt, failure, cancellationToken).ConfigureAwait(false); return NodeStatus.Failure; } } @@ -1807,7 +1827,9 @@ private async Task ExecuteMapAsync(WorkflowRun run, WorkflowDefiniti /// ) is NOT re-dispatched: its replayed result is slotted in directly, so a /// completed branch's side effect never re-fires on a sibling-triggered re-walk (the exactly-once-per- /// branch guarantee). A branch failure is honoured per the map's error policy: terminate rethrows - /// the first branch's failure to fail the map; continue stores a failure marker at that index + + /// the first branch's failure to fail the map — "first" is by ELEMENT ORDER, never by which branch finished + /// first, and the map's sentence and the cell its node.failed record reads its failure from + /// () are both that branch's; continue stores a failure marker at that index + /// counts it. A branch SUSPEND parks that branch's wait (PR2) and is counted in the returned suspended /// total — the caller suspends the whole map when it is > 0. Returns the ordered results, the /// failed-branch count, and the suspended-branch count. @@ -1836,12 +1858,16 @@ private async Task ExecuteMapAsync(WorkflowRun run, WorkflowDefiniti var failed = 0; var suspended = 0; string? terminateFailure = null; + FailedCell? terminatedAt = null; foreach (var outcome in outcomes) { results[outcome.Index] = outcome.Result; if (outcome.Failed) failed++; if (outcome.Suspended) suspended++; - terminateFailure ??= outcome.TerminateFailure; + + // The branch that fails the map is the FIRST one by element order that terminated it — never whichever finished first — so the sentence, and + // the failure the map's record carries, always name the same branch and a re-walk answers the same. + if (terminateFailure is null && outcome.TerminateFailure is not null) (terminateFailure, terminatedAt) = (outcome.TerminateFailure, outcome.TerminatedAt); } // A terminate-mode failure yields to a suspended sibling: re-suspend the map (its durable waits live, @@ -1849,7 +1875,7 @@ private async Task ExecuteMapAsync(WorkflowRun run, WorkflowDefiniti // that branch's Pending wait + leak its dispatched child/agent run. The terminate failure re-surfaces // on a later re-walk once no branch is parked. With no suspended branch it fails the map now. if (suspended == 0 && terminateFailure != null) - throw new NodeFailureException(terminateFailure); + throw new NodeFailureException(terminateFailure, terminatedAt); return (results, failed, suspended); } @@ -1910,9 +1936,11 @@ private async Task RunMapBranchInChildScopeAsync(WorkflowRun r /// barrier on EVERY branch: if any sibling suspended, the map re-suspends (its durable waits live, to be /// resumed) and the terminate failure is deferred to a later re-walk — never leaking a sibling's wait by /// failing the map out from under it. A suspended branch carries an empty result placeholder; the map - /// re-suspends rather than reducing, so the placeholder is never emitted. + /// re-suspends rather than reducing, so the placeholder is never emitted. TerminatedAt is the cell + /// whose failure ended the branch (set with TerminateFailure): where the map reads the failure its own + /// node.failed record carries. /// - private readonly record struct MapBranchOutcome(int Index, JsonElement Result, bool Failed, bool Suspended = false, string? TerminateFailure = null, Guid? SourceRecordId = null, Guid? SourceRunId = null); + private readonly record struct MapBranchOutcome(int Index, JsonElement Result, bool Failed, bool Suspended = false, string? TerminateFailure = null, Guid? SourceRecordId = null, Guid? SourceRunId = null, FailedCell? TerminatedAt = null); /// /// Run a map body subgraph for ONE element as a parallel WAVE — the same model as the loop body @@ -1983,7 +2011,7 @@ private async Task RunMapBranchOnceAsync(WorkflowRun run, Work var step = AdvanceMapBranchNodeOrAbandon(mapNode, regular[i], outcomes[i].Status, bodyState, plan.ErrorHandling, out var marker); if (step == MapBranchStep.Abandon) return new MapBranchOutcome(index, marker, Failed: true); - if (step == MapBranchStep.Terminate) return TerminateOutcome(index, mapNode, regular[i]); + if (step == MapBranchStep.Terminate) return TerminateOutcome(index, mapNode, regular[i], branchKey); } // A nested container (loop / try / map) recurses under THIS branch's key; it writes its own @@ -1995,7 +2023,7 @@ private async Task RunMapBranchOnceAsync(WorkflowRun run, Work var step = AdvanceMapBranchNodeOrAbandon(mapNode, container, status, bodyState, plan.ErrorHandling, out var marker); if (step == MapBranchStep.Abandon) return new MapBranchOutcome(index, marker, Failed: true); - if (step == MapBranchStep.Terminate) return TerminateOutcome(index, mapNode, container); + if (step == MapBranchStep.Terminate) return TerminateOutcome(index, mapNode, container, branchKey); } } } @@ -2018,8 +2046,8 @@ private async Task RunMapBranchOnceAsync(WorkflowRun run, Work /// A terminate-mode unhandled branch failure, carried back as an outcome (not thrown) so the /// fan-out can barrier on every branch first — if any sibling SUSPENDED, the map re-suspends and this /// failure is deferred, never orphaning the sibling's durable wait. - private static MapBranchOutcome TerminateOutcome(int index, NodeDefinition mapNode, NodeDefinition node) => - new(index, EmptyJsonObject(), Failed: true, TerminateFailure: $"Node '{node.Id}' in map '{mapNode.Id}' failed."); + private static MapBranchOutcome TerminateOutcome(int index, NodeDefinition mapNode, NodeDefinition node, string branchKey) => + new(index, EmptyJsonObject(), Failed: true, TerminateFailure: $"Node '{node.Id}' in map '{mapNode.Id}' failed.", TerminatedAt: new FailedCell(node.Id, branchKey)); /// The disposition of one map body node after it ran: keep walking the branch, ABANDON it with a /// continue-mode failure marker, or TERMINATE the whole map with an unhandled failure. @@ -2370,7 +2398,7 @@ private static bool TrySettleBranch(IReadOnlyDictionary r.Status == NodeStatus.Failure && !HasErrorEdgeInDefinition(body, r.NodeId)); if (terminateRow != null) { - outcome = new MapBranchOutcome(index, EmptyJsonObject(), Failed: true, TerminateFailure: $"Node '{terminateRow.NodeId}' in map '{mapNodeId}' failed."); + outcome = new MapBranchOutcome(index, EmptyJsonObject(), Failed: true, TerminateFailure: $"Node '{terminateRow.NodeId}' in map '{mapNodeId}' failed.", TerminatedAt: new FailedCell(terminateRow.NodeId, branchKey)); return true; } } @@ -2636,7 +2664,7 @@ private async Task RunLoopBodyOnceAsync(WorkflowRun run, Workfl for (var i = 0; i < regular.Count; i++) { MergeNodeOutcome(regular[i], outcomes[i], iterScope, bodyState); - if (AdvanceBodyNodeOrAbandon(regular[i], outcomes[i].Status, bodyState, errorHandling)) + if (AdvanceBodyNodeOrAbandon(regular[i], outcomes[i].Status, bodyState, errorHandling, iterationKey)) return new LoopBodyOutcome(executed, Failed: true); } @@ -2648,7 +2676,7 @@ private async Task RunLoopBodyOnceAsync(WorkflowRun run, Workfl executed++; var status = await ExecuteContainerAsync(run, definition, container, iterScope, bodyState, cancellationToken, iterationKey).ConfigureAwait(false); bodyState.Statuses[container.Id] = status; - if (AdvanceBodyNodeOrAbandon(container, status, bodyState, errorHandling)) + if (AdvanceBodyNodeOrAbandon(container, status, bodyState, errorHandling, iterationKey)) return new LoopBodyOutcome(executed, Failed: true); } } @@ -3949,8 +3977,15 @@ private sealed class NodeFailureException : Exception { public NodeFailureException(string message) : base(message) { } public NodeFailureException(string message, Exception innerException) : base(message, innerException) { } + public NodeFailureException(string message, FailedCell? failedCell) : base(message) { FailedCell = failedCell; } + + /// The cell whose unhandled failure ended the container this ended, when a body node did — what the container reads its failure from. Null for a container that failed on its own (a budget, a binding that is not an array). + public FailedCell? FailedCell { get; } } + /// A node at an iteration key: the cell a container's failure is traced back to. + private sealed record FailedCell(string NodeId, string IterationKey); + /// Thrown to abort the walk when a node parks the run. Caught in RunAfterClaimAsync, which sets Suspended. private sealed class RunSuspendedException : Exception { diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Failures/CellFailureReader.cs b/backend/src/CodeSpace.Core/Services/Workflows/Failures/CellFailureReader.cs new file mode 100644 index 000000000..2661abb54 --- /dev/null +++ b/backend/src/CodeSpace.Core/Services/Workflows/Failures/CellFailureReader.cs @@ -0,0 +1,51 @@ +using CodeSpace.Core.DependencyInjection; +using CodeSpace.Core.Persistence.Db; +using CodeSpace.Messages.Constants; +using CodeSpace.Messages.Failures; +using Microsoft.EntityFrameworkCore; + +namespace CodeSpace.Core.Services.Workflows.Failures; + +/// +/// Reads the provider failure one cell recorded from the two places a node's failure is written — the cell's own node.failed +/// record, then the agent run that worked in it — in the order, and with the parsers, uses for the +/// node that ended a run. So a failure inside a container reads exactly as it would have at the top level of a run. Never throws on a +/// record it cannot read: an old row, or one written before failures were kept, simply has none. +/// +public sealed class CellFailureReader : ICellFailureReader, IScopedDependency +{ + private readonly CodeSpaceDbContext _db; + + public CellFailureReader(CodeSpaceDbContext db) { _db = db; } + + public async Task ReadAsync(Guid runId, Guid teamId, string nodeId, string iterationKey, CancellationToken cancellationToken) + { + var own = await ReadOwnRecordAsync(runId, nodeId, iterationKey, cancellationToken).ConfigureAwait(false); + + return own ?? await ReadAgentRunAsync(runId, teamId, nodeId, iterationKey, cancellationToken).ConfigureAwait(false); + } + + /// The newest node.failed record of the cell: a node that failed on one walk and again on the next says why it failed last. + private async Task ReadOwnRecordAsync(Guid runId, string nodeId, string iterationKey, CancellationToken cancellationToken) + { + var payload = await _db.WorkflowRunRecord.AsNoTracking() + .Where(r => r.RunId == runId && r.RecordType == WorkflowRunRecordTypes.NodeFailed && r.NodeId == nodeId && r.IterationKey == iterationKey) + .OrderByDescending(r => r.Sequence) + .Select(r => r.PayloadJson) + .FirstOrDefaultAsync(cancellationToken).ConfigureAwait(false); + + return PersistedFailureJson.Read(payload); + } + + /// The last agent run of the cell only: an earlier attempt that met a disconnected credential and was respawned says nothing about why the cell ended. + private async Task ReadAgentRunAsync(Guid runId, Guid teamId, string nodeId, string iterationKey, CancellationToken cancellationToken) + { + var result = await _db.AgentRun.AsNoTracking() + .Where(a => a.WorkflowRunId == runId && a.TeamId == teamId && a.NodeId == nodeId && a.IterationKey == iterationKey && a.ResultJson != null) + .OrderByDescending(a => a.CreatedDate) + .Select(a => a.ResultJson) + .FirstOrDefaultAsync(cancellationToken).ConfigureAwait(false); + + return result is null ? null : RunFailureReader.ReadAgentResult(result); + } +} diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Failures/ICellFailureReader.cs b/backend/src/CodeSpace.Core/Services/Workflows/Failures/ICellFailureReader.cs new file mode 100644 index 000000000..73261f056 --- /dev/null +++ b/backend/src/CodeSpace.Core/Services/Workflows/Failures/ICellFailureReader.cs @@ -0,0 +1,20 @@ +using CodeSpace.Messages.Failures; + +namespace CodeSpace.Core.Services.Workflows.Failures; + +/// +/// Which provider failure one cell of a run — a node at an iteration key — is recorded as having failed on. The engine asks it when +/// a container (a map, a loop) ends on the unhandled failure of a node in its body, so the container's own node.failed record +/// can carry the failure of the cell that ended it: answers only for the node that ended the RUN, +/// and for a run that a container ended that node is the container, which spawned no agent run and threw nothing itself. A rerun asks it +/// too, for each failed cell it replays into its fork, so the fork's record of the cell carries what the original run's did. +/// +public interface ICellFailureReader +{ + /// + /// The failure the cell recorded, or null: it recorded none, or it did not fail on a provider connection. What the cell's own latest + /// node.failed record carries wins; failing that, the account of the agent run that last worked in that very cell — never another + /// iteration's, since a branch that failed on a sentence alone says nothing of its sibling's agent run. + /// + Task ReadAsync(Guid runId, Guid teamId, string nodeId, string iterationKey, CancellationToken cancellationToken); +} diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Failures/RunFailureReader.cs b/backend/src/CodeSpace.Core/Services/Workflows/Failures/RunFailureReader.cs index 25f3f54ee..e1130ecf4 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/Failures/RunFailureReader.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/Failures/RunFailureReader.cs @@ -17,7 +17,9 @@ namespace CodeSpace.Core.Services.Workflows.Failures; /// It answers for the node that ENDED the run, and for nothing earlier. A run walks past failures it handles (an error /// edge), so an older record that names a provider failure says only that the run once met one; the run's own error names the node /// that failed it (Node 'x' failed.), and that node's record decides — a failure carried there, else the agent run it spawned. -/// A node that failed for another reason, or a run ended by the reconciler, answers null rather than the first failure found. +/// A node that failed for another reason, or a run ended by the reconciler, answers null rather than the first failure found. +/// When the node that ended the run is a container (a map, a loop), its own record carries the failure of the cell that ended IT — +/// the engine writes it there from — so this reads a container exactly as it reads any node. /// public sealed class RunFailureReader : IRunFailureReader, IScopedDependency { diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Rerun/RerunCellSeeder.cs b/backend/src/CodeSpace.Core/Services/Workflows/Rerun/RerunCellSeeder.cs index 0e6bed356..c02bbe56f 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/Rerun/RerunCellSeeder.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/Rerun/RerunCellSeeder.cs @@ -2,9 +2,11 @@ using CodeSpace.Core.DependencyInjection; using CodeSpace.Core.Persistence.Db; using CodeSpace.Core.Persistence.Entities; +using CodeSpace.Core.Services.Workflows.Failures; using CodeSpace.Core.Services.Workflows.Lifecycle; using CodeSpace.Messages.Constants; using CodeSpace.Messages.Enums; +using CodeSpace.Messages.Failures; using Microsoft.EntityFrameworkCore; namespace CodeSpace.Core.Services.Workflows.Rerun; @@ -53,11 +55,13 @@ public sealed class RerunCellSeeder : IRerunCellSeeder, IScopedDependency private readonly CodeSpaceDbContext _db; private readonly IRunRecordLogger _recordLogger; + private readonly ICellFailureReader _cellFailureReader; - public RerunCellSeeder(CodeSpaceDbContext db, IRunRecordLogger recordLogger) + public RerunCellSeeder(CodeSpaceDbContext db, IRunRecordLogger recordLogger, ICellFailureReader cellFailureReader) { _db = db; _recordLogger = recordLogger; + _cellFailureReader = cellFailureReader; } public async Task SeedKeptCellsAsync(Guid originalRunId, Guid newRunId, IReadOnlySet keptNodeIds, bool writeEmptySnapshotSentinel, CancellationToken cancellationToken) @@ -129,8 +133,11 @@ private async Task ReEmitAsync(Guid newRunId, WorkflowRunNode cell, string itera // A re-emitted Failure cell is one of: a top-level kept failure (error-edge-handled); a map sibling's // continue-mode abandon row; or a map sibling's TERMINATE-mode failed row (both: a failed body node // with no error edge) — all round-trip faithfully: RehydrateFromLedger / TrySettleBranch re-settle - // each from the node.failed record (the terminate arm replays it as a preserved failure, no re-run). - await _recordLogger.NodeFailedAsync(newRunId, cell.NodeId, iterationKey, cell.Error ?? "(failed)", TimeSpan.Zero, cancellationToken).ConfigureAwait(false); + // each from the node.failed record (the terminate arm replays it as a preserved failure, no re-run). The failure the + // original recorded for the cell rides along, so a map the fork fails on that preserved branch answers as the original's did. + var failure = await ReadOriginalFailureAsync(originalRunId, cell, cancellationToken).ConfigureAwait(false); + + await _recordLogger.NodeFailedAsync(newRunId, cell.NodeId, iterationKey, cell.Error ?? "(failed)", TimeSpan.Zero, failure, cancellationToken).ConfigureAwait(false); break; default: @@ -142,6 +149,19 @@ private async Task ReEmitAsync(Guid newRunId, WorkflowRunNode cell, string itera } } + /// + /// What the original run says the cell failed on: its own node.failed record's failure, else the agent run that worked in it — read as the + /// engine reads any cell, so the account is the same wherever it is asked. The agent run belongs to the ORIGINAL run, so a fork that replays the + /// cell without carrying it would never find it: the cell's record on the fork is the only place the fork can hold it. Null when the cell recorded + /// none, so such a record is written exactly as it always was. + /// + private async Task ReadOriginalFailureAsync(Guid originalRunId, WorkflowRunNode cell, CancellationToken cancellationToken) + { + var teamId = await _db.WorkflowRun.AsNoTracking().Where(r => r.Id == originalRunId).Select(r => r.TeamId).SingleAsync(cancellationToken).ConfigureAwait(false); + + return await _cellFailureReader.ReadAsync(originalRunId, teamId, cell.NodeId, cell.IterationKey, cancellationToken).ConfigureAwait(false); + } + private static IReadOnlyDictionary ParseOutputs(string json) { try diff --git a/backend/tests/CodeSpace.E2ETests/Credentials/ProviderFailureInAContainerHttpE2ETests.cs b/backend/tests/CodeSpace.E2ETests/Credentials/ProviderFailureInAContainerHttpE2ETests.cs new file mode 100644 index 000000000..f32fe393e --- /dev/null +++ b/backend/tests/CodeSpace.E2ETests/Credentials/ProviderFailureInAContainerHttpE2ETests.cs @@ -0,0 +1,234 @@ +using System.Net; +using System.Text.Json; +using CodeSpace.Core.Failures; +using CodeSpace.Core.Persistence.Db; +using CodeSpace.Core.Persistence.Entities; +using CodeSpace.Core.Services.Credentials; +using CodeSpace.Core.Services.Workflows.Lifecycle; +using CodeSpace.Messages.Credentials; +using CodeSpace.E2ETests.Infrastructure; +using CodeSpace.IntegrationTests.Infrastructure; +using CodeSpace.Messages.Constants; +using CodeSpace.Messages.Enums; +using CodeSpace.Messages.Exceptions; +using Microsoft.Extensions.DependencyInjection; +using Shouldly; + +namespace CodeSpace.E2ETests.Credentials; + +/// +/// 🟢 High fidelity (HTTP surface) — a team authors a workflow whose flow.map (or flow.loop) body lists pull requests +/// through a GitLab connection whose OAuth grant is dead, runs it, and reads the run the way the SPA does. Everything runs through the +/// real ASP.NET pipeline (JWT, the X-Team-Id scope, the controllers and their JSON options) over real Postgres: the workflow is +/// saved with POST /api/workflows, started with POST /api/workflows/{id}/run, executed by the real engine (the in-test +/// deferred queue is the Hangfire transport) calling the real GitLab provider and credential refresher against a loopback +/// , and then read with GET /api/workflows/runs/{id} as an Owner, a Member and a Viewer. +/// +/// The run that a container ended says what a run that a top-level node ended does: the remedy names the credential and the +/// repository, each role is offered what the endpoint behind the button would accept, and the sentence the run always carried is +/// untouched. A run recorded BEFORE a container carried the failure — its body node recorded one, the container did not — answers +/// exactly as it did: the same sentence, no remedy, and an empty lean read. A map-branch rerun (POST …/rerun-map-branch) forks the run and +/// replays the failed branch it did not choose: the fork fails on it again and answers with the remedy the run it forked did. +/// +[Trait("Category", "E2E")] +[Trait("Surface", "Http")] +public sealed class ProviderFailureInAContainerHttpE2ETests : IClassFixture +{ + private readonly TaskLaunchApiFactory _factory; + private readonly TenantHarness _tenants; + + public ProviderFailureInAContainerHttpE2ETests(TaskLaunchApiFactory factory) + { + _factory = factory; + _tenants = new TenantHarness(factory); + } + + [Theory] + [InlineData("map")] + [InlineData("loop")] + public async Task A_run_a_container_ended_on_a_dead_connection_carries_the_remedy_a_top_level_node_would_for_an_owner_a_member_and_a_viewer(string container) + { + using var gitlab = new GitLabDoorkeeperStub(); // never heard of the pair the row holds: the refresh is answered invalid_grant + var world = await SeedWorldAsync(gitlab); + + var workflowId = Guid.Parse(await ReadStringAsync(await _tenants.SendAsync(HttpMethod.Post, "/api/workflows", world.Owner, world.Team, WorkflowBody(container, world.RepositoryId)), "id")); + var runId = Guid.Parse(await ReadStringAsync(await _tenants.SendAsync(HttpMethod.Post, $"/api/workflows/{workflowId}/run", world.Owner, world.Team, new { }), "runId")); + + await _factory.JobClient.DrainAsync(); + + foreach (var (asker, label, reconnect, relink) in new[] { (world.Owner, "owner", true, true), (world.Member, "member", false, false), (world.Viewer, "viewer", false, false) }) + { + var response = await _tenants.SendAsync(HttpMethod.Get, $"/api/workflows/runs/{runId}", asker, world.Team); + var body = await response.Content.ReadAsStringAsync(); + + response.StatusCode.ShouldBe(HttpStatusCode.OK, $"{label}: {body}"); + + var run = JsonDocument.Parse(body).RootElement; + run.GetProperty("status").GetString().ShouldBe("Failure", $"{label}: fixture check: the dead connection ended the container, which ended the run"); + run.GetProperty("error").GetString().ShouldBe($"Node '{container}' failed.", $"{label}: the sentence is untouched"); + + var remedy = run.GetProperty("remedy"); + remedy.GetProperty("code").GetString().ShouldBe("credential_needs_reconnect", label); + remedy.GetProperty("resolved").GetBoolean().ShouldBeFalse(label); + remedy.GetProperty("title").GetString().ShouldBe("GitLab no longer accepts the credential this run used", label); + remedy.GetProperty("credential").GetProperty("id").GetGuid().ShouldBe(world.CredentialId, label); + remedy.GetProperty("credential").GetProperty("displayName").GetString().ShouldBe("GitLab · NACHO.L", label); + remedy.GetProperty("repositories")[0].GetProperty("fullPath").GetString().ShouldBe("acme/api", label); + + var actions = remedy.GetProperty("actions").EnumerateArray().ToDictionary(a => a.GetProperty("kind").GetString()!, a => a); + actions["reconnect"].GetProperty("enabled").GetBoolean().ShouldBe(reconnect, $"{label} reconnect"); + actions["relink"].GetProperty("enabled").GetBoolean().ShouldBe(relink, $"{label} relink"); + actions["open_providers"].GetProperty("enabled").GetBoolean().ShouldBeTrue(label); + actions["rerun"].GetProperty("enabled").GetBoolean().ShouldBeFalse($"{label}: nothing is fixed yet"); + } + + var lean = await _tenants.SendAsync(HttpMethod.Get, $"/api/workflows/runs/{runId}/remedy", world.Viewer, world.Team); + + lean.StatusCode.ShouldBe(HttpStatusCode.OK, "the room reads the same remedy without the run detail around it"); + JsonDocument.Parse(await lean.Content.ReadAsStringAsync()).RootElement.GetProperty("credential").GetProperty("id").GetGuid().ShouldBe(world.CredentialId); + } + + [Fact] + public async Task A_map_branch_rerun_fork_that_fails_on_a_replayed_failed_branch_carries_the_remedy_of_the_run_it_forked() + { + using var gitlab = new GitLabDoorkeeperStub(); + var world = await SeedWorldAsync(gitlab); + + // Two elements through the one dead connection: the map fails on element 0, the first branch by element order. Re-running element 1 replays that failed branch on the fork. + var workflowId = Guid.Parse(await ReadStringAsync(await _tenants.SendAsync(HttpMethod.Post, "/api/workflows", world.Owner, world.Team, WorkflowBody("map", world.RepositoryId, elements: 2)), "id")); + var runId = Guid.Parse(await ReadStringAsync(await _tenants.SendAsync(HttpMethod.Post, $"/api/workflows/{workflowId}/run", world.Owner, world.Team, new { }), "runId")); + + await _factory.JobClient.DrainAsync(); + + var forkId = Guid.Parse(await ReadStringAsync(await _tenants.SendAsync(HttpMethod.Post, $"/api/workflows/runs/{runId}/rerun-map-branch", world.Owner, world.Team, new { mapNodeId = "map", branchIndex = 1 }), "runId")); + + await _factory.JobClient.DrainAsync(); + + var detail = await _tenants.SendAsync(HttpMethod.Get, $"/api/workflows/runs/{forkId}", world.Owner, world.Team); + var body = await detail.Content.ReadAsStringAsync(); + + detail.StatusCode.ShouldBe(HttpStatusCode.OK, body); + + var fork = JsonDocument.Parse(body).RootElement; + + fork.GetProperty("parentRunId").GetGuid().ShouldBe(runId, "fixture check: this is the fork of the run that failed"); + fork.GetProperty("status").GetString().ShouldBe("Failure", "fixture check: the replayed branch failed, so the map fails again"); + fork.GetProperty("error").GetString().ShouldBe("Node 'map' failed.", "the sentence is untouched"); + + var remedy = fork.GetProperty("remedy"); + + remedy.GetProperty("code").GetString().ShouldBe("credential_needs_reconnect", "the fork answers for the replayed branch as the run it forked did"); + remedy.GetProperty("credential").GetProperty("id").GetGuid().ShouldBe(world.CredentialId); + remedy.GetProperty("repositories")[0].GetProperty("fullPath").GetString().ShouldBe("acme/api"); + } + + [Fact] + public async Task A_container_run_recorded_before_the_container_carried_the_failure_answers_exactly_as_it_did() + { + var world = await SeedWorldAsync(gitlab: null); + var runId = Guid.NewGuid(); + var requestId = Guid.NewGuid(); + var now = DateTimeOffset.UtcNow; + + await _tenants.DbAsync(async db => + { + db.WorkflowRunRequest.Add(new WorkflowRunRequest { Id = requestId, TeamId = world.Team, WorkflowId = null, SourceType = WorkflowRunSourceTypes.Snapshot, ActorType = "user", ActorId = SystemUsers.SeederId, NormalizedPayloadJson = "{}", Status = WorkflowRunRequestStatus.Consumed, ReceivedAt = now, VerifiedAt = now, NormalizedAt = now }); + db.WorkflowRun.Add(new WorkflowRun { Id = runId, WorkflowId = null, WorkflowVersion = null, TeamId = world.Team, RunRequestId = requestId, SourceType = WorkflowRunSourceTypes.Snapshot, Status = WorkflowRunStatus.Failure, Error = "Node 'map' failed.", ScopeRepositoryIds = [], ScopeProjectIds = [], CreatedDate = now, CreatedBy = SystemUsers.SeederId, LastModifiedBy = SystemUsers.SeederId }); + await db.SaveChangesAsync(); + return 0; + }); + + // What a run that failed before containers carried failures left: the body node recorded the failure, the map recorded a sentence alone. + using (var scope = _factory.Services.CreateScope()) + { + var logger = scope.ServiceProvider.GetRequiredService(); + var thrown = new CredentialNeedsReconnectException(world.CredentialId, "GitLab · NACHO.L", "invalid_grant"); + + await logger.NodeStartedAsync(runId, "map", string.Empty, new Dictionary(), new Dictionary(), CancellationToken.None); + await logger.NodeFailedAsync(runId, "list", "map#0", thrown.Message, TimeSpan.FromMilliseconds(4), ProviderFailureCapture.From(thrown, world.RepositoryId), CancellationToken.None); + await logger.NodeFailedAsync(runId, "map", string.Empty, "Node 'list' in map 'map' failed.", TimeSpan.FromMilliseconds(9), CancellationToken.None); + } + + var detail = await _tenants.SendAsync(HttpMethod.Get, $"/api/workflows/runs/{runId}", world.Owner, world.Team); + var run = JsonDocument.Parse(await detail.Content.ReadAsStringAsync()).RootElement; + + detail.StatusCode.ShouldBe(HttpStatusCode.OK); + run.GetProperty("error").GetString().ShouldBe("Node 'map' failed."); + (!run.TryGetProperty("remedy", out var remedy) || remedy.ValueKind == JsonValueKind.Null).ShouldBeTrue("the container recorded no failure, so nothing claims a remedy for a run that ended on it"); + (await _tenants.SendAsync(HttpMethod.Get, $"/api/workflows/runs/{runId}/remedy", world.Owner, world.Team)).StatusCode.ShouldBe(HttpStatusCode.NoContent, "no remedy is an empty answer, not an error"); + } + + // ─── Seeds ───────────────────────────────────────────────────────────────── + + private sealed record World(Guid Team, Guid Owner, Guid Member, Guid Viewer, Guid CredentialId, Guid RepositoryId); + + /// A team of an Owner, a Member and a Viewer; one GitLab connection — an OAuth grant due for renewal that the stub never issued — and one repository running through it. + private async Task SeedWorldAsync(GitLabDoorkeeperStub? gitlab) + { + var owner = await _tenants.AccountAsync(); + var member = await _tenants.AccountAsync(); + var viewer = await _tenants.AccountAsync(); + var team = await _tenants.TeamAsync((owner, TeamRole.Owner), (member, TeamRole.Member), (viewer, TeamRole.Viewer)); + var baseUrl = gitlab?.BaseUrl ?? $"http://127.0.0.1:1/{Guid.NewGuid():N}"; + var instanceId = Guid.NewGuid(); + var credentialId = Guid.NewGuid(); + var repositoryId = Guid.NewGuid(); + + using (var scope = _factory.Services.CreateScope()) + { + var db = scope.ServiceProvider.GetRequiredService(); + var encryptor = scope.ServiceProvider.GetRequiredService(); + var serializer = scope.ServiceProvider.GetRequiredService(); + var payload = new OAuthPayload { AccessToken = "at-0", RefreshToken = "rt-0", ExpiresAt = DateTimeOffset.UtcNow.AddMinutes(-1) }; + + db.ProviderInstance.Add(new ProviderInstance { Id = instanceId, TeamId = team, Provider = ProviderKind.GitLab, DisplayName = "gitlab", BaseUrl = baseUrl, ApiUrl = baseUrl, OauthClientId = GitLabDoorkeeperStub.DefaultClientId, OauthClientSecretEnc = encryptor.Encrypt(GitLabDoorkeeperStub.DefaultClientSecret) }); + db.Credential.Add(new Credential { Id = credentialId, TeamId = team, ProviderInstanceId = instanceId, AuthType = AuthType.OAuth, Ownership = CredentialOwnership.Personal, OwnerUserId = owner, DisplayName = "GitLab · NACHO.L", EncryptedPayload = encryptor.Encrypt(serializer.Serialize(payload)), ExpiresDate = payload.ExpiresAt, Status = CredentialStatus.Active }); + db.Repository.Add(new Repository { Id = repositoryId, TeamId = team, ProviderInstanceId = instanceId, CredentialId = credentialId, ExternalId = Random.Shared.Next(100_000, 999_999).ToString(), NamespacePath = "acme", Name = "api", FullPath = "acme/api", DefaultBranch = "main", Visibility = RepositoryVisibility.Private, WebUrl = $"{baseUrl}/acme/api", CloneUrlHttps = $"{baseUrl}/acme/api.git", Status = RepositoryStatus.Active }); + await db.SaveChangesAsync(); + } + + return new World(team, owner, member, viewer, credentialId, repositoryId); + } + + // start → (body: → list) → end. `list` is the node the dead connection fails; no error edge anywhere, so it ends the container, which ends the run. + // A map runs one branch per element (`elements` of them, each the repository); a loop is given the repository itself. + private static object WorkflowBody(string container, Guid repositoryId, int elements = 1) => new + { + name = $"container-{Guid.NewGuid():N}"[..20], + definition = new + { + schemaVersion = 1, + nodes = container == "map" + ? new object[] + { + new { id = "start", typeKey = "trigger.manual", config = new { }, inputs = new { } }, + new { id = "map", typeKey = "flow.map", config = new { }, inputs = new { items = Enumerable.Repeat(repositoryId, elements).ToArray() } }, + new { id = "ms", typeKey = "flow.map_start", parentId = "map", config = new { }, inputs = new { } }, + new { id = "list", typeKey = "git.list_prs", parentId = "map", config = new { }, inputs = new { repositoryId = "{{item}}" } }, + new { id = "end", typeKey = "builtin.terminal", config = new { }, inputs = new { } }, + } + : new object[] + { + new { id = "start", typeKey = "trigger.manual", config = new { }, inputs = new { } }, + new { id = "loop", typeKey = "flow.loop", config = new { maxIterations = 2 }, inputs = new { } }, + new { id = "ls", typeKey = "flow.loop_start", parentId = "loop", config = new { }, inputs = new { } }, + new { id = "list", typeKey = "git.list_prs", parentId = "loop", config = new { }, inputs = new { repositoryId } }, + new { id = "end", typeKey = "builtin.terminal", config = new { }, inputs = new { } }, + }, + edges = new object[] { new { from = "start", to = container }, new { from = container, to = "end" }, new { from = container == "map" ? "ms" : "ls", to = "list" } }, + }, + activations = Array.Empty(), + enabled = true, + }; + + private static async Task ReadStringAsync(HttpResponseMessage response, string property) + { + var text = await response.Content.ReadAsStringAsync(); + + response.IsSuccessStatusCode.ShouldBeTrue($"{response.RequestMessage?.Method} {response.RequestMessage?.RequestUri?.AbsolutePath} answered {(int)response.StatusCode}: {text}"); + + var value = JsonDocument.Parse(text).RootElement.GetProperty(property); + + return value.ValueKind == JsonValueKind.String ? value.GetString()! : value.GetRawText(); + } +} diff --git a/backend/tests/CodeSpace.IntegrationTests/Credentials/CellFailureReaderFlowTests.cs b/backend/tests/CodeSpace.IntegrationTests/Credentials/CellFailureReaderFlowTests.cs new file mode 100644 index 000000000..79f622955 --- /dev/null +++ b/backend/tests/CodeSpace.IntegrationTests/Credentials/CellFailureReaderFlowTests.cs @@ -0,0 +1,265 @@ +using Autofac; +using CodeSpace.Core.Failures; +using CodeSpace.Core.Persistence.Db; +using CodeSpace.Core.Persistence.Entities; +using CodeSpace.Core.Services.Workflows.Failures; +using CodeSpace.Core.Services.Workflows.Lifecycle; +using CodeSpace.Core.Services.Workflows.Rerun; +using CodeSpace.IntegrationTests.Infrastructure; +using CodeSpace.IntegrationTests.Workflows.Infrastructure; +using CodeSpace.Messages.Agents; +using CodeSpace.Messages.Constants; +using CodeSpace.Messages.Enums; +using CodeSpace.Messages.Exceptions; +using CodeSpace.Messages.Failures; +using Microsoft.EntityFrameworkCore; +using Shouldly; + +namespace CodeSpace.IntegrationTests.Credentials; + +/// +/// 🟢 High fidelity — what one cell (a node at an iteration key) of a run is recorded as having failed on, read from real rows over real +/// Postgres: the ledger written by the real from the real exceptions, and agent runs whose stored result is +/// the one the real executor writes. This is the reader a container asks for the cell that ended it, so what is pinned is its boundary: +/// the cell's own newest record wins, the cell's last agent run is the fallback, and nothing outside the cell — another iteration, another +/// node, another run, another team — is ever read. A cell a rerun carries into its fork keeps its account: the real +/// re-emits a failed cell with the failure the reader answers for it on the original run, so the fork answers for it as the original did. +/// +[Collection(PostgresCollection.Name)] +[Trait("Category", "Integration")] +public class CellFailureReaderFlowTests +{ + private const string Branch = "map#0"; + + private readonly PostgresFixture _fixture; + + public CellFailureReaderFlowTests(PostgresFixture fixture) { _fixture = fixture; } + + [Fact] + public async Task A_cells_own_record_wins_over_its_agent_run() + { + var (teamId, runId) = await SeedRunAsync(); + await RecordFailureAsync(runId, "list", Branch, new CredentialDisconnectedException(Guid.NewGuid(), "own")); + await SeedAgentRunAsync(teamId, runId, "list", Branch, FailureOf(new CredentialNeedsReconnectException(Guid.NewGuid(), "agent", "invalid_grant"))); + + var failure = await ReadAsync(runId, teamId, "list", Branch); + + failure.ShouldNotBeNull(); + failure.Code.ShouldBe(FailureCodes.CredentialDisconnected, "what the node itself recorded is the first account"); + } + + [Fact] + public async Task A_cell_whose_record_carries_no_failure_is_traced_through_the_agent_run_that_last_worked_in_it() + { + var (teamId, runId) = await SeedRunAsync(); + var credentialId = Guid.NewGuid(); + await RecordFailureAsync(runId, "boom", Branch, exception: null); + await SeedAgentRunAsync(teamId, runId, "boom", Branch, FailureOf(new CredentialDisconnectedException(Guid.NewGuid(), "earlier attempt")), createdMinutesAgo: 10); + await SeedAgentRunAsync(teamId, runId, "boom", Branch, FailureOf(new CredentialNeedsReconnectException(credentialId, "last attempt", "invalid_grant")), createdMinutesAgo: 1); + + var failure = await ReadAsync(runId, teamId, "boom", Branch); + + failure.ShouldNotBeNull(); + failure.Code.ShouldBe(FailureCodes.CredentialNeedsReconnect, "only the LAST attempt speaks for the cell: an earlier one that was respawned says nothing about why it ended"); + failure.DetailGuid("credentialId").ShouldBe(credentialId); + + await SeedAgentRunAsync(teamId, runId, "boom", Branch, failure: null, createdMinutesAgo: 0); + + (await ReadAsync(runId, teamId, "boom", Branch)).ShouldBeNull("the newest attempt failed on something else, so the older failures are not why the cell ended"); + } + + [Fact] + public async Task The_newest_failed_record_of_a_cell_is_the_one_that_speaks() + { + var (teamId, runId) = await SeedRunAsync(); + await RecordFailureAsync(runId, "list", Branch, new CredentialDisconnectedException(Guid.NewGuid(), "first walk")); + await RecordFailureAsync(runId, "list", Branch, exception: null); + + (await ReadAsync(runId, teamId, "list", Branch)).ShouldBeNull("the cell failed again, on something else, on a later walk"); + + await RecordFailureAsync(runId, "list", Branch, new CredentialNeedsReconnectException(Guid.NewGuid(), "third walk", "invalid_grant")); + + (await ReadAsync(runId, teamId, "list", Branch))!.Code.ShouldBe(FailureCodes.CredentialNeedsReconnect); + } + + [Fact] + public async Task A_cell_answers_only_for_itself_not_for_another_iteration_node_run_or_team() + { + var (teamId, runId) = await SeedRunAsync(); + var (otherTeamId, otherRunId) = await SeedRunAsync(); + await RecordFailureAsync(runId, "list", "map#1", new CredentialDisconnectedException(Guid.NewGuid(), "another branch")); + await RecordFailureAsync(runId, "other", Branch, new CredentialDisconnectedException(Guid.NewGuid(), "another node")); + await RecordFailureAsync(otherRunId, "list", Branch, new CredentialDisconnectedException(Guid.NewGuid(), "another run")); + await SeedAgentRunAsync(teamId, runId, "list", "map#1", FailureOf(new CredentialDisconnectedException(Guid.NewGuid(), "another branch's agent run"))); + await SeedAgentRunAsync(teamId, runId, "other", Branch, FailureOf(new CredentialDisconnectedException(Guid.NewGuid(), "another node's agent run"))); + await SeedAgentRunAsync(otherTeamId, runId, "list", Branch, FailureOf(new CredentialDisconnectedException(Guid.NewGuid(), "another team's agent run"))); + + (await ReadAsync(runId, teamId, "list", Branch)).ShouldBeNull("nothing in this cell recorded a failure — and nothing outside it is evidence"); + (await ReadAsync(runId, teamId, "list", string.Empty)).ShouldBeNull("the top-level cell of the same node is another cell"); + (await ReadAsync(runId, teamId, "list", "map#1"))!.Code.ShouldBe(FailureCodes.CredentialDisconnected, "fixture check: the reader does read the cell the record belongs to"); + } + + [Fact] + public async Task An_agent_run_of_the_same_node_and_iteration_in_another_run_of_the_team_is_not_this_runs_evidence() + { + var (teamId, runId) = await SeedRunAsync(); + var (_, anotherRunId) = await SeedRunAsync(teamId); // a team runs the same workflow again and again: the node and the iteration key recur in every run of it + await SeedAgentRunAsync(teamId, anotherRunId, "boom", Branch, FailureOf(new CredentialDisconnectedException(Guid.NewGuid(), "another run's agent run"))); + + (await ReadAsync(runId, teamId, "boom", Branch)).ShouldBeNull("the agent run belongs to another run of the team, so it says nothing about why this run's cell ended"); + (await ReadAsync(anotherRunId, teamId, "boom", Branch))!.Code.ShouldBe(FailureCodes.CredentialDisconnected, "fixture check: the reader does read the run the agent run belongs to"); + } + + [Theory] + [InlineData(false, false)] // a kept top-level cell, whose own record carries the failure + [InlineData(false, true)] // a kept top-level cell whose node worded its failure alone: its agent run holds the account + [InlineData(true, false)] // a map's sibling branch cell, whose own record carries the failure + [InlineData(true, true)] // a map's sibling branch cell whose node worded its failure alone + public async Task A_failed_cell_a_rerun_carries_into_its_fork_is_recorded_as_failing_on_what_the_original_was(bool siblingBranch, bool viaTheAgentRun) + { + var (teamId, originalRunId) = await SeedRunAsync(); + var (_, forkRunId) = await SeedRunAsync(teamId); + var iterationKey = siblingBranch ? Branch : string.Empty; + var credentialId = Guid.NewGuid(); + var failing = new CredentialNeedsReconnectException(credentialId, "GitLab · NACHO.L", "invalid_grant"); + + if (viaTheAgentRun) + { + await RecordFailureAsync(originalRunId, "list", iterationKey, exception: null); + await SeedAgentRunAsync(teamId, originalRunId, "list", iterationKey, FailureOf(failing)); + } + else + await RecordFailureAsync(originalRunId, "list", iterationKey, failing); + + (await ReadAsync(forkRunId, teamId, "list", iterationKey)).ShouldBeNull("fixture check: the fork has no account of the cell until the rerun carries it there"); + + await SeedForkAsync(originalRunId, forkRunId, siblingBranch); + + var carried = await ReadAsync(forkRunId, teamId, "list", iterationKey); + + carried.ShouldNotBeNull("the fork's own record of the cell carries what the original run said it failed on — the agent run that worked in it belongs to the original run, not the fork"); + carried.Code.ShouldBe(FailureCodes.CredentialNeedsReconnect); + carried.DetailGuid("credentialId").ShouldBe(credentialId); + (await ReadForkRecordAsync(forkRunId, "list", iterationKey)).Error.ShouldBe("it failed", "the sentence is carried as it was"); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task A_failed_cell_that_recorded_no_failure_is_carried_into_a_fork_exactly_as_it_was(bool siblingBranch) + { + var (teamId, originalRunId) = await SeedRunAsync(); + var (_, forkRunId) = await SeedRunAsync(teamId); + var iterationKey = siblingBranch ? Branch : string.Empty; + + await RecordFailureAsync(originalRunId, "list", iterationKey, exception: null); + await SeedForkAsync(originalRunId, forkRunId, siblingBranch); + + var carried = await ReadForkRecordAsync(forkRunId, "list", iterationKey); + + carried.Error.ShouldBe("it failed"); + carried.HasFailureKey.ShouldBeFalse("a cell with nothing to say about a connection carries no failure key: its record is the one it always was"); + (await ReadAsync(forkRunId, teamId, "list", iterationKey)).ShouldBeNull(); + } + + [Theory] + [InlineData("""{"error":"it failed"}""")] + [InlineData("""{"error":"it failed","failure":"credential_disconnected"}""")] + [InlineData("""{"error":"it failed","failure":{"details":{"credentialId":"x"}}}""")] + [InlineData("""{"error":"it failed","failure":{"code":""}}""")] + public async Task A_record_that_carries_no_readable_failure_is_no_failure_and_never_an_error(string payload) + { + var (teamId, runId) = await SeedRunAsync(); + + using (var scope = _fixture.BeginScope()) + { + var db = scope.Resolve(); + + db.WorkflowRunRecord.Add(new WorkflowRunRecord { Id = Guid.NewGuid(), RunId = runId, RecordType = WorkflowRunRecordTypes.NodeFailed, NodeId = "list", IterationKey = Branch, PayloadJson = payload, OccurredAt = DateTimeOffset.UtcNow }); + await db.SaveChangesAsync(); + } + + (await ReadAsync(runId, teamId, "list", Branch)).ShouldBeNull(); + } + + // ─── Helpers ─────────────────────────────────────────────────────────────── + + private async Task ReadAsync(Guid runId, Guid teamId, string nodeId, string iterationKey) + { + using var scope = _fixture.BeginScope(); + + return await scope.Resolve().ReadAsync(runId, teamId, nodeId, iterationKey, CancellationToken.None); + } + + private static PersistedFailure FailureOf(Exception exception) => ProviderFailureCapture.From(exception, repositoryId: null)!; + + private async Task RecordFailureAsync(Guid runId, string nodeId, string iterationKey, Exception? exception) + { + using var scope = _fixture.BeginScope(); + + await scope.Resolve().NodeFailedAsync(runId, nodeId, iterationKey, "it failed", TimeSpan.FromMilliseconds(5), exception is null ? null : ProviderFailureCapture.From(exception, repositoryId: null), CancellationToken.None); + } + + private async Task SeedAgentRunAsync(Guid teamId, Guid runId, string nodeId, string iterationKey, PersistedFailure? failure, int createdMinutesAgo = 1) + { + using var scope = _fixture.BeginScope(); + var db = scope.Resolve(); + var at = DateTimeOffset.UtcNow.AddMinutes(-createdMinutesAgo); + var result = new AgentRunResult { Status = AgentRunStatus.Failed, ExitReason = "executor_error", Failure = failure }; + + db.AgentRun.Add(new AgentRun + { + Id = Guid.NewGuid(), TeamId = teamId, WorkflowRunId = runId, NodeId = nodeId, IterationKey = iterationKey, Harness = "claude-cli", Status = AgentRunStatus.Failed, + ResultJson = System.Text.Json.JsonSerializer.Serialize(result, CodeSpace.Core.Services.Agents.AgentJson.Options), CompletedAt = at, CreatedDate = at, CreatedBy = SystemUsers.SeederId, LastModifiedBy = SystemUsers.SeederId, + }); + await db.SaveChangesAsync(); + } + + /// What the real seeder does for a rerun: re-emits the original's settled cells onto the fork — a kept top-level cell, or the sibling branch cells of a map whose other branch is the one re-run. + private async Task SeedForkAsync(Guid originalRunId, Guid forkRunId, bool siblingBranch) + { + using var scope = _fixture.BeginScope(); + var seeder = scope.Resolve(); + + if (siblingBranch) + await seeder.SeedSiblingBranchCellsAsync(originalRunId, forkRunId, "map", new HashSet { 1 }, branchCount: 2, CancellationToken.None); + else + await seeder.SeedKeptCellsAsync(originalRunId, forkRunId, new HashSet { "list" }, writeEmptySnapshotSentinel: false, CancellationToken.None); + } + + private sealed record ForkRecord(string Error, bool HasFailureKey); + + /// The one node.failed record the fork holds for the cell, as written. + private async Task ReadForkRecordAsync(Guid forkRunId, string nodeId, string iterationKey) + { + using var scope = _fixture.BeginScope(); + var payload = await scope.Resolve().WorkflowRunRecord.AsNoTracking() + .Where(r => r.RunId == forkRunId && r.RecordType == WorkflowRunRecordTypes.NodeFailed && r.NodeId == nodeId && r.IterationKey == iterationKey) + .Select(r => r.PayloadJson).SingleAsync(); + using var document = System.Text.Json.JsonDocument.Parse(payload); + + return new ForkRecord(document.RootElement.GetProperty("error").GetString()!, document.RootElement.TryGetProperty(PersistedFailureJson.PayloadKey, out _)); + } + + /// A failed run of a team — its own unless one is given — the ledger and agent runs are what the reader looks at, so the run needs no workflow. + private async Task<(Guid TeamId, Guid RunId)> SeedRunAsync(Guid? existingTeamId = null) + { + var teamId = existingTeamId ?? (await WorkflowsTestSeed.SeedTeamAsync(_fixture)).TeamId; + + using var scope = _fixture.BeginScope(); + var db = scope.Resolve(); + var requestId = Guid.NewGuid(); + var runId = Guid.NewGuid(); + var now = DateTimeOffset.UtcNow; + + db.WorkflowRunRequest.Add(new WorkflowRunRequest { Id = requestId, TeamId = teamId, SourceType = WorkflowRunSourceTypes.Snapshot, ActorType = "user", ActorId = SystemUsers.SeederId, NormalizedPayloadJson = "{}", Status = WorkflowRunRequestStatus.Consumed, ReceivedAt = now, VerifiedAt = now, NormalizedAt = now }); + db.WorkflowRun.Add(new WorkflowRun + { + Id = runId, TeamId = teamId, RunRequestId = requestId, SourceType = WorkflowRunSourceTypes.Snapshot, Status = WorkflowRunStatus.Failure, Error = "Node 'map' failed.", + ScopeRepositoryIds = [], ScopeProjectIds = [], DefinitionSnapshotJson = "{\"nodes\":[],\"edges\":[]}", DefinitionSnapshotHash = "sha256:test", CreatedBy = SystemUsers.SeederId, LastModifiedBy = SystemUsers.SeederId, + }); + await db.SaveChangesAsync(); + + return (teamId, runId); + } +} diff --git a/backend/tests/CodeSpace.IntegrationTests/Credentials/ProviderFailureThroughTheEngineFlowTests.Containers.cs b/backend/tests/CodeSpace.IntegrationTests/Credentials/ProviderFailureThroughTheEngineFlowTests.Containers.cs new file mode 100644 index 000000000..53774b1c8 --- /dev/null +++ b/backend/tests/CodeSpace.IntegrationTests/Credentials/ProviderFailureThroughTheEngineFlowTests.Containers.cs @@ -0,0 +1,393 @@ +using System.Text.Json; +using Autofac; +using CodeSpace.Core.Failures; +using CodeSpace.Core.Persistence.Db; +using CodeSpace.Core.Persistence.Entities; +using CodeSpace.Core.Services.Workflows.Failures; +using CodeSpace.IntegrationTests.Infrastructure; +using CodeSpace.IntegrationTests.Workflows.Infrastructure; +using CodeSpace.Core.Services.Workflows.Engine; +using CodeSpace.Messages.Agents; +using CodeSpace.Messages.Constants; +using CodeSpace.Messages.Dtos.Credentials; +using CodeSpace.Messages.Dtos.Workflows; +using CodeSpace.Messages.Enums; +using CodeSpace.Messages.Exceptions; +using CodeSpace.Messages.Failures; +using Microsoft.EntityFrameworkCore; +using Shouldly; + +namespace CodeSpace.IntegrationTests.Credentials; + +/// +/// 🟢 High fidelity — a provider connection that fails INSIDE a container (a flow.map branch, a flow.loop pass, either one +/// nested in the other) and ends the container. The real engine walks real git.list_prs nodes against the real GitLab provider, the +/// real credential resolver and OAuth refresher over real Postgres; the one fake is GitLab (). What is +/// pinned is where the failure is written down: the container's own node.failed record carries the failure of the cell that ended it, +/// so the run's remedy is the one a top-level node of the same failure gets — and carries nothing when the cell that ended the container +/// did not fail on a connection, however many provider failures the run handled on the way. +/// +/// The cell is chosen deterministically. A terminate map whose branches fail concurrently answers for the first failing branch +/// BY ELEMENT ORDER — the same branch whose node the map's sentence names — never for whichever finished first. A branch that failed on an +/// earlier walk and was replayed from the ledger (its sibling parked, so the map could not fail until a later walk) is still the cell that +/// ended the map, and its failure is read back from its own record. A cell whose node worded its failure without a typed cause falls back to +/// the agent run that last worked in that very cell — never to another branch's. +/// +public partial class ProviderFailureThroughTheEngineFlowTests +{ + public enum ContainerShape { Map, Loop, LoopInMap, MapInLoop } + + // shape → the containers that failed with the sentence each one wrote (innermost first); the outermost is the one the run's own error names. + private static (string Id, string Sentence)[] ContainersOf(ContainerShape shape) => shape switch + { + ContainerShape.Map => [("map", "Node 'list' in map 'map' failed.")], + ContainerShape.Loop => [("loop", "Node 'list' failed.")], + ContainerShape.LoopInMap => [("loop", "Node 'list' failed."), ("map", "Node 'loop' in map 'map' failed.")], + _ => [("map", "Node 'list' in map 'map' failed."), ("loop", "Node 'map' failed.")], + }; + + [Theory] + [InlineData(ContainerShape.Map)] + [InlineData(ContainerShape.Loop)] + [InlineData(ContainerShape.LoopInMap)] + [InlineData(ContainerShape.MapInLoop)] + public async Task A_provider_failure_that_ends_a_container_is_recorded_on_every_container_it_ended_and_the_remedy_is_the_one_a_top_level_node_gets(ContainerShape shape) + { + using var gitlab = new GitLabDoorkeeperStub(); // never heard of the pair the row holds: the refresh is answered invalid_grant + var team = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var seed = await GitLabCredentialSeed.SeedAsync(_fixture, gitlab.BaseUrl, "at-0", "rt-0", expired: true, existingTeam: team); + await RecordThePersonItWasConnectedForAsync(seed.CredentialId, team.UserId); + + var runId = await RunContainerAsync(team, ContainerDefinition(shape, seed.RepositoryId), seed.RepositoryId); + + var records = await ReadFailedRecordsAsync(runId); + var containers = ContainersOf(shape); + + foreach (var (id, sentence) in containers) + { + var container = records.Single(r => r.NodeId == id); + + container.Error.ShouldBe(sentence, $"{id}: the sentence the container always wrote is untouched"); + container.Failure.ShouldNotBeNull($"{id}: the container's own record carries the failure of the cell that ended it"); + container.Failure.Code.ShouldBe(FailureCodes.CredentialNeedsReconnect, id); + container.Failure.DetailGuid("credentialId").ShouldBe(seed.CredentialId, id); + container.Failure.RepositoryId.ShouldBe(seed.RepositoryId, $"{id}: the repository the failing node declared it acted on travels with it"); + } + + records.Single(r => r.NodeId == "list").Failure!.Code.ShouldBe(FailureCodes.CredentialNeedsReconnect, "fixture check: the node's own record said it first"); + (await ReadRunErrorAsync(runId)).ShouldBe($"Node '{containers[^1].Id}' failed.", "the run's own sentence is untouched"); + + var failure = await ReadRecordedFailureAsync(runId); + + failure.Code.ShouldBe(FailureCodes.CredentialNeedsReconnect, "the run's failure is the one the container that ended it recorded"); + + var remedy = await RemedyForAsync(team.UserId, team.TeamId, runId); + + remedy.Code.ShouldBe(FailureCodes.CredentialNeedsReconnect); + remedy.Resolved.ShouldBeFalse(); + remedy.Credential!.Id.ShouldBe(seed.CredentialId); + remedy.Repositories.ShouldHaveSingleItem().FullPath.ShouldBe("acme/api"); + remedy.Title.ShouldBe("GitLab no longer accepts the credential this run used", "the words a top-level node of the same failure gets"); + remedy.Actions.Select(a => (a.Kind, a.Enabled)).ShouldBe(new[] + { + (RemedyActionKind.Reconnect, true), (RemedyActionKind.Relink, true), (RemedyActionKind.OpenProviders, true), (RemedyActionKind.Rerun, false) + }, "and the buttons: an Owner may do everything but run it again — that waits for the fix"); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task A_map_whose_branches_fail_on_different_credentials_answers_for_the_first_branch_by_element_order(bool theFirstElementIsTheFirstCredential) + { + using var gitlab = new GitLabDoorkeeperStub(); + using var otherGitlab = new GitLabDoorkeeperStub(); // a second provider instance: a team holds one per address + var team = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var first = await GitLabCredentialSeed.SeedAsync(_fixture, gitlab.BaseUrl, "at-0", "rt-0", expired: true, existingTeam: team); + var second = await GitLabCredentialSeed.SeedAsync(_fixture, otherGitlab.BaseUrl, "at-1", "rt-1", expired: true, existingTeam: team); + var (elementZero, elementOne) = theFirstElementIsTheFirstCredential ? (first, second) : (second, first); + + var runId = await RunContainerAsync(team, ContainerDefinition(ContainerShape.Map, elementZero.RepositoryId), elementZero.RepositoryId, elementOne.RepositoryId); + + var records = await ReadFailedRecordsAsync(runId); + + records.Where(r => r.NodeId == "list").Select(r => (r.IterationKey, r.Failure!.DetailGuid("credentialId"))).ShouldBe(new[] { ("map#0", (Guid?)elementZero.CredentialId), ("map#1", (Guid?)elementOne.CredentialId) }, ignoreOrder: true, customMessage: "fixture check: both branches failed, each on its own credential"); + + var map = records.Single(r => r.NodeId == "map"); + + map.Failure.ShouldNotBeNull("the map's own record carries the failure of the cell that ended it"); + map.Failure.DetailGuid("credentialId").ShouldBe(elementZero.CredentialId, "element 0 — whichever branch happened to finish first, the map's sentence names the first branch by element order and so does its failure"); + map.Failure.RepositoryId.ShouldBe(elementZero.RepositoryId); + (await RemedyForAsync(team.UserId, team.TeamId, runId)).Credential!.Id.ShouldBe(elementZero.CredentialId); + } + + [Theory] + // the container, and the node that fails it for another reason: a node that worded its failure, or a flow.iterate handed something that is not a list + [InlineData(ContainerShape.Map, FlakyTestNode.Key)] + [InlineData(ContainerShape.Loop, FlakyTestNode.Key)] + [InlineData(ContainerShape.Map, "flow.iterate")] + [InlineData(ContainerShape.Loop, "flow.iterate")] + public async Task A_container_that_fails_on_something_else_after_a_provider_failure_was_handled_has_no_remedy_for_the_old_failure(ContainerShape shape, string failingTypeKey) + { + using var gitlab = new GitLabDoorkeeperStub(); + var team = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var seed = await GitLabCredentialSeed.SeedAsync(_fixture, gitlab.BaseUrl, "at-0", "rt-0", expired: false, existingTeam: team); + await DisconnectAsync(seed.CredentialId); + + var runId = await RunContainerAsync(team, HandledThenFailingDefinition(shape, seed.RepositoryId, failingTypeKey), seed.RepositoryId); + + var records = await ReadFailedRecordsAsync(runId); + var container = shape == ContainerShape.Map ? "map" : "loop"; + + records.Single(r => r.NodeId == "list").Failure!.Code.ShouldBe(FailureCodes.CredentialDisconnected, "fixture check: the OLD failure is on the ledger, in the same container, carrying the code the page must not show"); + records.Single(r => r.NodeId == "other").Failure.ShouldBeNull("fixture check: the cell that ended the container failed on something else"); + records.Single(r => r.NodeId == container).Failure.ShouldBeNull("the container answers for the cell that ended it, not for a failure its body handled earlier"); + (await ReadRunErrorAsync(runId)).ShouldBe($"Node '{container}' failed."); + + using var scope = _fixture.BeginScope(); + + (await scope.Resolve().ReadAsync(runId, team.TeamId, CancellationToken.None)).ShouldBeNull(); + (await RemedyForAsync(team.UserId, team.TeamId, runId, allowNone: true)).ShouldBeNull("so no card blames a credential for a run that failed for another reason"); + } + + [Fact] + public async Task A_provider_failure_a_try_caught_leaves_a_successful_run_and_a_try_record_with_no_failure() + { + using var gitlab = new GitLabDoorkeeperStub(); + var team = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var seed = await GitLabCredentialSeed.SeedAsync(_fixture, gitlab.BaseUrl, "at-0", "rt-0", expired: false, existingTeam: team); + await DisconnectAsync(seed.CredentialId); + var workflowId = await CreateWorkflowAsync(team, CaughtByATryDefinition(seed.RepositoryId)); + var runId = await WorkflowsTestSeed.SeedManualRunAsync(_fixture, workflowId, team.TeamId); + + await RunEngineAsync(runId); + + var records = await ReadFailedRecordsAsync(runId); + + using var scope = _fixture.BeginScope(); + + (await scope.Resolve().WorkflowRun.AsNoTracking().SingleAsync(r => r.Id == runId)).Status.ShouldBe(WorkflowRunStatus.Success, "fixture check: the try caught the failure"); + records.Select(r => r.NodeId).ShouldBe(new[] { "list" }, "the try itself did not fail, so it wrote no failed record of its own"); + records[0].Failure!.Code.ShouldBe(FailureCodes.CredentialDisconnected); + (await scope.Resolve().ReadAsync(runId, team.TeamId, CancellationToken.None)).ShouldBeNull("the run did not fail on it"); + (await RemedyForAsync(team.UserId, team.TeamId, runId, allowNone: true)).ShouldBeNull(); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task A_branch_whose_node_worded_its_failure_without_a_cause_is_traced_through_its_own_agent_run_and_never_through_a_siblings(bool theAgentRunIsTheEndingBranchs) + { + using var gitlab = new GitLabDoorkeeperStub(); + var team = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var seed = await GitLabCredentialSeed.SeedAsync(_fixture, gitlab.BaseUrl, "at-0", "rt-0", expired: false, existingTeam: team); + await DisconnectAsync(seed.CredentialId); + var workflowId = await CreateWorkflowAsync(team, FlakyBranchesDefinition(Guid.NewGuid().ToString("N"))); + var runId = await WorkflowsTestSeed.SeedManualRunAsync(_fixture, workflowId, team.TeamId, payloadJson: """{ "things": ["x", "y"] }"""); + + // The agent run the branch's agent.run node would have spawned: the executor's own account of a clone GitLab refused, in the cell it worked in. + await SeedAgentRunAsync(team.TeamId, runId, "boom", theAgentRunIsTheEndingBranchs ? "map#0" : "map#1", ProviderFailureCapture.From(new CredentialDisconnectedException(seed.CredentialId, "GitLab · NACHO.L"), seed.RepositoryId)!); + await RunEngineAsync(runId); + + var records = await ReadFailedRecordsAsync(runId); + var map = records.Single(r => r.NodeId == "map"); + + records.Where(r => r.NodeId == "boom").Select(r => r.IterationKey).ShouldBe(new[] { "map#0", "map#1" }, ignoreOrder: true, customMessage: "fixture check: both branches failed on a sentence alone — no node recorded a failure"); + records.Where(r => r.NodeId == "boom").ShouldAllBe(r => r.Failure == null); + + if (!theAgentRunIsTheEndingBranchs) + { + map.Failure.ShouldBeNull("the map's first branch is element 0, whose cell has no agent run of its own — another branch's account is not why the map ended"); + (await RemedyForAsync(team.UserId, team.TeamId, runId, allowNone: true)).ShouldBeNull(); + return; + } + + map.Failure.ShouldNotBeNull("the cell that ended the map has an agent run, as a top-level agent.run node would"); + map.Failure.Code.ShouldBe(FailureCodes.CredentialDisconnected); + map.Failure.DetailGuid("credentialId").ShouldBe(seed.CredentialId); + (await RemedyForAsync(team.UserId, team.TeamId, runId)).Credential!.Id.ShouldBe(seed.CredentialId); + } + + [Fact] + public async Task A_branch_that_failed_on_an_earlier_walk_while_a_sibling_was_parked_is_still_the_cell_that_ended_the_map_when_it_fails_on_the_next() + { + using var gitlab = new GitLabDoorkeeperStub(); + using var otherGitlab = new GitLabDoorkeeperStub(); // a second provider instance: a team holds one per address + var key = "park-" + Guid.NewGuid().ToString("N"); + SuspendProbeNode.Reset(key); + var team = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var first = await GitLabCredentialSeed.SeedAsync(_fixture, gitlab.BaseUrl, "at-0", "rt-0", expired: true, existingTeam: team); + var second = await GitLabCredentialSeed.SeedAsync(_fixture, otherGitlab.BaseUrl, "at-1", "rt-1", expired: true, existingTeam: team); + var workflowId = await CreateWorkflowAsync(team, ParkThenListDefinition(key)); + var runId = await WorkflowsTestSeed.SeedManualRunAsync(_fixture, workflowId, team.TeamId, payloadJson: JsonSerializer.Serialize(new { things = new[] { first.RepositoryId, second.RepositoryId } })); + + await RunEngineAsync(runId); + await AssertRunStatusAsync(runId, WorkflowRunStatus.Suspended, "both branches parked"); + + // Branch 0 resumes and fails on its connection; branch 1 is still parked, so the map cannot fail yet — it parks again, holding branch 0's failure in its ledger. + await ResumeBranchAsync(runId, key, first.RepositoryId); + await RunEngineAsync(runId); + await AssertRunStatusAsync(runId, WorkflowRunStatus.Suspended, "the failure waits for the parked sibling"); + (await ReadFailedRecordsAsync(runId)).Select(r => r.NodeId).ShouldBe(new[] { "list" }, "fixture check: branch 0 failed, the map has not"); + + // Branch 1 resumes and fails too. Branch 0 is REPLAYED from the ledger now, not re-run — and it is still the branch the map answers for. + await ResumeBranchAsync(runId, key, second.RepositoryId); + await RunEngineAsync(runId); + await AssertRunStatusAsync(runId, WorkflowRunStatus.Failure, "both branches failed"); + + var records = await ReadFailedRecordsAsync(runId); + + records.Where(r => r.NodeId == "list").Select(r => r.IterationKey).ShouldBe(new[] { "map#0", "map#1" }, "fixture check: branch 0 did not run its node again"); + var map = records.Single(r => r.NodeId == "map"); + + map.Failure.ShouldNotBeNull("the map's own record carries the failure of the cell that ended it"); + map.Failure.DetailGuid("credentialId").ShouldBe(first.CredentialId, "branch 0's failure was written on an earlier walk and is read back from its own record"); + (await RemedyForAsync(team.UserId, team.TeamId, runId)).Credential!.Id.ShouldBe(first.CredentialId); + } + + // ─── Helpers ─────────────────────────────────────────────────────────────── + + private sealed record FailedRecord(string NodeId, string IterationKey, string Error, PersistedFailure? Failure); + + private async Task RunContainerAsync((Guid TeamId, Guid UserId) team, WorkflowDefinition definition, params Guid[] elements) + { + var workflowId = await CreateWorkflowAsync(team, definition); + var runId = await WorkflowsTestSeed.SeedManualRunAsync(_fixture, workflowId, team.TeamId, payloadJson: JsonSerializer.Serialize(new { things = elements })); + + await RunEngineAsync(runId); + await AssertRunStatusAsync(runId, WorkflowRunStatus.Failure, "fixture check: the connection's failure ended the container, which failed the run"); + + return runId; + } + + private async Task AssertRunStatusAsync(Guid runId, WorkflowRunStatus status, string because) + { + using var scope = _fixture.BeginScope(); + + (await scope.Resolve().WorkflowRun.AsNoTracking().SingleAsync(r => r.Id == runId)).Status.ShouldBe(status, because); + } + + /// Every node.failed record of the run, oldest first, with the failure it carries (none for a record that carries none). + private async Task> ReadFailedRecordsAsync(Guid runId) + { + using var scope = _fixture.BeginScope(); + var rows = await scope.Resolve().WorkflowRunRecord.AsNoTracking().Where(r => r.RunId == runId && r.RecordType == WorkflowRunRecordTypes.NodeFailed).OrderBy(r => r.Sequence).Select(r => new { r.NodeId, r.IterationKey, r.PayloadJson }).ToListAsync(); + + return rows.Select(r => new FailedRecord(r.NodeId!, r.IterationKey, JsonDocument.Parse(r.PayloadJson).RootElement.GetProperty("error").GetString()!, PersistedFailureJson.Read(r.PayloadJson))).ToList(); + } + + private async Task SeedAgentRunAsync(Guid teamId, Guid runId, string nodeId, string iterationKey, PersistedFailure failure) + { + using var scope = _fixture.BeginScope(); + var db = scope.Resolve(); + var at = DateTimeOffset.UtcNow.AddMinutes(-1); + var result = new AgentRunResult { Status = AgentRunStatus.Failed, ExitReason = "executor_error", Failure = failure }; + + db.AgentRun.Add(new AgentRun + { + Id = Guid.NewGuid(), TeamId = teamId, WorkflowRunId = runId, NodeId = nodeId, IterationKey = iterationKey, Harness = "claude-cli", Status = AgentRunStatus.Failed, + ResultJson = JsonSerializer.Serialize(result, CodeSpace.Core.Services.Agents.AgentJson.Options), CompletedAt = at, CreatedDate = at, CreatedBy = SystemUsers.SeederId, LastModifiedBy = SystemUsers.SeederId, + }); + await db.SaveChangesAsync(); + } + + /// Resolves the parked branch whose element is the repository — the single-wait path an approval or a callback resumes by, which re-walks at once with a sibling still parked. + private async Task ResumeBranchAsync(Guid runId, string key, Guid element) + { + using var scope = _fixture.BeginScope(); + var waitId = await scope.Resolve().WorkflowRunWait.AsNoTracking().Where(w => w.RunId == runId && w.Token == $"{key}::{element}" && w.Status == WorkflowWaitStatuses.Pending).Select(w => w.Id).SingleAsync(); + + (await scope.Resolve().ResumeWaitAsync(runId, waitId, """{"summary":"go"}""", CancellationToken.None)).ShouldBeTrue(); + } + + // ─── Definitions ─────────────────────────────────────────────────────────── + + private static NodeDefinition Node(string id, string typeKey, string? parentId, string config = "{}", string inputs = "{}") => + new() { Id = id, TypeKey = typeKey, ParentId = parentId, Config = WorkflowsTestSeed.Json(config), Inputs = WorkflowsTestSeed.Json(inputs) }; + + private static NodeDefinition ListIn(string parentId, string repository) => Node("list", "git.list_prs", parentId, inputs: $$"""{"repositoryId":"{{repository}}"}"""); + + private static NodeDefinition MapNode(string id, string? parentId) => Node(id, "flow.map", parentId, inputs: """{ "items": "{{trigger.things}}" }"""); + + private static NodeDefinition LoopNode(string id, string? parentId) => Node(id, "flow.loop", parentId, config: """{ "maxIterations": 2 }"""); + + private static EdgeDefinition Edge(string from, string to, string? handle = null) => new() { From = from, To = to, SourceHandle = handle }; + + // start → → end. The container's body ends in `list`, the node the dead connection fails; no error edge anywhere, so it ends the container, which ends the run. + // `list` is given {{item}} where an enclosing map supplies the repository, and the repository itself under a loop alone. + private static WorkflowDefinition ContainerDefinition(ContainerShape shape, Guid onlyRepository) + { + const string repository = "{{item}}"; + var literal = onlyRepository.ToString(); + var nodes = new List { Node("start", "trigger.manual", null), Node("end", "builtin.terminal", null) }; + var edges = new List(); + + switch (shape) + { + case ContainerShape.Map: + nodes.AddRange([MapNode("map", null), Node("ms", "flow.map_start", "map"), ListIn("map", repository)]); + edges.AddRange([Edge("start", "map"), Edge("map", "end"), Edge("ms", "list")]); + break; + case ContainerShape.Loop: + nodes.AddRange([LoopNode("loop", null), Node("ls", "flow.loop_start", "loop"), ListIn("loop", literal)]); + edges.AddRange([Edge("start", "loop"), Edge("loop", "end"), Edge("ls", "list")]); + break; + case ContainerShape.LoopInMap: + nodes.AddRange([MapNode("map", null), Node("ms", "flow.map_start", "map"), LoopNode("loop", "map"), Node("ls", "flow.loop_start", "loop"), ListIn("loop", repository)]); + edges.AddRange([Edge("start", "map"), Edge("map", "end"), Edge("ms", "loop"), Edge("ls", "list")]); + break; + default: + nodes.AddRange([LoopNode("loop", null), Node("ls", "flow.loop_start", "loop"), MapNode("map", "loop"), Node("ms", "flow.map_start", "map"), ListIn("map", repository)]); + edges.AddRange([Edge("start", "loop"), Edge("loop", "end"), Edge("ls", "map"), Edge("ms", "list")]); + break; + } + + return new WorkflowDefinition { SchemaVersion = 1, Nodes = nodes, Edges = edges }; + } + + // start → (body: → list =(error)=> other) → end. `list` fails on its connection and its error edge handles that; `other` — the body's terminal — then fails on its own, with no cause a person can repair. + private static WorkflowDefinition HandledThenFailingDefinition(ContainerShape shape, Guid onlyRepository, string failingTypeKey) + { + const string repository = "{{item}}"; + var map = shape == ContainerShape.Map; + var container = map ? "map" : "loop"; + var start = map ? "ms" : "ls"; + var other = failingTypeKey == FlakyTestNode.Key + ? Node("other", FlakyTestNode.Key, container, config: $$"""{ "key": "{{Guid.NewGuid():N}}", "failTimes": 99 }""") + : Node("other", failingTypeKey, container, config: """{ "template": "x" }""", inputs: """{ "items": "not a list" }"""); + + return new WorkflowDefinition + { + SchemaVersion = 1, + Nodes = [Node("start", "trigger.manual", null), map ? MapNode("map", null) : LoopNode("loop", null), Node(start, map ? "flow.map_start" : "flow.loop_start", container), ListIn(container, map ? repository : onlyRepository.ToString()), other, Node("end", "builtin.terminal", null)], + Edges = [Edge("start", container), Edge(container, "end"), Edge(start, "list"), Edge("list", "other", WorkflowHandles.Error)], + }; + } + + // start → try(body: → list) =(catch)=> caught; try → done. The try catches the connection's failure. + private static WorkflowDefinition CaughtByATryDefinition(Guid repositoryId) => new() + { + SchemaVersion = 1, + Nodes = [Node("start", "trigger.manual", null), Node("try", "flow.try", null), Node("ts", "flow.try_start", "try"), ListIn("try", repositoryId.ToString()), Node("done", "builtin.terminal", null), Node("caught", "builtin.terminal", null)], + Edges = [Edge("start", "try"), Edge("ts", "list"), Edge("try", "done"), Edge("try", "caught", WorkflowHandles.Catch)], + }; + + // start → map(body: → boom[fails on a sentence alone]) → end. + private static WorkflowDefinition FlakyBranchesDefinition(string flakyKey) => new() + { + SchemaVersion = 1, + Nodes = [Node("start", "trigger.manual", null), MapNode("map", null), Node("ms", "flow.map_start", "map"), Node("boom", FlakyTestNode.Key, "map", config: $$"""{ "key": "{{flakyKey}}", "failTimes": 99 }"""), Node("end", "builtin.terminal", null)], + Edges = [Edge("start", "map"), Edge("map", "end"), Edge("ms", "boom")], + }; + + // start → map(body: → park[parks each branch's wait, resumes with a summary] → list) → end. Each element is a repository; `list` is the node its dead connection fails. + private static WorkflowDefinition ParkThenListDefinition(string key) => new() + { + SchemaVersion = 1, + Nodes = + [ + Node("start", "trigger.manual", null), MapNode("map", null), Node("ms", "flow.map_start", "map"), + Node("park", SuspendProbeNode.Key, "map", inputs: $$$"""{ "key": "{{{key}}}", "item": "{{item}}" }"""), ListIn("map", "{{item}}"), Node("end", "builtin.terminal", null), + ], + Edges = [Edge("start", "map"), Edge("map", "end"), Edge("ms", "park"), Edge("park", "list")], + }; +} diff --git a/backend/tests/CodeSpace.IntegrationTests/Credentials/ProviderFailureThroughTheEngineFlowTests.Rerun.cs b/backend/tests/CodeSpace.IntegrationTests/Credentials/ProviderFailureThroughTheEngineFlowTests.Rerun.cs new file mode 100644 index 000000000..51c8fd429 --- /dev/null +++ b/backend/tests/CodeSpace.IntegrationTests/Credentials/ProviderFailureThroughTheEngineFlowTests.Rerun.cs @@ -0,0 +1,94 @@ +using System.Text.Json; +using Autofac; +using CodeSpace.Core.Failures; +using CodeSpace.Core.Services.Workflows; +using CodeSpace.IntegrationTests.Infrastructure; +using CodeSpace.IntegrationTests.Workflows.Infrastructure; +using CodeSpace.Messages.Enums; +using CodeSpace.Messages.Exceptions; +using CodeSpace.Messages.Failures; +using Shouldly; + +namespace CodeSpace.IntegrationTests.Credentials; + +/// +/// 🟢 High fidelity — a map-branch rerun forks the run and replays the sibling branches that were not chosen, a failed sibling among them +/// (a terminate map keeps one: the branch that ended it). The fork walks them through the real engine, so the map that fails on the replayed +/// branch must answer for it as the original run did: the cell the fork replays is re-emitted by the real seeder with the failure the original +/// recorded for it — on its own record, or on the agent run that worked in it, which belongs to the original run and not the fork — and the run +/// the fork ends on says what to repair. The fork's one fresh branch fails on its own connection, so a remedy that named THAT credential would be +/// the wrong branch's. +/// +public partial class ProviderFailureThroughTheEngineFlowTests +{ + [Theory] + [InlineData(false)] // the replayed branch's node recorded its own failure + [InlineData(true)] // the replayed branch's node worded its failure alone: the original run's agent run holds the account + public async Task A_map_branch_rerun_fork_that_fails_on_a_replayed_failed_branch_answers_with_the_remedy_the_original_run_did(bool viaTheAgentRun) + { + using var gitlab = new GitLabDoorkeeperStub(); + using var otherGitlab = new GitLabDoorkeeperStub(); // a second provider instance: a team holds one per address + var team = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var first = await GitLabCredentialSeed.SeedAsync(_fixture, gitlab.BaseUrl, "at-0", "rt-0", expired: !viaTheAgentRun, existingTeam: team); + var second = await GitLabCredentialSeed.SeedAsync(_fixture, otherGitlab.BaseUrl, "at-1", "rt-1", expired: true, existingTeam: team); + if (viaTheAgentRun) await DisconnectAsync(first.CredentialId); + + var originalRunId = await RunAdmittedMapAsync(team, viaTheAgentRun, first, second); + + (await ReadFailedRecordsAsync(originalRunId)).Single(r => r.NodeId == "map").Failure!.DetailGuid("credentialId").ShouldBe(first.CredentialId, "fixture check: the original run's map answers for element 0, the first branch to fail by element order"); + + var forkRunId = await RerunBranchAsync(team, originalRunId, branchIndex: 1); + + await RunEngineAsync(forkRunId); + await AssertRunStatusAsync(forkRunId, WorkflowRunStatus.Failure, "fixture check: element 0 is replayed failed, so the map fails again"); + + var records = await ReadFailedRecordsAsync(forkRunId); + var node = viaTheAgentRun ? "boom" : "list"; + + records.Single(r => r.NodeId == node && r.IterationKey == "map#0").Failure.ShouldNotBeNull("the replayed branch's record on the fork says what the original run said it failed on"); + + var fresh = records.Single(r => r.NodeId == node && r.IterationKey == "map#1"); + + if (viaTheAgentRun) fresh.Failure.ShouldBeNull("fixture check: the fresh branch failed too, on a sentence alone"); + else fresh.Failure!.DetailGuid("credentialId").ShouldBe(second.CredentialId, "fixture check: the fresh branch failed on its own connection, which is not the one the map answers for"); + + var map = records.Single(r => r.NodeId == "map"); + + map.Error.ShouldBe($"Node '{node}' in map 'map' failed.", "the sentence the map always wrote"); + map.Failure.ShouldNotBeNull("the fork's map answers for the replayed branch it failed on, as the original run's did"); + map.Failure.DetailGuid("credentialId").ShouldBe(first.CredentialId, "element 0's credential — never the fresh branch's"); + + var remedy = await RemedyForAsync(team.UserId, team.TeamId, forkRunId); + + remedy.Code.ShouldBe(viaTheAgentRun ? FailureCodes.CredentialDisconnected : FailureCodes.CredentialNeedsReconnect); + remedy.Credential!.Id.ShouldBe(first.CredentialId); + } + + // ─── Helpers ─────────────────────────────────────────────────────────────── + + /// + /// An ADMITTED run (the one a rerun forks) of a terminate map over two elements, element i being Elements[i]'s repository, that fails. With + /// each branch's node words its failure alone and element 0's cell has the agent run that holds the account. + /// + private async Task RunAdmittedMapAsync((Guid TeamId, Guid UserId) team, bool viaTheAgentRun, SeededGitLabCredential first, SeededGitLabCredential second) + { + var definition = viaTheAgentRun ? FlakyBranchesDefinition(Guid.NewGuid().ToString("N")) : ContainerDefinition(ContainerShape.Map, first.RepositoryId); + var workflowId = await CreateWorkflowAsync(team, definition); + var runId = await WorkflowsTestSeed.SeedAdmittedManualRunAsync(_fixture, workflowId, team.TeamId, payloadJson: JsonSerializer.Serialize(new { things = new[] { first.RepositoryId, second.RepositoryId } })); + + if (viaTheAgentRun) + await SeedAgentRunAsync(team.TeamId, runId, "boom", "map#0", ProviderFailureCapture.From(new CredentialDisconnectedException(first.CredentialId, "GitLab · NACHO.L"), first.RepositoryId)!); + + await RunEngineAsync(runId); + await AssertRunStatusAsync(runId, WorkflowRunStatus.Failure, "fixture check: the map failed"); + + return runId; + } + + private async Task RerunBranchAsync((Guid TeamId, Guid UserId) team, Guid originalRunId, int branchIndex) + { + using var scope = _fixture.BeginScope(); + + return await scope.Resolve().RerunMapBranchAsync(originalRunId, "map", branchIndex, team.TeamId, team.UserId, operationId: null, CancellationToken.None); + } +} diff --git a/backend/tests/CodeSpace.IntegrationTests/Credentials/ProviderFailureThroughTheEngineFlowTests.cs b/backend/tests/CodeSpace.IntegrationTests/Credentials/ProviderFailureThroughTheEngineFlowTests.cs index fec36e6be..570d2e69a 100644 --- a/backend/tests/CodeSpace.IntegrationTests/Credentials/ProviderFailureThroughTheEngineFlowTests.cs +++ b/backend/tests/CodeSpace.IntegrationTests/Credentials/ProviderFailureThroughTheEngineFlowTests.cs @@ -38,7 +38,7 @@ namespace CodeSpace.IntegrationTests.Credentials; /// The one fake is GitLab, so the only thing between these tests and production is the provider's own answers. [Collection(PostgresCollection.Name)] [Trait("Category", "Integration")] -public class ProviderFailureThroughTheEngineFlowTests +public partial class ProviderFailureThroughTheEngineFlowTests { private readonly PostgresFixture _fixture; diff --git a/backend/tests/TESTING.md b/backend/tests/TESTING.md index e69807896..edeb5e4f7 100644 --- a/backend/tests/TESTING.md +++ b/backend/tests/TESTING.md @@ -76,6 +76,19 @@ Provider-credential tests share two more pieces, both in `CodeSpace.IntegrationT another node than the one that handled a provider failure has none; a standing lookup that got no answer blames nobody); `ProviderRemedyHttpE2ETests` pins the wire shape for an Owner, a Member and a Viewer. Those fixtures write the failure with the real `IRunRecordLogger` from the real exception through `ProviderFailureCapture`, never a hand-typed payload. +- A provider failure inside a container (a `flow.map` branch, a `flow.loop` pass, either nested in the other) is carried onto the + container's own `node.failed` record, because the run's failure is the one the node that ended it recorded and that node is the + container. `ProviderFailureThroughTheEngineFlowTests.Containers.cs` drives the real engine through a map, a loop and both nestings against + the stub (the sentence each container writes is asserted untouched), and pins the choices that keep it honest: a `terminate` map answers for + the first failing branch by element order (not whichever finished first), a branch replayed from the ledger while a sibling was parked is + still the one that ended the map, a failure the body handled — or a `flow.try` caught — leaves nothing on a container that later failed on + something else, and a cell whose node worded its failure without a cause is traced through its own agent run, never a sibling's. + `CellFailureReaderFlowTests` pins the reader's boundary (the cell's newest record, then its last agent run; nothing outside the cell — not + another run of the same team either) and that the real rerun seeder carries a failed cell's failure into the fork it replays it in, from the + cell's own record or from the original run's agent run, and leaves a failure-less cell's record as it was. A map-branch rerun whose fork fails + again on the replayed failed branch answers with the original run's remedy, never the fresh branch's + (`ProviderFailureThroughTheEngineFlowTests.Rerun.cs`, and over HTTP in `ProviderFailureInAContainerHttpE2ETests`, which also pins what the run + detail and the lean remedy read answer, plus that a run recorded before containers carried the failure answers exactly as it did). - `RefreshPodHarness` — pods as real OS processes (`CodeSpace.StorageTestWorker`'s `oauth-refresh-pod` mode) running the production refresher, or the pinned previous build's (`StorageTestWorker/PreviousBuild`), against the fixture's Postgres. One process shares one refresh semaphore, so only separate processes exercise the cross-pod advisory lock.