Pause (resumable) on a transient infra failure instead of failing

src/Squid.Core/Halibut/Resilience/TransientFailureClassifier.cs

@@ -0,0 +1,62 @@
+using Halibut;
+
+namespace Squid.Core.Halibut.Resilience;
+
+/// <summary>
+/// The single definition of "is this a TRANSIENT infrastructure failure?" — a
+/// blip that should PAUSE a deployment (resumable, re-attach to the still-running
+/// script) rather than fail it terminally. Referenced across layers (the pipeline
+/// runner's pause-classification, the per-target <c>TargetCatchClassifier</c>, the
+/// per-action catch in <c>ExecuteStepsPhase</c>, and the Halibut execution
+/// strategy's re-attach probe), so the definition lives in ONE place.
+///
+/// <para>Transient = a single round-trip to the agent could not be completed, so
+/// the script may still be running and a resume should re-attach to it:</para>
+/// <list type="bullet">
+///   <item><see cref="AgentUnreachableException"/> — the liveness probe gave up
+///         mid-script (the script was dispatched and may still be running).</item>
+///   <item>A base/transport <see cref="HalibutClientException"/> — connection
+///         reset, EOF, TLS handshake failure, timeout, etc.</item>
+/// </list>
+///
+/// <para>NOT transient — these are sustained or permanent conditions, so pausing
+/// for resume would only pause-loop; they must fail the deployment so an operator
+/// investigates:</para>
+/// <list type="bullet">
+///   <item>The Halibut protocol/invocation subtypes — the request reached the
+///         agent and was PERMANENTLY rejected (version mismatch → no matching
+///         service/method; the agent's service itself threw).</item>
+///   <item><see cref="CircuitOpenException"/> — the per-machine breaker only opens
+///         after the failure threshold (3+ consecutive failures), so it signals a
+///         SUSTAINED agent problem, not a one-off blip. It is also raised BEFORE
+///         any script is dispatched (fail-fast), so there is no in-flight script to
+///         re-attach to. Pausing on it would loop on a genuinely dead agent and
+///         flip the deliberate fail-fast-on-open-breaker contract
+///         (<c>HalibutCircuitBreakerE2ETests.BreakerForcedOpen_*</c> → Failed).</item>
+/// </list>
+/// </summary>
+public static class TransientFailureClassifier
+{
+    public static bool IsTransient(Exception ex)
+    {
+        if (ex is AggregateException aggregate)
+            return aggregate.InnerExceptions.Count > 0 && aggregate.InnerExceptions.All(IsTransient);
+
+        if (ex is AgentUnreachableException)
+            return true;
+
+        // Halibut transport failures are transient EXCEPT the protocol/invocation
+        // subtypes (the request reached the agent and was permanently rejected).
+        if (ex is HalibutClientException)
+            return !IsPermanentHalibutFailure(ex);
+
+        return false;
+    }
+
+    private static bool IsPermanentHalibutFailure(Exception ex)
+        => ex is global::Halibut.Exceptions.ServiceInvocationHalibutClientException
+            or global::Halibut.Exceptions.NoMatchingServiceOrMethodHalibutClientException
+            or global::Halibut.Exceptions.MethodNotFoundHalibutClientException
+            or global::Halibut.Exceptions.ServiceNotFoundHalibutClientException
+            or global::Halibut.Exceptions.AmbiguousMethodMatchHalibutClientException;
+}

src/Squid.Core/Services/DeploymentExecution/Pipeline/DeploymentCompletionHandler.cs

@@ -94,6 +94,31 @@ await genericDataProvider.ExecuteInTransactionAsync(async cancellationToken =>
         }, ct).ConfigureAwait(false);
     }
 
