fix: retry mid-turn 429s with backoff and native continue

.release-files

@@ -32,6 +32,8 @@ internal/agy/process_unix.go
 internal/agy/process_unix_test.go
 internal/agy/process_windows.go
 internal/agy/process_windows_test.go
+internal/agy/retry.go
+internal/agy/retry_test.go
 internal/agy/runner.go
 internal/agy/runner_test.go
 internal/agy/signal_unix.go

README.md

@@ -49,6 +49,24 @@ Prompts exceeding the configurable byte threshold (default 8KB) are written unde
 first session for a workdir starts, when the last session in that workdir closes,
 and on wrapper shutdown.
 
+### Provider 429 / quota continue
+
+A mid-turn Gemini `429` / `RESOURCE_EXHAUSTED` used to fail the ACP `session/prompt`
+immediately (or, before fail-fast, hang until `--timeout-seconds`). The wrapper now
+keeps that prompt open and retries the **native** `agy --conversation` turn:
+
+1. Parse `retryDelay` / "retry in …" / `Retry-After` when present; otherwise use
+   exponential backoff (`2s` … `60s`).
+2. Wait up to the remaining per-turn timeout (or `--quota-retry-max-wait-seconds`).
+3. Stream a short "waiting … before retry" notice so the ACP client is not silent.
+4. If the failed attempt already produced a conversation ID, send a continue cue
+   instead of replaying the user prompt.
+5. After the attempt budget is exhausted, still return ACP `-32003` and **do not**
+   dump fallback context (that would burn more quota).
+
+This is independent of the virtual-context fallback used for other native-resume
+failures.
+
 ### Response Extraction
 When `agy --print` produces no stdout (a known platform-specific issue), the
 wrapper extracts the model's response directly from agy's transcript log at
