feat(flow): postpone-and-auto-resume steps on transient provider errors

.agents/skills/flow-creator/references/flow-spec.md

@@ -37,6 +37,7 @@ flow:               # flow specification (required)
   fallback: object       # retry and error routing (optional)
   maxTurns: int          # per-step override for agent's maxTurns. 0 (unset) inherits from agent. (optional)
   maxIterations: int     # cap on in-process self-loop iterations. 0 (unset) is unbounded — only flow timeout applies. (optional)
+  resume_after: string   # Go duration (e.g. "15m", "1h"). Opts a postponed step into timed auto-resume, and parks (not fails) the step on a transient provider error. See `resume_after` below. (optional)
   interactive: bool      # if true, the step is human-in-the-loop via the chat bridge — see Interactive Steps. (optional)
   interaction:           # required when interactive: true; ignored otherwise
     target: string       # ${args.NAME} expression resolving to a PeerRef or []PeerRef
@@ -216,6 +217,12 @@ Cap unconditional self-loops with `maxIterations` — if the agent's termination
 
 Cap semantics: counts **in-process** iterations only — a `postpone: true` self-route does not bump the counter, so a postpone loop is not bounded by `maxIterations` (bound those externally). The cap is a **post-step check** — with `maxIterations: N`, exactly N agent calls happen before the step fails.
 