+    /// <summary>
+    /// A transient infrastructure failure (a Halibut RPC drop that outlived the
+    /// library's own retries, or an agent that went unreachable mid-script) pauses
+    /// the deployment rather than failing it: the task transitions to
+    /// <see cref="TaskState.Paused"/> with its checkpoint AND in-flight script
+    /// pointer preserved, so a resume re-attaches to the still-running script
+    /// instead of re-dispatching a duplicate. Like <see cref="OnTimedOutAsync"/> we
+    /// deliberately do NOT delete the checkpoint and do NOT write a
+    /// <c>DeploymentCompletion</c> record (the deployment has not completed). The
+    /// historical fail-fast behaviour remains available via the
+    /// <c>SQUID_DEPLOYMENT_TRANSIENT_RESUMABLE</c> escape hatch, which routes
+    /// transient failures back through <see cref="OnFailureAsync"/>.
+    /// </summary>
+    public async Task OnTransientPauseAsync(DeploymentTaskContext ctx, Exception ex, CancellationToken ct)
+    {
+        Log.Warning(ex, "[Deploy] Task {TaskId} hit a transient infrastructure failure; pausing for resume, checkpoint preserved", ctx.ServerTaskId);
+
+        var fromState = await ResolveCurrentActiveStateAsync(ctx.ServerTaskId, ct).ConfigureAwait(false);
+
+        await genericDataProvider.ExecuteInTransactionAsync(async cancellationToken =>
+        {
+            await serverTaskService.TransitionStateAsync(ctx.ServerTaskId, fromState, TaskState.Paused, cancellationToken).ConfigureAwait(false);
+        }, ct).ConfigureAwait(false);
+    }
+
     private async Task<string> ResolveCurrentActiveStateAsync(int serverTaskId, CancellationToken ct)
     {
         var task = await serverTaskService.GetTaskAsync(serverTaskId, ct).ConfigureAwait(false);

src/Squid.Core/Services/DeploymentExecution/Pipeline/DeploymentPipelineRunner.cs

@@ -1,5 +1,6 @@
 using Squid.Core.Services.DeploymentExecution.Exceptions;
 using Squid.Core.Services.DeploymentExecution.Lifecycle;
+using Squid.Core.Services.DeploymentExecution.Pipeline.Phases;
 using Squid.Core.Services.DeploymentExecution.Script;
 using Squid.Core.Services.Deployments.ServerTask;
 
@@ -114,6 +115,26 @@ public async Task ProcessAsync(int serverTaskId, CancellationToken ct)
         {
             await SafeCompleteAsync(ctx, () => completion.OnCancelledAsync(ctx, CancellationToken.None), new DeploymentCancelledEvent(new DeploymentEventContext()));
         }
+        catch (Exception ex) when (!timeoutCts.IsCancellationRequested && !registryCts.IsCancellationRequested && !ct.IsCancellationRequested
+            && TargetCatchClassifier.IsTransientInfraFailure(ex))
+        {
+            // A transient infra failure (Halibut RPC drop after the library's
+            // retries, an unreachable agent, or an open breaker) pauses the
+            // deployment instead of failing it: the in-flight script pointer is
+            // preserved (the strategy clears it only on a definitive observation),
+            // so a resume re-attaches to the still-running script rather than
+            // re-dispatching a duplicate. We do NOT rethrow — Paused is a clean,
+            // expected outcome. This is unconditional (no opt-out): failing fast on
+            // a transient blip would just discard already-completed progress and
+            // risk a duplicate run, which has no legitimate use case.
+            //
+            // The cancel/timeout guard preserves the established precedence: a
+            // user-cancel (registryCts/ct) or a wall-clock timeout (timeoutCts)
+            // that races a transient RPC drop must NOT be reclassified as a
+            // transient pause — cancel/timeout win, so the exception falls through
+            // to their handlers (or the generic failure path).
+            await SafeCompleteAsync(ctx, () => completion.OnTransientPauseAsync(ctx, ex, CancellationToken.None), new DeploymentPausedEvent(new DeploymentEventContext()));
+        }
         catch (Exception ex)
         {
             await SafeCompleteAsync(ctx, () => completion.OnFailureAsync(ctx, ex, CancellationToken.None), new DeploymentFailedEvent(new DeploymentEventContext { Exception = ex }));

src/Squid.Core/Services/DeploymentExecution/Pipeline/IDeploymentPipelinePhase.cs

@@ -13,4 +13,11 @@ public interface IDeploymentCompletionHandler : IScopedDependency
     Task OnCancelledAsync(DeploymentTaskContext ctx, CancellationToken ct);
     Task OnPausedAsync(DeploymentTaskContext ctx, CancellationToken ct);
     Task OnTimedOutAsync(DeploymentTaskContext ctx, Exception ex, CancellationToken ct);
+
+    // A transient infrastructure failure (Halibut RPC drop after the library's
+    // retries, or an unreachable agent) pauses the deployment for resume rather
+    // than failing it terminally — the still-running script is re-attached to on
+    // resume. Mirrors OnTimedOutAsync (Paused + checkpoint preserved); distinct
+    // method so the audit/log reason stays honest.
+    Task OnTransientPauseAsync(DeploymentTaskContext ctx, Exception ex, CancellationToken ct);
 }

src/Squid.Core/Services/DeploymentExecution/Pipeline/Phases/6_ExecuteStepsPhase.Execute.cs

@@ -1,3 +1,4 @@
+using Squid.Core.Halibut.Resilience;
 using Squid.Core.Observability;
 using Squid.Core.Persistence.Entities.Deployments;
 using Squid.Core.Services.DeploymentExecution.Exceptions;
@@ -397,6 +398,20 @@ private async Task ExecuteSingleActionAsync(PreparedAction prepared, DeploymentS
         {
             throw;
         }
+        catch (Exception ex) when (TransientFailureClassifier.IsTransient(ex))
+        {
+            // A transient infra failure (Halibut RPC drop after the library's
+            // retries, an unreachable agent, or an open breaker) must PROPAGATE so
+            // the runner pauses the deployment for resume — regardless of whether
+            // the step is required. Recording it as a per-target/terminal failure
+            // (the path below) would route to OnFailureAsync, which deletes the
+            // checkpoint AND the in-flight pointer the strategy deliberately
+            // preserved — orphaning the still-running script and defeating resume.
+            // The per-target catch (TargetCatchClassifier) leaves the target
+            // in-flight; here, for a non-required step (which has no per-target
+            // catch), we still must not swallow it.
+            throw;
+        }
         catch (Exception ex)
         {
             result.Failed = true;

src/Squid.Core/Services/DeploymentExecution/Pipeline/Phases/TargetCatchClassifier.cs

@@ -1,3 +1,5 @@
+using Squid.Core.Halibut.Resilience;
+
 namespace Squid.Core.Services.DeploymentExecution.Pipeline.Phases;
 
 /// <summary>
@@ -44,10 +46,32 @@ public static Classification Classify(System.Exception ex, bool failFastCancelle
         if (ex is System.OperationCanceledException && failFastCancelled && !parentCtCancelled)
             return new Classification(MarkFailed: false, TriggerFailFast: false);
 
+        // The transient-infra case: a Halibut RPC drop (after the library's own
+        // retries) or an unreachable agent. The script may still be running on the
+        // agent, so we do NOT mark the target terminal — a resume re-attaches to it
+        // via its preserved in-flight pointer. We also do NOT fail-fast the peers:
+        // healthy targets finish their work, then the runner pauses the deployment
+        // (Task.WhenAll waits for all targets, so this target's exception surfaces
+        // after the peers complete). Guarded by !parentCtCancelled so a transient
+        // drop racing a real cancel/timeout still terminates the deployment.
+        if (IsTransientInfraFailure(ex) && !parentCtCancelled)
+            return new Classification(MarkFailed: false, TriggerFailFast: false);
+
         // Everything else (genuine exception, OCE from user cancel, OCE
         // unrelated to our CTs): treat as failure. Cancel the failFast
         // cascade so peers stop work; the re-throw upstream surfaces the
         // actual exception.
         return new Classification(MarkFailed: true, TriggerFailFast: true);
     }
+
+    /// <summary>
+    /// Whether <paramref name="ex"/> is a transient infrastructure failure that
+    /// should pause (resumable) rather than fail the deployment. Delegates to the
+    /// single source of truth, <see cref="TransientFailureClassifier.IsTransient"/>
+    /// (which excludes the permanent Halibut protocol/invocation subtypes and
+    /// includes <c>CircuitOpenException</c>), kept here as the name the pipeline
+    /// callers already reference.
+    /// </summary>
+    public static bool IsTransientInfraFailure(System.Exception ex)
+        => TransientFailureClassifier.IsTransient(ex);
 }

src/Squid.Core/Services/DeploymentExecution/Targets/Tentacle/Transport/HalibutMachineExecutionStrategy.cs

@@ -253,15 +253,24 @@ private async Task<ScriptExecutionResult> DispatchOrReattachAsync(
         Log.Information("[Deploy] Dispatching script to agent {MachineName} with ticket {Ticket}",
             request.Machine.Name, scriptTicket);
 
-        try
-        {
-            return await _observer.ObserveAndCompleteAsync(request.Machine, scriptClient, scriptTicket, scriptTimeout, ct, request.Masker, startResponse, endpoint, request.OutputSink).ConfigureAwait(false);
-        }
-        finally
-        {
-            if (_inFlightStore != null)
-                await _inFlightStore.ClearAsync(request.ServerTaskId, slot, ct).ConfigureAwait(false);
-        }
+        var result = await _observer.ObserveAndCompleteAsync(request.Machine, scriptClient, scriptTicket, scriptTimeout, ct, request.Masker, startResponse, endpoint, request.OutputSink).ConfigureAwait(false);
+
+        // Clear the in-flight pointer ONLY on a definitive observation — i.e. the
+        // observer RETURNED (script completed, reported a non-zero exit, or hit the
+        // per-script timeout and was cancelled). On a THROW (a transient RPC drop
+        // that outlived Halibut's retries, an unreachable agent, or cancellation)
+        // the script may still be running on the agent, so we PRESERVE the pointer:
+        // the deployment pauses and a resumed run re-attaches to the still-running
+        // script instead of dispatching a duplicate.
+        await ClearInFlightAsync(request.ServerTaskId, slot).ConfigureAwait(false);
+
+        return result;
+    }
+
+    private async Task ClearInFlightAsync(int serverTaskId, DispatchSlot slot)
+    {
+        if (_inFlightStore != null)
+            await _inFlightStore.ClearAsync(serverTaskId, slot, CancellationToken.None).ConfigureAwait(false);
     }
 
     /// <summary>
@@ -295,6 +304,19 @@ private async Task<ScriptExecutionResult> TryReattachAsync(
         }
         catch (Exception ex)
         {
+            // A TRANSIENT probe failure (the agent is still unreachable on resume)
+            // must NOT clear the pointer or dispatch fresh — the recorded script may
+            // still be running, so a fresh dispatch would duplicate it. Preserve the
+            // pointer and propagate so the deployment pauses again; a later resume
+            // re-probes. Only a non-transient probe error falls back to a fresh
+            // dispatch (today's behaviour).
+            if (TransientFailureClassifier.IsTransient(ex))
+            {
+                Log.Information(ex, "[Deploy] Re-attach probe hit a transient failure for agent {MachineName} (ticket {Ticket}); preserving in-flight pointer for a later resume.",
+                    request.Machine.Name, existingTicketId);
+                throw;
+            }
+
             Log.Information(ex, "[Deploy] Re-attach probe failed for agent {MachineName} (ticket {Ticket}); dispatching fresh.",
                 request.Machine.Name, existingTicketId);
             await _inFlightStore.ClearAsync(request.ServerTaskId, slot, ct).ConfigureAwait(false);
@@ -310,14 +332,13 @@ private async Task<ScriptExecutionResult> TryReattachAsync(
         Log.Information("[Deploy] Re-attaching to in-flight script on agent {MachineName} (ticket {Ticket}, state {State}) after resume — skipping duplicate dispatch.",
             request.Machine.Name, existingTicketId, probe.State);
 
-        try
-        {
-            return await _observer.ObserveAndCompleteAsync(request.Machine, scriptClient, existingTicket, scriptTimeout, ct, request.Masker, probe, endpoint, request.OutputSink).ConfigureAwait(false);
-        }
-        finally
-        {
-            await _inFlightStore.ClearAsync(request.ServerTaskId, slot, ct).ConfigureAwait(false);
-        }
+        var result = await _observer.ObserveAndCompleteAsync(request.Machine, scriptClient, existingTicket, scriptTimeout, ct, request.Masker, probe, endpoint, request.OutputSink).ConfigureAwait(false);
+
+        // Clear only on a definitive observation; preserve on a throw so a further
+        // resume re-attaches again (same rationale as DispatchOrReattachAsync).
+        await ClearInFlightAsync(request.ServerTaskId, slot).ConfigureAwait(false);
+
+        return result;
     }
 
     /// <summary>