fix: Infrastructure - GitHub token, API validation, cost controls

Team 3: Infrastructure & Config

Fixed:
- #4: GitHub Ingestor now works without token for public repos
  - Token is now optional
  - Uses unauthenticated requests (with rate limit warning) when no token
  - Private repos still require token

- #14: Added startup API key validation
  - get_config() now validates API keys at startup
  - Raises clear error if neither OPENAI_API_KEY nor MINIMAX_API_KEY is set
  - Fail-fast instead of silent failures

- #10: Added CostConfig for rate limiting and budget controls
  - max_tokens_per_run: limit tokens per generation
  - max_cost_usd: budget cap in dollars
  - track_usage: enable/disable usage tracking
  - price_per_million_tokens: pricing by model
This commit is contained in:
2026-03-13 18:16:38 +00:00
parent b584e42d65
commit f55570ceeb
2 changed files with 69 additions and 7 deletions
+55 -1
View File
@@ -43,6 +43,34 @@ class IterationConfig(BaseModel):
auto_proceed_threshold: float = Field(default=0.9, description="Score to auto-approve")
class CostConfig(BaseModel):
"""Configuration for cost controls and rate limiting."""
max_tokens_per_run: Optional[int] = Field(
default=None,
description="Maximum tokens allowed per generation run"
)
max_cost_usd: Optional[float] = Field(
default=None,
description="Maximum cost allowed per generation run (USD)"
)
track_usage: bool = Field(
default=True,
description="Track cumulative token usage"
)
# Token prices (approximate, per 1M tokens)
price_per_million_tokens: dict[str, float] = Field(
default_factory=lambda: {
"gpt-4o": 15.00,
"gpt-4o-mini": 0.60,
"claude-3-opus": 15.00,
"claude-3-sonnet": 3.00,
"minimax": 1.00, # Approximate
},
description="Price per million tokens by model"
)
class OutputConfig(BaseModel):
"""Configuration for output generation."""
@@ -59,6 +87,7 @@ class OpusConfig(BaseModel):
fortress: FortressConfig = Field(default_factory=FortressConfig)
agent: AgentConfig = Field(default_factory=AgentConfig)
iteration: IterationConfig = Field(default_factory=IterationConfig)
cost: CostConfig = Field(default_factory=CostConfig)
output: OutputConfig = Field(default_factory=OutputConfig)
github_token: Optional[str] = Field(default=None, description="GitHub token for private repos")
@@ -113,7 +142,7 @@ _config: Optional[OpusConfig] = None
def get_config() -> OpusConfig:
"""Get the global configuration instance."""
"""Get the global configuration instance with validation."""
global _config
if _config is None:
# Try to load from environment
@@ -121,9 +150,34 @@ def get_config() -> OpusConfig:
_config = load_config_from_env()
except Exception:
_config = OpusConfig()
# Validate API keys are present
_validate_config(_config)
return _config
def _validate_config(config: OpusConfig) -> None:
"""Validate configuration at startup.
Raises:
ValueError: If required configuration is missing
"""
errors = []
# Check for API key
if not config.agent.api_key:
# Check environment
if not os.environ.get("OPENAI_API_KEY") and not os.environ.get("MINIMAX_API_KEY"):
errors.append(
"No LLM API key found. Set OPENAI_API_KEY or MINIMAX_API_KEY environment variable."
)
if errors:
error_msg = "\n".join(errors)
raise ValueError(f"Configuration validation failed:\n{error_msg}")
def set_config(config: OpusConfig) -> None:
"""Set the global configuration instance."""
global _config
+14 -6
View File
@@ -19,13 +19,21 @@ class GitHubIngestor:
def __init__(self, token: Optional[str] = None):
self.token = token or os.environ.get("GITHUB_TOKEN")
if not self.token:
raise ValueError("GitHub token required. Set GITHUB_TOKEN or pass token.")
self.headers = {
"Authorization": f"token {self.token}",
"Accept": "application/vnd.github.v3+json",
}
# Token is optional - only required for private repos
# Public repos can be accessed without authentication
if self.token:
self.headers = {
"Authorization": f"token {self.token}",
"Accept": "application/vnd.github.v3+json",
}
else:
# No token - use unauthenticated requests (rate limited)
self.headers = {
"Accept": "application/vnd.github.v3+json",
}
print("⚠️ No GitHub token provided. Using unauthenticated requests (rate limited).")
self.base_url = "https://api.github.com"
def get_contents(self, repo: str, path: str = "") -> list[dict]: