Implement Fallback Portfolio wiring in tasks.py

gitea_client.py

@@ -19,7 +19,6 @@ import os
 import urllib.request
 import urllib.error
 import urllib.parse
-import time
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from pathlib import Path
@@ -212,53 +211,37 @@ class GiteaClient:
 
     # -- HTTP layer ----------------------------------------------------------
 
-
     def _request(
         self,
         method: str,
         path: str,
         data: Optional[dict] = None,
         params: Optional[dict] = None,
-        retries: int = 3,
-        backoff: float = 1.5,
     ) -> Any:
-        """Make an authenticated API request with exponential backoff retries."""
+        """Make an authenticated API request. Returns parsed JSON."""
         url = f"{self.api}{path}"
         if params:
             url += "?" + urllib.parse.urlencode(params)
 
         body = json.dumps(data).encode() if data else None
-        
-        for attempt in range(retries):
-            req = urllib.request.Request(url, data=body, method=method)
-            req.add_header("Authorization", f"token {self.token}")
-            req.add_header("Content-Type", "application/json")
-            req.add_header("Accept", "application/json")
+        req = urllib.request.Request(url, data=body, method=method)
+        req.add_header("Authorization", f"token {self.token}")
+        req.add_header("Content-Type", "application/json")
+        req.add_header("Accept", "application/json")
 
+        try:
+            with urllib.request.urlopen(req, timeout=30) as resp:
+                raw = resp.read().decode()
+                if not raw:
+                    return {}
+                return json.loads(raw)
+        except urllib.error.HTTPError as e:
+            body_text = ""
             try:
-                with urllib.request.urlopen(req, timeout=30) as resp:
-                    raw = resp.read().decode()
-                    if not raw:
-                        return {}
-                    return json.loads(raw)
-            except urllib.error.HTTPError as e:
-                # Don't retry client errors (4xx) except 429
-                if 400 <= e.code < 500 and e.code != 429:
-                    body_text = ""
-                    try:
-                        body_text = e.read().decode()
-                    except Exception:
-                        pass
-                    raise GiteaError(e.code, body_text, url) from e
-                
-                if attempt == retries - 1:
-                    raise GiteaError(e.code, str(e), url) from e
-                
-                time.sleep(backoff ** attempt)
-            except (urllib.error.URLError, TimeoutError) as e:
-                if attempt == retries - 1:
-                    raise GiteaError(500, str(e), url) from e
-                time.sleep(backoff ** attempt)
+                body_text = e.read().decode()
+            except Exception:
+                pass
+            raise GiteaError(e.code, body_text, url) from e
 
     def _get(self, path: str, **params) -> Any:
         # Filter out None values

tasks.py

@@ -2126,3 +2126,27 @@ def cross_review_prs():
                 continue
 
     return {"reviews": len(results), "details": results}
+
+def get_model_for_task(task_class, agent_name=None):
+    """Implement the Fallback Portfolio doctrine (docs/fallback-portfolios.md).
+    
+    Reads fallback-portfolios.yaml and returns the primary model for the given task class.
+    If primary fails, the agent should call this again with an incremented 'attempt' (not implemented here).
+    """
+    import yaml
+    portfolio_path = Path(__file__).parent / "fallback-portfolios.yaml"
+    if not portfolio_path.exists():
+        return "gemini-1.5-flash" # Default fallback
+        
+    with open(portfolio_path, "r") as f:
+        portfolios = yaml.safe_load(f)
+        
+    # 1. Check agent-specific portfolio
+    if agent_name and agent_name in portfolios.get("agents", {}):
+        return portfolios["agents"][agent_name]["primary"]["model"]
+        
+    # 2. Check task-class portfolio
+    if task_class in portfolios.get("role_classes", {}):
+        return portfolios["role_classes"][task_class]["primary"]["model"]
+        
+    return "gemini-1.5-flash"