Per review quota

Cargo.toml

@@ -40,7 +40,7 @@ tempfile = "3.24.0"
 termcolor = "1.4.1"
 thiserror = "2.0.17"
 tiktoken-rs = "0.6.0"
-tokio = { version = "1.48.0", features = ["full"] }
+tokio = { version = "1.48.0", features = ["full", "test-util"] }
 tokio-util = { version = "0.7.17", features = ["codec", "io"] }
 toml = "0.8"
 tower-http = { version = "0.6.8", features = ["cors", "fs", "trace"] }

Settings.toml

@@ -29,6 +29,10 @@ model = "gemini-3.1-pro-preview"
 max_input_tokens = 900000
 max_interactions = 50
 temperature = 1.0
+# global_backoff = true
+# quota_backoff_secs = 60
+# transient_backoff_type = "exponential"
+# transient_flat_backoff_secs = 30
 
 # To use Claude API directly:
 # provider = "claude"

designs/ai_error_handling.md

@@ -1,118 +1,90 @@
 # AI Provider Error Handling and Rate Limiting Design
 
 ## Objective
-Currently, Sashiko's handling of AI provider errors (like 503 Service Unavailable, 529 Overloaded, and 429 Too Many Requests) is fragmented and localized. Some errors cause individual worker threads to sleep and retry locally (which can lead to thread exhaustion and timeout issues), while only specific quota errors trigger the global `QuotaManager`. 
-
-The goal of this redesign is to:
-1.  **Unify error handling:** All transient and rate-limiting errors from any LLM provider (Gemini, Claude) should trigger a global backoff/delay mechanism.
-2.  **Global Pause:** When an AI provider is overloaded or returns a transient error, *all* AI interactions across *all* workers should be delayed to allow the provider to recover and to prevent exacerbating the issue.
-3.  **Infinite Retries:** There should be no hard limit on the number of retries for transient/overloaded errors. The system should patiently wait for the provider to become available again.
-4.  **Timeout Exemption:** The time spent waiting for global delays (due to quota or transient errors) must *not* be counted towards any per-worker or per-task timeouts. Worker timeouts should only apply to actual processing and execution time.
-
-## Proposed Architecture
-
-### 1. Unified Global AI Rate Limiter / Quota Manager
-The existing `QuotaManager` in `src/ai/proxy.rs` (or potentially moved to a better location like `src/ai/mod.rs` or `src/ai/quota.rs`) will be expanded and renamed to `GlobalAiBackoffManager` (or similar).
-*   It will manage a global `blocked_until` timestamp.
-*   It will keep track of consecutive transient failures to implement global exponential backoff.
-*   It will provide a way for any worker to report *any* type of retryable AI error (Quota Exceeded, Overloaded, Transient 5xx).
-*   It will provide a `wait_for_access()` method that workers must call *before* attempting any AI request.
-*   **Crucially**, `wait_for_access()` will return the `Duration` that the task actually spent sleeping.
-*   **Global Pause:** When *any* worker receives a 503 (or other retryable error), it will report it to the manager. The manager will update the global `blocked_until` timestamp. This ensures that *all other workers* will pause and wait when they call `wait_for_access()`, preventing the system from bombarding an overloaded provider.
-
-### 2. Error Categorization & Backoff Strategy
-Errors from providers will be categorized into:
-*   **Fatal Errors (400, 401, 403 - except Gemini cache expiration, 404):** These should fail immediately. No retries.
-*   **Rate Limit / Quota (429):** Providers often include a `Retry-After` header. If present, the global delay is set to this value. If not, a default (e.g., 60s) is used.
-*   **Overloaded / Transient (500, 502, 503, 504, 529):** These will trigger a global exponential backoff.
-    *   The backoff will start small (e.g., 5s) and increase (e.g., multiply by 1.5x or 2x) on consecutive failures, up to a maximum cap (e.g., 5 minutes).
-    *   A successful request will reset the consecutive failure counter.
-
-### 3. Provider Client Changes (`src/ai/gemini.rs`, `src/ai/claude.rs`)
-*   Remove the internal `loop` and retry logic (`max_retries`) from `post_request_with_retry` (Claude) and `generate_content` / `generate_content_with_cache` (Gemini).
-*   The clients should simply execute the request once and return the result or the categorized error. Both clients should map HTTP status codes to specific Rust enum variants (e.g., `GeminiError::QuotaExceeded`, `GeminiError::TransientError`).
-
-### 4. Worker Timeout Exemption (Dynamic Deadline)
-In `src/reviewer.rs`, the AI interaction is currently wrapped in a fixed `tokio::time::timeout`:
-```rust
-let interaction_result = tokio::time::timeout(
-    std::time::Duration::from_secs(settings.review.timeout_seconds),
-    async { ... }
-).await;
+Sashiko interacts with external LLM providers (like Gemini and Claude) which can occasionally return transient errors (e.g., 503 Service Unavailable, 529 Overloaded, or connection timeouts) or rate-limiting errors (429 Too Many Requests). 
+
+To ensure robustness and high availability, Sashiko implements a unified, configurable error handling and rate limiting strategy with the following goals:
+1.  **Unified Error Handling:** All transient and rate-limiting errors from any provider trigger a consistent backoff/delay mechanism.
+2.  **Configurable Scope (Global vs. Local):** Support both global rate limiting (blocking all workers to protect API quota) and local rate limiting (per-worker backoff to maximize concurrency).
+3.  **Flexible Backoff Strategies:** Support both exponential backoff (for gradual recovery) and flat backoff (for aggressive, predictable retries).
+4.  **Infinite Retries for Transients:** Transient errors are retried indefinitely to ensure reviews eventually complete.
+5.  **Timeout Exemption:** Time spent waiting for rate limits or transient backoffs does *not* count towards review timeouts. The active deadline is dynamically extended by the duration of the sleep.
+
+## Architecture
+
+### 1. AI Rate Limiter / Quota Manager (`QuotaManager`)
+The `QuotaManager` (in `src/ai/quota.rs`) manages the backoff state:
+*   It tracks a `blocked_until` timestamp.
+*   It tracks consecutive transient failures to implement exponential backoff.
+*   It provides `wait_for_access()` which blocks the caller if the rate limit is active, returning the `Duration` spent sleeping.
+*   It supports two backoff types for transient errors:
+    *   **Exponential:** Backoff starts at 1s and doubles on consecutive failures (`1s, 2s, 4s, 8s...`) capped at 60 seconds.
+    *   **Flat:** Backoff always waits a configured flat duration (e.g., 30 seconds).
+
+### 2. Global vs. Local Backoff
+The scope of the backoff is determined by how the `QuotaManager` is shared:
+*   **Global Backoff (`global_backoff = true`):** A single `QuotaManager` is shared as a singleton (`Arc<QuotaManager>`) across all workers. If one worker hits a rate limit or transient error, it blocks *all* workers. This is useful to protect API keys from getting banned under severe quota limits.
+*   **Local Backoff (`global_backoff = false`):** Each review task instantiates its own local `QuotaManager`. If a worker handling a review hits an error, it backs off individually without affecting other active workers. This maximizes throughput and retries "as soon as possible" across different reviews.
+
+### 3. Configuration (`Settings.toml`)
+The behavior is fully configurable under the `[ai]` section:
+```toml
+[ai]
+# Scope of the backoff
+global_backoff = false # If false, workers back off independently (local backoff)
+
+# Quota (429) error handling
+quota_backoff_secs = 60 # Flat delay when hitting 429
+
+# Transient (503, timeout) error handling
+transient_backoff_type = "flat" # "exponential" or "flat"
+transient_flat_backoff_secs = 30 # Flat delay when using "flat" transient backoff
 ```
-If `wait_for_access()` sleeps for 5 minutes inside that `async` block, the timeout will fire incorrectly.
 
-*   **Solution:** We will replace `tokio::time::timeout` with a custom implementation or loop that tracks the *active* deadline. 
-*   Because `tokio::time::timeout` cannot have its deadline extended once started, we should track the deadline manually using `tokio::time::sleep` combined with `tokio::select!`.
-*   Alternatively, we can track the accumulated "sleep time" inside the async block.
-*   **Best approach for Sashiko:** The AI loop is already yielding events (reading lines from the child process). 
-*   We can create a struct `ActiveTimeout`:
+### 4. Standardized Error Categorization (`AiError`)
+To avoid fragile string-matching in the orchestration layer (`reviewer.rs`), Sashiko abstracts provider-specific errors at the API boundary.
+
+We introduce a unified `AiError` enum in `src/ai/mod.rs`:
 ```rust
