From 956a818a06eaff60ddf0bec8c422d27ad2f8dbe6 Mon Sep 17 00:00:00 2001 From: Ravil Giniyatullin Date: Fri, 31 Jul 2026 11:07:09 +0400 Subject: [PATCH 1/4] feat(flow): postpone-and-auto-resume steps on transient provider errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A step whose agent run fails with a transient LLM-provider/gateway error — rate limit (HTTP 429), overloaded/5xx upstream, or an HTTP/2 stream reset the provider already exhausted its in-call retries on — currently fails the whole flow (routing to fallback.to). These are self-healing capacity blips: three consecutive MICRO-1014 runs died this way, e.g. failed to process events: maximum retry attempts reached for HTTP 429: 8 retries failed to process events: stream error: stream ID 147; INTERNAL_ERROR; received from peer Park such a step as `postponed` (not `failed`) when it opts in via `resume_after`, so the orchestrator's existing postpone sweep re-enters it after the delay, when the endpoint has recovered. Its resume-chain loop-breaker already bounds a persistently-degraded upstream, and steps without `resume_after` keep their current terminal-failure behaviour. - flow.go: add Step.ResumeAfter (parsed here now; the orchestrator already reads the same `resume_after` key to schedule the timed resume — no orchestrator change needed, both parsers are lenient on the other's keys). - service.go: isTransientProviderError classifier + postponeStepForTransientError; guard both step-failure paths (the retry-loop exhaustion path and handleStepError). - No AgentEvent error is emitted and no fallback is routed — this is a pause, not a failure, so the run ends terminal-postponed and auto-resumes. Tests: internal/flow/transient_postpone_test.go covers the classifier (incl. the two observed error strings, wrapped errors, and non-transient negatives) and the resume_after opt-in gate. Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/flow/flow.go | 10 ++ internal/flow/service.go | 134 +++++++++++++++++++++++ internal/flow/transient_postpone_test.go | 57 ++++++++++ 3 files changed, 201 insertions(+) create mode 100644 internal/flow/transient_postpone_test.go diff --git a/internal/flow/flow.go b/internal/flow/flow.go index 6a1f0403fe..3041b300d1 100644 --- a/internal/flow/flow.go +++ b/internal/flow/flow.go @@ -96,6 +96,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 diff --git a/internal/flow/service.go b/internal/flow/service.go index d63462a8da..871ccff2d6 100644 --- a/internal/flow/service.go +++ b/internal/flow/service.go @@ -688,6 +688,15 @@ 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. 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 +855,124 @@ 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) != "" +} + +// 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. The orchestrator's postpone sweep +// then re-enters the step after the step's resume_after (its resume-chain +// loop-breaker bounds a persistently-degraded upstream). No AgentEvent error is +// emitted and no fallback is routed — this is a pause, not a failure. +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, +) { + if iteration < 1 { + iteration = 1 + } + logging.Warn("Flow step hit a transient provider error; postponing for timed auto-resume", + "step", step.ID, "resume_after", *step.ResumeAfter, "error", cause) + + // 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() + } + + argsJSON, _ := json.Marshal(args) + var updatedAt int64 + 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 + } + + 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) +} + func (s *service) handleStepError( ctx context.Context, step Step, @@ -861,6 +988,13 @@ 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. + 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..2bf37615f6 --- /dev/null +++ b/internal/flow/transient_postpone_test.go @@ -0,0 +1,57 @@ +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 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) + } + }) + } +} From a0b68418c847831c7eea56e9ce2afded7e706bc0 Mon Sep 17 00:00:00 2001 From: Artem Obukhov Date: Fri, 31 Jul 2026 11:30:49 +0400 Subject: [PATCH 2/4] docs(flow): document resume_after in flow-creator skill Adds the `resume_after` step field to the flow-spec reference: timed auto-resume of a postponed step, and the opt-in it grants for postpone-and-auto-resume on a transient provider error (rate limit / overloaded / HTTP-2 stream reset). Documents the fallback-vs-postpone distinction and the must-run-now caveat. Doc counterpart to the runtime behaviour added in this branch (Step.ResumeAfter + isTransientProviderError/postponeStepForTransientError in internal/flow). Co-Authored-By: Claude Opus 4.8 (1M context) --- .agents/skills/flow-creator/references/flow-spec.md | 7 +++++++ 1 file changed, 7 insertions(+) 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 From e7c1996c488dbd342dc7d8d252e63426d4c53220 Mon Sep 17 00:00:00 2001 From: Ravil Giniyatullin Date: Fri, 31 Jul 2026 16:18:27 +0400 Subject: [PATCH 3/4] fix(flow): validate resume_after at load; make postpone row-safe (review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review notes on #20: - (note 2) Add Step.ResumeAfterDuration() and validate it in validateFlow, so a malformed resume_after (e.g. "15minutes") fails fast at flow-load instead of silently degrading to the orchestrator's default wake — mirrors the existing TimeoutDuration() load-time check. Blank/unset stays valid (bare opt-in). - (note 4) postponeStepForTransientError now Get-or-Creates the flow_states row instead of a blind UpdateFlowState. handleStepError can fire from the pre-persist setup paths where the entry-time `running` row doesn't exist yet; the blind UPDATE would silently no-op there. Not reachable today (those errors carry no provider signatures) but makes the helper correct regardless of call site. Tests: add TestStepResumeAfterDuration (valid/blank/typo/zero/negative). Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/flow/flow.go | 25 ++++++++++++++ internal/flow/registry.go | 8 +++++ internal/flow/service.go | 44 ++++++++++++++++++------ internal/flow/transient_postpone_test.go | 25 ++++++++++++++ 4 files changed, 91 insertions(+), 11 deletions(-) diff --git a/internal/flow/flow.go b/internal/flow/flow.go index 3041b300d1..61370898c4 100644 --- a/internal/flow/flow.go +++ b/internal/flow/flow.go @@ -3,6 +3,7 @@ package flow import ( "errors" "fmt" + "strings" "time" ) @@ -197,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 871ccff2d6..ddafafc77e 100644 --- a/internal/flow/service.go +++ b/internal/flow/service.go @@ -945,18 +945,40 @@ func (s *service) postponeStepForTransientError( argsJSON, _ := json.Marshal(args) var updatedAt int64 - 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() + // 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 := s.querier.GetFlowState(writeCtx, sessionID); 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 { - updatedAt = state.UpdatedAt + 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{ diff --git a/internal/flow/transient_postpone_test.go b/internal/flow/transient_postpone_test.go index 2bf37615f6..c57932096e 100644 --- a/internal/flow/transient_postpone_test.go +++ b/internal/flow/transient_postpone_test.go @@ -35,6 +35,31 @@ func TestIsTransientProviderError(t *testing.T) { } } +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 { From 7aaca79c369547fae11d94ec2190e80a7ec6727a Mon Sep 17 00:00:00 2001 From: Ravil Giniyatullin Date: Fri, 31 Jul 2026 16:23:44 +0400 Subject: [PATCH 4/4] fix(flow): bound the transient-postpone chain with an age backstop (review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review note 1: the PR claimed the orchestrator's resume-chain loop-breaker bounds a persistently-throttled endpoint. It does not — breakResumeLoop returns early unless the job's match_key has the TeamCity build-await prefix (postpone_sweep.go), so a work-item job that keeps transient-postponing was unbounded. Add an opencode-side backstop: postponeStepForTransientError now declines to park (returns false → caller fails the step terminally) once the step's flow_states row is older than maxTransientPostponeAge (2h). created_at is set on insert and preserved across the UPDATE-on-resume, so it measures total park→resume wall-clock without needing a per-resume counter (args are replaced by caller args on resume; iteration is semantically overloaded — neither can carry one). Both call sites now gate the early return on the returned bool. Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/flow/service.go | 53 +++++++++++++++++++++++++++++----------- 1 file changed, 39 insertions(+), 14 deletions(-) diff --git a/internal/flow/service.go b/internal/flow/service.go index ddafafc77e..f5090b2a50 100644 --- a/internal/flow/service.go +++ b/internal/flow/service.go @@ -691,10 +691,11 @@ doneRetry: // 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. See + // 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) + 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 @@ -909,12 +910,23 @@ 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. The orchestrator's postpone sweep -// then re-enters the step after the step's resume_after (its resume-chain -// loop-breaker bounds a persistently-degraded upstream). No AgentEvent error is -// emitted and no fallback is routed — this is a pause, not a failure. +// `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, @@ -925,12 +937,10 @@ func (s *service) postponeStepForTransientError( iteration int, cause error, flowStates chan<- *FlowState, -) { +) bool { if iteration < 1 { iteration = 1 } - logging.Warn("Flow step hit a transient provider error; postponing for timed auto-resume", - "step", step.ID, "resume_after", *step.ResumeAfter, "error", cause) // A cancelled parent ctx (graceful shutdown) would fail the SQL write // immediately and strand the row on `running`; persist with a fresh @@ -943,13 +953,26 @@ func (s *service) postponeStepForTransientError( 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 := s.querier.GetFlowState(writeCtx, sessionID); getErr == nil { + if getErr == nil { if state, updateErr := s.querier.UpdateFlowState(writeCtx, db.UpdateFlowStateParams{ Status: string(FlowStatusPostponed), Args: sql.NullString{String: string(argsJSON), Valid: true}, @@ -993,6 +1016,7 @@ func (s *service) postponeStepForTransientError( } flowStates <- postponedState s.Publish(pubsub.UpdatedEvent, *postponedState) + return true } func (s *service) handleStepError( @@ -1011,9 +1035,10 @@ func (s *service) handleStepError( 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. - if stepPostponesOnProviderError(step) && isTransientProviderError(err) { - s.postponeStepForTransientError(ctx, step, sessionID, rootSessionID, flowID, args, iteration, err, flowStates) + // 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 }