Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions env.example
Original file line number Diff line number Diff line change
@@ -1,8 +1,15 @@
# Note: If you have these environment variables set in your own environment, the ones in your environment will take precedence over any you put here.

# GitHub Configuration
# Option 1: Single token (simplest)
GITHUB_TOKEN=your_github_token_here

# Option 2: Multiple tokens (for working across multiple projects with fine-grained PATs)
# Comma-separated list of tokens. When --post is used, each token is tested for write access
# and the first one that works is used. This is useful if you have separate fine-grained PATs
# for different repositories. Takes precedence over GITHUB_TOKEN for write operations.
# REVIEW_ROADMAP_GITHUB_TOKENS=ghp_token1,ghp_token2,ghp_token3

# Application Settings

REVIEW_ROADMAP_LOG_LEVEL=INFO
Expand Down
61 changes: 59 additions & 2 deletions review_roadmap/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,8 @@

import os
from pathlib import Path
from typing import Optional
from typing import List, Optional
from pydantic import model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict


Expand All @@ -21,6 +22,9 @@ class Settings(BaseSettings):

Attributes:
GITHUB_TOKEN: GitHub API token for fetching PR data.
REVIEW_ROADMAP_GITHUB_TOKENS: Comma-separated list of GitHub tokens.
When set, these are tried in order during write access checks.
Takes precedence over GITHUB_TOKEN for write operations.
REVIEW_ROADMAP_LLM_PROVIDER: LLM provider to use. Options:
'anthropic', 'anthropic-vertex', 'openai', 'google'.
REVIEW_ROADMAP_MODEL_NAME: Model name (e.g., 'claude-opus-4-5', 'gpt-4o').
Expand All @@ -39,7 +43,60 @@ class Settings(BaseSettings):
)

# GitHub
GITHUB_TOKEN: str
GITHUB_TOKEN: Optional[str] = None
REVIEW_ROADMAP_GITHUB_TOKENS: Optional[str] = None

@model_validator(mode="after")
def validate_github_token(self) -> "Settings":
"""Ensure at least one GitHub token is configured."""
if not self.GITHUB_TOKEN and not self.REVIEW_ROADMAP_GITHUB_TOKENS:
raise ValueError(
"Either GITHUB_TOKEN or REVIEW_ROADMAP_GITHUB_TOKENS must be set"
)
return self

def get_github_tokens(self) -> List[str]:
"""Get the list of GitHub tokens to try, in order of precedence.

When REVIEW_ROADMAP_GITHUB_TOKENS is set, those tokens take precedence
and are returned first. GITHUB_TOKEN (if set and not already in the list)
is appended as a fallback.

Returns:
List of unique, non-empty GitHub tokens to try.
"""
tokens: List[str] = []

# REVIEW_ROADMAP_GITHUB_TOKENS takes precedence
if self.REVIEW_ROADMAP_GITHUB_TOKENS:
tokens.extend(
t.strip()
for t in self.REVIEW_ROADMAP_GITHUB_TOKENS.split(",")
if t.strip()
)

# Add GITHUB_TOKEN as fallback if not already included
if self.GITHUB_TOKEN and self.GITHUB_TOKEN not in tokens:
tokens.append(self.GITHUB_TOKEN)

return tokens

def get_default_github_token(self) -> str:
"""Get the default GitHub token for read operations.

Returns the first available token (REVIEW_ROADMAP_GITHUB_TOKENS
takes precedence over GITHUB_TOKEN).

Returns:
The first available GitHub token.

Raises:
ValueError: If no tokens are configured.
"""
tokens = self.get_github_tokens()
if not tokens:
raise ValueError("No GitHub tokens configured")
return tokens[0]

# LLM Configuration (prefixed to avoid conflicts with shell environment)
REVIEW_ROADMAP_LLM_PROVIDER: str = "anthropic"
Expand Down
97 changes: 95 additions & 2 deletions review_roadmap/github/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
generate review roadmaps.
"""

from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple

import httpx
Expand All @@ -16,6 +17,21 @@
)


@dataclass
class TokenSearchResult:
"""Result of searching for a token with write access.

Attributes:
token: The token that was found with write access, or None if no token worked.
access_result: The WriteAccessResult from checking the successful token,
or the last failed result if no token worked.
tokens_tried: Number of tokens that were tested.
"""
token: Optional[str]
access_result: WriteAccessResult
tokens_tried: int


class GitHubClient:
"""Synchronous GitHub API client for PR data retrieval.

Expand All @@ -37,9 +53,10 @@ def __init__(self, token: Optional[str] = None):
"""Initialize the GitHub client.

Args:
token: GitHub API token. If not provided, uses GITHUB_TOKEN from settings.
token: GitHub API token. If not provided, uses the first available
token from settings (REVIEW_ROADMAP_GITHUB_TOKENS takes precedence).
"""
self.token = token or settings.GITHUB_TOKEN
self.token = token or settings.get_default_github_token()
self.headers = {
"Authorization": f"Bearer {self.token}",
"Accept": "application/vnd.github.v3+json",
Expand Down Expand Up @@ -471,3 +488,79 @@ def post_pr_comment(self, owner: str, repo: str, pr_number: int, body: str) -> D
)
resp.raise_for_status()
return resp.json()