@@ -116,6 +134,8 @@ Use `go-agy-acp-wrapper --version` for executable validation without starting AC
 | `AGY_MODEL` | _(empty = catalog default)_ | Default model for new sessions (e.g. `gemini-3.6-flash-high`) |
 | `AGY_PROMPT_THRESHOLD` | `8000` | Byte threshold above which prompts are written to temp files |
 | `AGY_TIMEOUT_SECONDS` | `14400` (4 hours) | Per-turn execution timeout in seconds. Forwarded to agy as `--print-timeout` (agy's own print-mode default is 5 minutes). |
+| `AGY_QUOTA_RETRY_ATTEMPTS` | `5` | How many `agy --print` invocations to make for one ACP prompt after a 429 / quota error (1 initial + retries). |
+| `AGY_QUOTA_RETRY_MAX_WAIT_SECONDS` | `0` (remaining turn timeout) | Cap for a single backoff wait. `0` waits up to the remaining per-turn timeout (so a 3h quota reset can still resume inside a 4h turn). |
 | `AGY_SKIP_PERMISSIONS` | `true` | Whether to pass `--dangerously-skip-permissions` to `agy`; set to `false` to opt out |
 | `AGY_ACP_SKIP_ENV_NOTE` | `false` | Set to `true` to skip the once-per-session execution-environment steering note |
 
@@ -127,6 +147,8 @@ Equivalent CLI flags are available and override environment values:
 | `--model <model>` | Default model for new sessions |
 | `--prompt-threshold <bytes>` | Byte threshold above which prompts are written to workdir files |
 | `--timeout-seconds <seconds>` | Per-turn execution timeout and agy `--print-timeout` |
+| `--quota-retry-attempts <n>` | Native continue retries after 429 / quota errors |
+| `--quota-retry-max-wait-seconds <seconds>` | Cap for a single 429 backoff (`0` = remaining turn timeout) |
 | `--skip-permissions` | Force-enable `--dangerously-skip-permissions` |
 | `--no-skip-permissions` | Opt out of `--dangerously-skip-permissions` |
 | `--execution-env-note` | Force-enable the once-per-session execution-environment note |

internal/acp/agent.go

@@ -28,6 +28,7 @@ type AgyAgent struct {
 	modelCatalog *agy.ModelCatalog
 	promptWriter *agy.PromptFileWriter
 	coordinator  *agy.RepoCoordinator
+	sleep        func(context.Context, time.Duration) error
 	mu           sync.Mutex
 	workdirs     map[string]int
 	cancels      map[string]activePrompt
@@ -55,6 +56,7 @@ func NewAgyAgent(cfg *config.Config) *AgyAgent {
 		modelCatalog: agy.NewModelCatalog(cfg.AgyBinary),
 		promptWriter: agy.NewPromptFileWriter(cfg.PromptThreshold),
 		coordinator:  agy.NewRepoCoordinator(cfg.AgyConfigDir()),
+		sleep:        sleepContext,
 		workdirs:     make(map[string]int),
 		cancels:      make(map[string]activePrompt),
 		finalizers:   make(map[string]*sessionFinalizer),
@@ -173,6 +175,9 @@ func (a *AgyAgent) Prompt(ctx context.Context, params acp.PromptRequest) (acp.Pr
 		if !ok {
 			return
 		}
+		if a.conn == nil {
+			return
+		}
 		if err := a.conn.SessionUpdate(promptCtx, acp.SessionNotification{
 			SessionId: params.SessionId,
 			Update:    update,
@@ -205,7 +210,7 @@ func (a *AgyAgent) Prompt(ctx context.Context, params acp.PromptRequest) (acp.Pr
 		tail = response[len(streamedText):]
 	}
 
-	if tail != "" {
+	if tail != "" && a.conn != nil {
 		if err := a.conn.SessionUpdate(ctx, acp.SessionNotification{
 			SessionId: params.SessionId,
 			Update:    acp.UpdateAgentMessageText(tail),
@@ -218,6 +223,99 @@ func (a *AgyAgent) Prompt(ctx context.Context, params acp.PromptRequest) (acp.Pr
 }
 
 func (a *AgyAgent) executeTurn(ctx context.Context, sess *session.Context, promptText string, onEvent func(agy.StreamEvent)) (string, error) {
+	attempts := a.quotaRetryAttempts()
+	turnDeadline := time.Now().Add(time.Duration(a.cfg.TimeoutSeconds) * time.Second)
+	if deadline, ok := ctx.Deadline(); ok && deadline.Before(turnDeadline) {
+		turnDeadline = deadline
+	}
+
+	prompt := promptText
+	var native nativeAttemptState
+	var lastErr error
+	for attempt := 0; attempt < attempts; attempt++ {
+		remaining := time.Until(turnDeadline)
+		if remaining < 0 {
+			remaining = 0
+		}
+		timeout := remaining
+		configured := time.Duration(a.cfg.TimeoutSeconds) * time.Second
+		if configured > 0 && (timeout == 0 || timeout > configured) {
+			timeout = configured
+		}
+		out, err := a.executeTurnOnce(ctx, sess, prompt, timeout, onEvent, &native)
+		if err == nil {
+			return out, nil
+		}
+		lastErr = err
+		if ctx.Err() != nil {
+			return "", err
+		}
+		if !agy.IsProviderLimitError(err) || attempt+1 >= attempts {
+			return "", err
+		}
+		wait, ok := agy.NextProviderLimitWait(attempt, err, remaining, a.quotaRetryMaxWait())
+		if !ok {
+			return "", err
+		}
+		notice := fmt.Sprintf("\nProvider rate limit; waiting %s before retry %d/%d.\n", wait.Round(time.Second), attempt+2, attempts)
+		if onEvent != nil {
+			onEvent(agy.StreamEvent{Kind: agy.StreamEventText, Text: notice})
+		}
+		slog.Warn("provider rate limit, backing off before native continue",
+			"sessionId", sess.ID,
+			"wait", wait,
+			"attempt", attempt+2,
+			"attempts", attempts,
+			"error", err,
+		)
+		if sleepErr := a.sleep(ctx, wait); sleepErr != nil {
+			return "", sleepErr
+		}
+		if sess.GetConversationID() != "" {
+			prompt = providerLimitContinuePrompt
+		}
+	}
+	return "", lastErr
+}
+
+const providerLimitContinuePrompt = "Continue from where you left off. Do not restart the task."
+
+type nativeAttemptState struct {
+	previousID     string
+	startedAt      time.Time
+	snapshotDone   bool
+	snapshotFailed bool
+}
+
+func (a *AgyAgent) quotaRetryAttempts() int {
+	if a.cfg == nil || a.cfg.QuotaRetryAttempts < 1 {
+		return config.DefaultQuotaRetryAttempts
+	}
+	return a.cfg.QuotaRetryAttempts
+}
+
+func (a *AgyAgent) quotaRetryMaxWait() time.Duration {
+	if a.cfg == nil || a.cfg.QuotaRetryMaxWaitSeconds <= 0 {
+		return 0
+	}
+	return time.Duration(a.cfg.QuotaRetryMaxWaitSeconds) * time.Second
+}
+
+func sleepContext(ctx context.Context, d time.Duration) error {
+	if d <= 0 {
+		return ctx.Err()
+	}
+	timer := time.NewTimer(d)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return nil
+	}
+}
+
+func (a *AgyAgent) executeTurnOnce(ctx context.Context, sess *session.Context, promptText string, timeout time.Duration, onEvent func(agy.StreamEvent), native *nativeAttemptState) (string, error) {
 	mode := sess.GetMode()
 	convID := sess.GetConversationID()
 	turnCount := sess.GetTurnCount()
@@ -230,15 +328,15 @@ func (a *AgyAgent) executeTurn(ctx context.Context, sess *session.Context, promp
 	opts := agy.ExecuteOpts{
 		Cwd:       sess.Cwd,
 		Model:     nativeModel,
-		Timeout:   time.Duration(a.cfg.TimeoutSeconds) * time.Second,
+		Timeout:   timeout,
 		SkipPerms: a.cfg.SkipPerms,
 	}
 
 	switch {
 	case mode == session.ModeFallbackContext:
 		return a.executeFallbackTurn(ctx, sess, opts, promptText, onEvent)
 
-	case convID != "" && turnCount > 1:
+	case convID != "":
 		opts.ConversationID = convID
 		opts.Prompt = promptText
 		if a.promptWriter.NeedsFile(promptText) {
@@ -251,6 +349,7 @@ func (a *AgyAgent) executeTurn(ctx context.Context, sess *session.Context, promp
 		}
 
 		resp, err := a.runner.ExecuteStream(ctx, opts, onEvent)
+		captureConversationID(sess, resp)
 		out, err := turnOutput(resp, err, opts)
 		if err != nil {
 			if !agy.ShouldFallback(err) {
@@ -263,13 +362,22 @@ func (a *AgyAgent) executeTurn(ctx context.Context, sess *session.Context, promp
 		return out, nil
 
 	default:
-		previousID, err := a.discoverer.SnapshotConversationID(ctx, sess.Cwd)
-		if err != nil {
-			slog.Warn("conversation cache snapshot failed, using fallback context", "error", err, "sessionId", sess.ID)
+		if native.snapshotFailed {
 			sess.SwitchToFallback()
 			return a.executeFallbackTurn(ctx, sess, opts, promptText, onEvent)
 		}
-		startedAt := time.Now()
+		if !native.snapshotDone {
+			previousID, snapErr := a.discoverer.SnapshotConversationID(ctx, sess.Cwd)
+			native.snapshotDone = true
+			if snapErr != nil {
+				slog.Warn("conversation cache snapshot failed, using fallback context", "error", snapErr, "sessionId", sess.ID)
+				native.snapshotFailed = true
+				sess.SwitchToFallback()
+				return a.executeFallbackTurn(ctx, sess, opts, promptText, onEvent)
+			}
+			native.previousID = previousID
+			native.startedAt = time.Now()
+		}
 		opts.Prompt = promptText
 		if a.promptWriter.NeedsFile(promptText) {
 			path, err := a.promptWriter.WritePromptFile(sess.Cwd, sess.ID, turnCount, promptText)
@@ -281,16 +389,17 @@ func (a *AgyAgent) executeTurn(ctx context.Context, sess *session.Context, promp
 		}
 
 		resp, err := a.runner.ExecuteStream(ctx, opts, onEvent)
+		captureConversationID(sess, resp)
 		out, err := turnOutput(resp, err, opts)
 		if err != nil {
+			if sess.GetConversationID() == "" && !native.startedAt.IsZero() {
+				_ = a.discoverAndSetConversationID(ctx, sess, native.previousID, native.startedAt)
+			}
 			return "", err
 		}
 
-		if convID == "" {
-			if resp.ConversationID != "" {
-				sess.SetConversationID(resp.ConversationID)
-				slog.Info("conversation ID received from agy stream", "conversationId", resp.ConversationID, "sessionId", sess.ID)
-			} else if !a.discoverAndSetConversationID(ctx, sess, previousID, startedAt) {
+		if sess.GetConversationID() == "" {
+			if !a.discoverAndSetConversationID(ctx, sess, native.previousID, native.startedAt) {
 				sess.SwitchToFallback()
 			}
 		}
@@ -299,6 +408,17 @@ func (a *AgyAgent) executeTurn(ctx context.Context, sess *session.Context, promp
 	}
 }
 
+func captureConversationID(sess *session.Context, resp *agy.Response) {
+	if resp == nil || strings.TrimSpace(resp.ConversationID) == "" {
+		return
+	}
+	if sess.GetConversationID() == resp.ConversationID {
+		return
+	}
+	sess.SetConversationID(resp.ConversationID)
+	slog.Info("conversation ID received from agy stream", "conversationId", resp.ConversationID, "sessionId", sess.ID)
+}
+
 func (a *AgyAgent) executeFallbackTurn(ctx context.Context, sess *session.Context, opts agy.ExecuteOpts, promptText string, onEvent func(agy.StreamEvent)) (string, error) {
 	transcript := sess.GetTranscript()
 	turnCount := sess.GetTurnCount()