-struct ActiveTimeout {
-    base_duration: Duration,
-    total_sleep_added: Duration,
-    start: Instant,
-}
-impl ActiveTimeout {
-    fn new(base: Duration) -> Self { ... }
-    fn add_sleep(&mut self, d: Duration) { self.total_sleep_added += d; }
-    fn is_expired(&self) -> bool {
-        self.start.elapsed() > (self.base_duration + self.total_sleep_added)
-    }
-    fn remaining(&self) -> Duration { ... }
+#[derive(Debug, thiserror::Error, Clone)]
+pub enum AiError {
+    #[error("Quota exceeded: retry after {0:?}")]
+    QuotaExceeded(std::time::Duration),
+    #[error("Transient error: {1}, retry after {0:?}")]
+    Transient(std::time::Duration, String),
+    #[error("Fatal AI error: {0}")]
+    Fatal(String),
 }
 ```
-*   And wrap the reading of lines in a `timeout` that is re-calculated for each line or each AI request. Or better yet, run the entire child process reading loop inside a custom future that we poll alongside a sleep future that gets its deadline updated.
 
-Let's refine the timeout approach. Instead of wrapping the whole block in `tokio::time::timeout`, we can do:
+Each provider client (Gemini, Claude, OpenAI) is responsible for mapping its raw HTTP status codes or client-specific errors into this standardized enum before returning them:
+*   **Gemini:** Maps `GeminiError::QuotaExceeded` and `GeminiError::TransientError`.
+*   **Claude:** Maps `ClaudeError::RateLimitExceeded` and `ClaudeError::OverloadedError`.
+*   **OpenAI:** Maps `OpenAiCompatError::RateLimitExceeded` and `OpenAiCompatError::TransientError`.
 
+In `reviewer.rs`, the orchestration layer downcasts the returned `anyhow::Error` to `AiError`. This allows robust, type-safe classification to trigger the appropriate backoff:
+*   `AiError::QuotaExceeded(delay)`: Triggers `report_quota_error(delay)`.
+*   `AiError::Transient(delay, msg)`: Triggers `report_transient_error()`.
+*   `AiError::Fatal(msg)`: Fails the review immediately (no retries).
+
+A fallback string-matching mechanism is preserved only for backward compatibility with custom or legacy external tools.
+
+### 5. Worker Timeout Exemption (Dynamic Deadline)
+To prevent reviews from timing out while waiting for API recovery, the review deadline is dynamically extended.
+
+In `src/reviewer.rs` (`run_review_tool`), the active timeout is managed using a mutable `deadline` (`TokioInstant`):
 ```rust
-let deadline = Instant::now() + Duration::from_secs(settings.review.timeout_seconds);
-// ... inside the loop handling messages ...
-    let resp_payload = loop {
-        let slept = quota_manager.wait_for_access().await;
-        deadline += slept; // Extend deadline by the time we slept!
-
-        // Check timeout manually before heavy work, or rely on read timeouts
-        if Instant::now() > deadline {
-             break Err(anyhow!("Review tool timed out (active time exceeded)"));
-        }
-
-        match provider.generate_content(req.clone()).await {
-            Ok(resp) => {
-                quota_manager.report_success().await;
-                break Ok(resp);
-            }
-            Err(e) => {
-                if is_retryable(&e) {
-                     quota_manager.report_error(e).await;
-                     continue; // Loop indefinitely until success or fatal error
-                }
-                break Err(e);
-            }
-        }
-    }
+let mut deadline = TokioInstant::now() + Duration::from_secs(settings.review.timeout_seconds);
 ```
-However, the `tokio::time::timeout` also guards against the *child process* hanging while producing stdout. We still need a timeout around the `lines.next_line().await`.
-
-**Updated Timeout Strategy:**
-1. We will use a mutable `deadline` variable: `let mut deadline = Instant::now() + Duration::from_secs(timeout_seconds);`
-2. We wrap the `lines.next_line().await` (and other I/O) in `tokio::time::timeout_at(deadline.into(), ...)`
-3. Whenever `quota_manager.wait_for_access().await` returns a `Duration > 0`, we do `deadline += duration`.
-
-## Execution Plan
-
-1.  **Refactor `QuotaManager`:**
-    *   Rename to `QuotaManager` (keep name to minimize churn) but add tracking for consecutive transient failures.
-    *   Update `wait_for_access` to return the `Duration` slept.
-    *   Add `report_transient_error()` to trigger global exponential backoff.
-    *   Add `report_success()` to reset transient error counter.
-2.  **Update `gemini.rs` and `claude.rs`:**
-    *   Remove local `retry_count` loops and `tokio::time::sleep`.
-    *   Return errors directly (ensure 503/529 map to `TransientError` and 429 maps to `QuotaExceeded`).
-3.  **Update Callers (`reviewer.rs`, `proxy.rs`):**
-    *   Implement infinite loop around `provider.generate_content`.
-    *   Call `wait_for_access()` and capture the sleep duration.
-    *   Extend timeouts (`deadline += slept`).
-    *   Report errors (`report_quota_error` or `report_transient_error`) or success (`report_success`).
-4.  **Implement `timeout_at` logic in `reviewer.rs`:** Replace the single block `tokio::time::timeout` with loop-level `timeout_at` to allow deadline extension.
+
+During the AI interaction loop, before each request, the worker calls `wait_for_access()` and extends the deadline by the slept duration:
+```rust
+let slept = quota_manager.wait_for_access().await;
+deadline += slept; // Extend deadline by the time we slept!
+
+if TokioInstant::now() > deadline {
+    return Err(anyhow!("Review tool timed out (active time exceeded)"));
+}
+```
+
+This ensures that the `timeout_seconds` config applies strictly to **active processing time** (e.g., waiting for child process I/O, local processing), while all time spent waiting for external AI provider availability is exempt.

sashiko.dev/base/app/sashiko-k8s.yaml

@@ -102,6 +102,17 @@ spec:
               value: "false"
             - name: SASHIKO__REVIEW__EMAIL_POLICY_PATH
               value: "/app/email_policy.toml"
+            # Enable per-worker (local) backoff isolation
+            - name: SASHIKO__AI__GLOBAL_BACKOFF
+              value: "false"
+            # Configure flat backoffs instead of exponential
+            - name: SASHIKO__AI__TRANSIENT_BACKOFF_TYPE
+              value: "flat"
+            - name: SASHIKO__AI__TRANSIENT_FLAT_BACKOFF_SECS
+              value: "30"
+            # Set flat quota retry to 65s too (optional, defaults to 60s if omitted)
+            - name: SASHIKO__AI__QUOTA_BACKOFF_SECS
+              value: "65"
             - name: LLM_API_KEY
               valueFrom:
                 secretKeyRef:

src/ai/claude.rs

@@ -574,7 +574,27 @@ impl AiProvider for ClaudeClient {
         claude_req.model = self.model.clone();
 
         // 3. Make API call
-        let response = self.post_request(&claude_req).await?;
+        let response = match self.post_request(&claude_req).await {
+            Ok(resp) => resp,
+            Err(e) => {
+                if let Some(claude_err) = e.downcast_ref::<ClaudeError>() {
+                    match claude_err {
+                        ClaudeError::RateLimitExceeded(d) => {
+                            return Err(crate::ai::AiError::QuotaExceeded(*d).into());
+                        }
+                        ClaudeError::OverloadedError(d) => {
+                            return Err(crate::ai::AiError::Transient(
+                                *d,
+                                "Claude API overloaded".to_string(),
+                            )
+                            .into());
+                        }
+                        _ => {}
+                    }
+                }
+                return Err(e);
+            }
+        };
 
         // 4. Translate response back to generic format
         translate_ai_response(&response)

src/ai/gemini.rs

@@ -756,7 +756,39 @@ fn estimate_tokens_generic(request: &AiRequest) -> usize {
 impl AiProvider for GeminiClient {
     async fn generate_content(&self, request: AiRequest) -> Result<AiResponse> {
         let gen_req = translate_ai_request(request)?;
-        let resp = GenAiClient::generate_content(self, gen_req).await?;
+        let resp = match GenAiClient::generate_content(self, gen_req).await {
+            Ok(resp) => resp,
+            Err(e) => {
+                if let Some(gemini_err) = e.downcast_ref::<GeminiError>() {
+                    match gemini_err {
+                        GeminiError::QuotaExceeded(d) => {
+                            return Err(crate::ai::AiError::QuotaExceeded(*d).into());
+                        }
+                        GeminiError::TransientError(d, msg) => {
+                            return Err(crate::ai::AiError::Transient(*d, msg.clone()).into());
+                        }
+                        GeminiError::PermissionDenied(msg) => {
+                            return Err(crate::ai::AiError::Fatal(format!(
+                                "Permission denied: {}",
+                                msg
+                            ))
+                            .into());
+                        }
+                        GeminiError::ApiError(status, msg) => {
+                            return Err(crate::ai::AiError::Fatal(format!(
+                                "API Error ({}): {}",
+                                status, msg
+                            ))
+                            .into());
+                        }
+                        GeminiError::Other(msg) => {
+                            return Err(crate::ai::AiError::Fatal(msg.clone()).into());
+                        }
+                    }
+                }
+                return Err(e);
+            }
+        };
         translate_ai_response(resp)
     }
 

src/ai/mod.rs

@@ -176,6 +176,17 @@ pub struct CacheStats {
     pub tokens_saved_prev_session: u64,
 }
 
+/// Generic errors to all AI providers
+#[derive(Debug, thiserror::Error, Clone)]
+pub enum AiError {
+    #[error("Quota exceeded: retry after {0:?}")]
+    QuotaExceeded(std::time::Duration),
+    #[error("Transient error: {1}, retry after {0:?}")]
+    Transient(std::time::Duration, String),
+    #[error("Fatal AI error: {0}")]
+    Fatal(String),
+}
+
 /// Trait defining the standard interface for all AI providers in Sashiko.
 #[async_trait]
 pub trait AiProvider: Send + Sync {

src/ai/openai.rs

@@ -482,7 +482,18 @@ impl AiProvider for OpenAiCompatClient {
         openai_req.model = self.model.clone();
 
         let resp_body = serde_json::to_value(&openai_req)?;
-        let resp = self.post_request(&resp_body).await?;
+        let resp = match self.post_request(&resp_body).await {
+            Ok(resp) => resp,
+            Err(e) => match e {
+                OpenAiCompatError::RateLimitExceeded(d) => {
+                    return Err(crate::ai::AiError::QuotaExceeded(d).into());
+                }
+                OpenAiCompatError::TransientError(d, msg) => {
+                    return Err(crate::ai::AiError::Transient(d, msg).into());
+                }
+                _ => return Err(e.into()),
+            },
+        };
         translate_ai_response(resp)
     }