diff --git a/.agents/skills/flow-creator/references/flow-spec.md b/.agents/skills/flow-creator/references/flow-spec.md index 62335bd49d..1ce53705fc 100644 --- a/.agents/skills/flow-creator/references/flow-spec.md +++ b/.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: ""` (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 diff --git a/internal/flow/flow.go b/internal/flow/flow.go index 6a1f0403fe..61370898c4 100644 --- a/internal/flow/flow.go +++ b/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 +} diff --git a/internal/flow/registry.go b/internal/flow/registry.go index 3531ac8e38..618c9ff9b0 100644 --- a/internal/flow/registry.go +++ b/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 diff --git a/internal/flow/service.go b/internal/flow/service.go index d63462a8da..f5090b2a50 100644 --- a/internal/flow/service.go +++ b/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 { diff --git a/internal/flow/transient_postpone_test.go b/internal/flow/transient_postpone_test.go new file mode 100644 index 0000000000..c57932096e --- /dev/null +++ b/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) + } + }) + } +}