def find_working_token(
owner: str, repo: str, pr_number: int
) -> TokenSearchResult:
"""Find a GitHub token with write access from the configured tokens.

Iterates through all configured tokens (from REVIEW_ROADMAP_GITHUB_TOKENS
and GITHUB_TOKEN) and tests each one for write access using the
check_write_access method with a live reaction test.

Args:
owner: Repository owner.
repo: Repository name.
pr_number: PR number for live write testing.

Returns:
TokenSearchResult containing:
- token: The first token with GRANTED status, or None if none worked
- access_result: The WriteAccessResult from the successful check,
or the last failed result if no token worked
- tokens_tried: Number of tokens that were tested

Example:
>>> result = find_working_token("owner", "repo", 123)
>>> if result.token:
... client = GitHubClient(token=result.token)
... client.post_pr_comment(owner, repo, 123, "Hello!")
"""
tokens = settings.get_github_tokens()

if not tokens:
return TokenSearchResult(
token=None,
access_result=WriteAccessResult(
status=WriteAccessStatus.DENIED,
is_fine_grained_pat=False,
message="No GitHub tokens configured."
),
tokens_tried=0
)

last_result: Optional[WriteAccessResult] = None

for i, token in enumerate(tokens, start=1):
client = GitHubClient(token=token)
try:
result = client.check_write_access(owner, repo, pr_number)
last_result = result

if result.status == WriteAccessStatus.GRANTED:
return TokenSearchResult(
token=token,
access_result=result,
tokens_tried=i
)
except Exception:
# Token failed to even check access (e.g., network error, invalid token)
# Continue to next token
last_result = WriteAccessResult(
status=WriteAccessStatus.DENIED,
is_fine_grained_pat=False,
message=f"Token failed basic validation (may be invalid or revoked)."
)
continue

# No token worked - return the last result
return TokenSearchResult(
token=None,
access_result=last_result or WriteAccessResult(
status=WriteAccessStatus.DENIED,
is_fine_grained_pat=False,
message="No tokens were successfully tested."
),
tokens_tried=len(tokens)
)
50 changes: 36 additions & 14 deletions review_roadmap/main.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import typer
from rich.console import Console
from rich.markdown import Markdown
from review_roadmap.github.client import GitHubClient
from review_roadmap.github.client import GitHubClient, find_working_token
from review_roadmap.agent.graph import build_graph
from review_roadmap.config import settings
from review_roadmap.logging import configure_logging
Expand Down Expand Up @@ -59,27 +59,49 @@ def generate(
console.print("[red]Invalid PR format. Use 'owner/repo/number' or a full URL.[/red]")
raise typer.Exit(code=1)

# Initialize GitHub client
# Initialize GitHub client (default token for read operations)
gh_client = GitHubClient()

# Token with write access (may differ from default if using multi-token)
write_token = None

# Check write access early if posting is requested (fail fast before LLM generation)
if post:
console.print(f"[bold blue]Checking write access for {owner}/{repo}...[/bold blue]")
tokens = settings.get_github_tokens()

if len(tokens) > 1:
console.print(f"[bold blue]Searching {len(tokens)} tokens for write access to {owner}/{repo}...[/bold blue]")
else:
console.print(f"[bold blue]Checking write access for {owner}/{repo}...[/bold blue]")

try:
# Pass pr_number to enable live write test for fine-grained PATs
access_result = gh_client.check_write_access(owner, repo, pr_number)
# Search through available tokens for one with write access
search_result = find_working_token(owner, repo, pr_number)

if access_result.status == WriteAccessStatus.DENIED:
console.print(
f"[red]Error: {access_result.message}[/red]\n"
"[yellow]To use --post, your token needs 'Pull requests: Read and write' permission.[/yellow]"
)
raise typer.Exit(code=1)
elif access_result.status == WriteAccessStatus.UNCERTAIN:
console.print(f"[yellow]Warning: {access_result.message}[/yellow]")
if search_result.token:
write_token = search_result.token
if search_result.tokens_tried > 1:
console.print(f"[green]Write access confirmed (token {search_result.tokens_tried} of {len(tokens)}).[/green]")
else:
console.print("[green]Write access confirmed.[/green]")
# Update client to use the working token for subsequent operations
gh_client = GitHubClient(token=write_token)
elif search_result.access_result.status == WriteAccessStatus.UNCERTAIN:
console.print(f"[yellow]Warning: {search_result.access_result.message}[/yellow]")
console.print("[yellow]Proceeding, but posting may fail...[/yellow]")
else:
console.print("[green]Write access confirmed.[/green]")
if search_result.tokens_tried > 1:
console.print(
f"[red]Error: None of the {search_result.tokens_tried} configured tokens have write access.[/red]\n"
f"[red]Last error: {search_result.access_result.message}[/red]\n"
"[yellow]To use --post, at least one token needs 'Pull requests: Read and write' permission.[/yellow]"
)
else:
console.print(
f"[red]Error: {search_result.access_result.message}[/red]\n"
"[yellow]To use --post, your token needs 'Pull requests: Read and write' permission.[/yellow]"
)
raise typer.Exit(code=1)
except typer.Exit:
raise
except Exception as e:
Expand Down
Loading