+### `resume_after` — timed auto-resume, incl. on transient provider errors
+
+A step-level `resume_after: "<duration>"` (e.g. `"15m"`, `"1h"`) opts a **postponed** step into timed auto-resume: the orchestrator's postpone sweep re-enters the parked step after the delay (clamped to `POSTPONE_RESUME_MAX_TIMEOUT`; its resume-chain loop-breaker bounds a persistently-stuck resume), instead of waiting for a webhook / manual re-trigger.
+
+Declaring `resume_after` ALSO opts the step into **postpone-and-auto-resume on a transient provider error**: when a step's agent run fails with a rate limit (HTTP 429), an overloaded/5xx upstream, or an HTTP/2 stream reset the provider already exhausted its in-call retries on, the runtime parks the step as `postponed` (not `failed`, and no `fallback` routing) so it retries later when the endpoint recovers. Steps **without** `resume_after` keep such failures terminal (they route to `fallback.to`). Do NOT set `resume_after` on must-run-now steps (e.g. a salvage/safety-net step that pushes local WIP before the workspace is torn down) — a resume runs in a fresh workspace where that local state is gone.
+
 ```yaml
 - id: build-level
   agent: coder

internal/flow/flow.go

@@ -3,6 +3,7 @@ package flow
 import (
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 )
 
@@ -96,6 +97,16 @@ type Step struct {
 	// stays inside the model's cached prompt. See flow-creator SKILL
 	// "Per-step context compaction".
 	Compact *StepCompact `yaml:"compact,omitempty"`
+	// ResumeAfter opts this step into timed auto-resume when it postpones:
+	// the orchestrator's postpone sweep re-enters the parked step after the
+	// given delay (clamped to POSTPONE_RESUME_MAX_TIMEOUT). It is consumed
+	// orchestrator-side for the wake schedule; the flow runner reads it only
+	// to decide whether a step MAY park rather than fail on a transient
+	// provider error (rate-limit / stream reset) — see
+	// isTransientProviderError + postponeStepForTransientError in service.go.
+	// A Go duration string (`10m`, `1h`). Absent (nil) keeps the step's
+	// failures terminal.
+	ResumeAfter *string `yaml:"resume_after,omitempty"`
 }
 
 // StepCompact configures per-step overrides to the auto-compaction
@@ -187,3 +198,27 @@ func (s Step) TimeoutDuration() (time.Duration, error) {
 	}
 	return d, nil
 }
+
+// ResumeAfterDuration parses the step's resume_after opt-in. Returns (0, nil)
+// when unset or blank (a bare opt-in the orchestrator resolves to its default).
+// A non-blank value MUST be a positive Go duration string; anything else is a
+// load-time error so a typo (e.g. "15minutes") surfaces immediately instead of
+// silently degrading to the orchestrator's default wake — and instead of the
+// flow runner parking a step on a value the orchestrator can't schedule from.
+func (s Step) ResumeAfterDuration() (time.Duration, error) {
+	if s.ResumeAfter == nil {
+		return 0, nil
+	}
+	raw := strings.TrimSpace(*s.ResumeAfter)
+	if raw == "" {
+		return 0, nil
+	}
+	d, err := time.ParseDuration(raw)
+	if err != nil {
+		return 0, fmt.Errorf("step %q: invalid resume_after %q: %w", s.ID, raw, err)
+	}
+	if d <= 0 {
+		return 0, fmt.Errorf("step %q: resume_after must be positive, got %v", s.ID, d)
+	}
+	return d, nil
+}

internal/flow/registry.go

@@ -450,6 +450,14 @@ func validateFlow(f *Flow) error {
 		if _, err := step.TimeoutDuration(); err != nil {
 			return fmt.Errorf("%w: %v", ErrInvalidYAML, err)
 		}
+		// resume_after, when set non-blank, must parse as a positive Go
+		// duration. The orchestrator falls back to its default on an
+		// unparseable value, so validating here is what turns a typo into a
+		// fast load-time failure instead of a silently-wrong wake timer on a
+		// parked step.
+		if _, err := step.ResumeAfterDuration(); err != nil {
+			return fmt.Errorf("%w: %v", ErrInvalidYAML, err)
+		}
 	}
 
 	// Validate rule and fallback references

internal/flow/service.go

@@ -688,6 +688,16 @@ func (s *service) runStep(
 doneRetry:
 
 	if lastErr != nil {
+		// A transient provider error (rate limit / HTTP-2 stream reset) that
+		// survived the in-call and fallback.retry budgets parks an opted-in
+		// step (resume_after) for timed auto-resume instead of failing the
+		// flow — the endpoint is expected to recover. Past the age cap it
+		// declines to park and we fall through to terminal failure. See
+		// postponeStepForTransientError.
+		if stepPostponesOnProviderError(step) && isTransientProviderError(lastErr) &&
+			s.postponeStepForTransientError(ctx, step, sessionID, rootSessionID, f.ID, args, iteration, lastErr, flowStates) {
+			return
+		}
 		// When the parent ctx is cancelled (graceful shutdown, ctx-cancelled
 		// retry-delay path above) the failure-state UPDATE would also fail
 		// the SQL call immediately, leaving the flow_state row stuck on
@@ -846,6 +856,169 @@ doneRetry:
 	}
 }
 
+// transientProviderErrorSignatures are lowercased substrings of surfaced
+// error messages that indicate a transient LLM-provider / gateway failure the
+// provider already exhausted its in-call retry budget on — a rate limit or an
+// upstream blip that is expected to clear on its own. The HTTP/2 RST_STREAM
+// case (peer-initiated reset) is matched separately in isTransientProviderError
+// because it needs two co-occurring markers.
+var transientProviderErrorSignatures = []string{
+	"http 429",                       // provider retry budget exhausted on a 429
+	"maximum retry attempts reached", // retry-exhaustion wrapper
+	"rate limit",
+	"rate_limit",
+	"too many requests",
+	"overloaded",                  // Anthropic 529
+	"service unavailable",         // 503
+	"serviceunavailableexception", // Bedrock 503
+	"throttlingexception",         // Bedrock 429
+	"modeltimeoutexception",       // Bedrock 408
+	"internalserverexception",     // Bedrock 500
+}
+
+// isTransientProviderError reports whether err is a transient LLM-provider /
+// gateway failure worth parking the whole step for (postpone-and-auto-resume)
+// rather than failing the flow. Matching is on the surfaced message: the typed
+// provider error is wrapped by the agent runtime before it reaches the flow
+// layer, so string signatures are the reliable surface. Mirrors the provider's
+// own transient/RST classifiers (internal/llm/provider) without importing them.
+func isTransientProviderError(err error) bool {
+	if err == nil {
+		return false
+	}
+	msg := strings.ToLower(err.Error())
+	// Peer-initiated HTTP/2 RST_STREAM whose per-call retry budget was
+	// exhausted (e.g. "stream error: stream ID 147; INTERNAL_ERROR; received
+	// from peer"). Require both markers so we don't match unrelated errors —
+	// same shape as provider.isRetryableRSTStreamError.
+	if strings.Contains(msg, "stream error") && strings.Contains(msg, "received from peer") {
+		return true
+	}
+	for _, sig := range transientProviderErrorSignatures {
+		if strings.Contains(msg, sig) {
+			return true
+		}
+	}
+	return false
+}
+
+// stepPostponesOnProviderError reports whether the step opted into
+// postpone-and-auto-resume on a transient provider error, which it does by
+// declaring resume_after (the same field that drives the orchestrator's timed
+// resume). Without it, transient failures stay terminal.
+func stepPostponesOnProviderError(step Step) bool {
+	return step.ResumeAfter != nil && strings.TrimSpace(*step.ResumeAfter) != ""
+}
+
+// maxTransientPostponeAge bounds how long a step may keep parking-and-resuming
+// on transient provider errors before it is failed terminally. The orchestrator
+// has a resume-chain loop-breaker, but it only fires for TeamCity build-await
+// chains (match_key prefix), NOT for a work-item job that keeps transient-
+// postponing — so this is the backstop that keeps a persistently-throttled
+// endpoint from parking a step forever. Measured against the step's flow_states
+// row created_at, which is stable across resume (the row is UPDATEd, not
+// recreated), so it survives the park→resume cycle without a per-resume counter.
+const maxTransientPostponeAge = 2 * time.Hour
+
+// postponeStepForTransientError parks a step that failed with a transient
+// provider error instead of failing it: it persists the flow_state row as
+// `postponed` and emits the postponed state, so the orchestrator's postpone
+// sweep re-enters the step after its resume_after. No AgentEvent error is
+// emitted and no fallback is routed — this is a pause, not a failure. Returns
+// false WITHOUT parking when the step has been retrying past
+// maxTransientPostponeAge, so the caller fails it terminally instead.
+func (s *service) postponeStepForTransientError(
+	ctx context.Context,
+	step Step,
+	sessionID string,
+	rootSessionID string,
+	flowID string,
+	args map[string]any,
+	iteration int,
+	cause error,
+	flowStates chan<- *FlowState,
+) bool {
+	if iteration < 1 {
+		iteration = 1
+	}
+
+	// A cancelled parent ctx (graceful shutdown) would fail the SQL write
+	// immediately and strand the row on `running`; persist with a fresh
+	// deadline so the postponed status lands regardless. Mirrors the failed-
+	// state write path.
+	writeCtx := ctx
+	if ctx.Err() != nil {
+		var cancelWrite context.CancelFunc
+		writeCtx, cancelWrite = context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancelWrite()
+	}
+
+	existingFS, getErr := s.querier.GetFlowState(writeCtx, sessionID)
+	// Age backstop for the park→resume→park chain (see maxTransientPostponeAge).
+	if getErr == nil && existingFS.CreatedAt > 0 {
+		if age := time.Since(time.Unix(existingFS.CreatedAt, 0)); age > maxTransientPostponeAge {
+			logging.Warn("Transient-postpone age cap exceeded; failing step terminally instead of re-parking",
+				"step", step.ID, "age", age.Round(time.Second), "cap", maxTransientPostponeAge, "error", cause)
+			return false
+		}
+	}
+
+	logging.Warn("Flow step hit a transient provider error; postponing for timed auto-resume",
+		"step", step.ID, "resume_after", *step.ResumeAfter, "error", cause)
+
+	argsJSON, _ := json.Marshal(args)
+	var updatedAt int64
+	// Update the entry-time `running` row to `postponed`. handleStepError (the
+	// other caller) can fire from the pre-persist setup paths where that row
+	// doesn't exist yet, so create it if it's missing rather than silently
+	// no-op'ing the UPDATE. Mirrors the entry-time Get-or-Create write.
+	if getErr == nil {
+		if state, updateErr := s.querier.UpdateFlowState(writeCtx, db.UpdateFlowStateParams{
+			Status:         string(FlowStatusPostponed),
+			Args:           sql.NullString{String: string(argsJSON), Valid: true},
+			Output:         sql.NullString{},
+			IsStructOutput: false,
+			Iteration:      int64(iteration),
+			SessionID:      sessionID,
+		}); updateErr != nil {
+			logging.Warn("Failed to persist step postpone state", "session_id", sessionID, "error", updateErr)
+			updatedAt = time.Now().Unix()
+		} else {
+			updatedAt = state.UpdatedAt
+		}
+	} else {
+		if state, createErr := s.querier.CreateFlowState(writeCtx, db.CreateFlowStateParams{
+			SessionID:      sessionID,
+			RootSessionID:  rootSessionID,
+			FlowID:         flowID,
+			StepID:         step.ID,
+			Status:         string(FlowStatusPostponed),
+			Args:           sql.NullString{String: string(argsJSON), Valid: true},
+			IsStructOutput: false,
+			Iteration:      int64(iteration),
+		}); createErr != nil {
+			logging.Warn("Failed to persist step postpone state", "session_id", sessionID, "error", createErr)
+			updatedAt = time.Now().Unix()
+		} else {
+			updatedAt = state.CreatedAt
+		}
+	}
+
+	postponedState := &FlowState{
+		SessionID:     sessionID,
+		RootSessionID: rootSessionID,
+		FlowID:        flowID,
+		StepID:        step.ID,
+		Status:        FlowStatusPostponed,
+		Args:          args,
+		Iteration:     iteration,
+		UpdatedAt:     updatedAt,
+	}
+	flowStates <- postponedState
+	s.Publish(pubsub.UpdatedEvent, *postponedState)
+	return true
+}
+
 func (s *service) handleStepError(
 	ctx context.Context,
 	step Step,
@@ -861,6 +1034,14 @@ func (s *service) handleStepError(
 	nextSteps chan<- stepWork,
 	f *Flow,
 ) {
+	// A transient provider error (rate limit / stream reset) on an opted-in
+	// step parks it for timed auto-resume rather than failing the flow — unless
+	// it has been retrying past the age cap, in which case fail terminally.
+	if stepPostponesOnProviderError(step) && isTransientProviderError(err) &&
+		s.postponeStepForTransientError(ctx, step, sessionID, rootSessionID, flowID, args, iteration, err, flowStates) {
+		return
+	}
+
 	logging.Error("Flow step failed", "step", step.ID, "error", err)
 
 	if iteration < 1 {

internal/flow/transient_postpone_test.go

@@ -0,0 +1,82 @@
+package flow
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+)
+
+func TestIsTransientProviderError(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"nil", nil, false},
+		// The two errors observed on real MICRO-1014 runs:
+		{"429 retry exhausted", errors.New("maximum retry attempts reached for HTTP 429: 8 retries"), true},
+		{"http2 RST INTERNAL_ERROR", errors.New("stream error: stream ID 147; INTERNAL_ERROR; received from peer"), true},
+		// Other transient shapes.
+		{"anthropic overloaded", errors.New("Overloaded"), true},
+		{"bedrock throttling", errors.New("received exception ThrottlingException: rate exceeded"), true},
+		{"bedrock 503", errors.New("ServiceUnavailableException: Bedrock is unable to process"), true},
+		{"wrapped 429", fmt.Errorf("step %q failed: %w", "implement", errors.New("maximum retry attempts reached for HTTP 429: 8 retries")), true},
+		// Must NOT match: a local stream error not from the peer, or unrelated failures.
+		{"local stream error, not peer", errors.New("stream error: stream ID 5; CANCEL; sent by client"), false},
+		{"plain build failure", errors.New("go build failed: undefined: Foo"), false},
+		{"generic error", errors.New("boom"), false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isTransientProviderError(tt.err); got != tt.want {
+				t.Fatalf("isTransientProviderError(%v) = %v, want %v", tt.err, got, tt.want)
+			}
+		})
+	}
+}
+
+func TestStepResumeAfterDuration(t *testing.T) {
+	dur := func(s string) *string { return &s }
+	tests := []struct {
+		name    string
+		step    Step
+		wantErr bool
+	}{
+		{"unset", Step{ID: "a"}, false},
+		{"blank (bare opt-in)", Step{ID: "a", ResumeAfter: dur("  ")}, false},
+		{"valid 15m", Step{ID: "a", ResumeAfter: dur("15m")}, false},
+		{"valid 1h30m", Step{ID: "a", ResumeAfter: dur("1h30m")}, false},
+		{"typo 15minutes", Step{ID: "a", ResumeAfter: dur("15minutes")}, true},
+		{"zero", Step{ID: "a", ResumeAfter: dur("0s")}, true},
+		{"negative", Step{ID: "a", ResumeAfter: dur("-5m")}, true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := tt.step.ResumeAfterDuration()
+			if (err != nil) != tt.wantErr {
+				t.Fatalf("ResumeAfterDuration() err = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestStepPostponesOnProviderError(t *testing.T) {
+	dur := func(s string) *string { return &s }
+	tests := []struct {
+		name string
+		step Step
+		want bool
+	}{
+		{"no resume_after", Step{ID: "resolve-team"}, false},
+		{"resume_after set", Step{ID: "implement", ResumeAfter: dur("30m")}, true},
+		{"resume_after blank", Step{ID: "x", ResumeAfter: dur("   ")}, false},
+		{"resume_after empty string", Step{ID: "x", ResumeAfter: dur("")}, false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := stepPostponesOnProviderError(tt.step); got != tt.want {
+				t.Fatalf("stepPostponesOnProviderError(%+v) = %v, want %v", tt.step, got, tt.want)
+			}
+		})
+	}
+}