Skip to content
Closed
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
3 changes: 2 additions & 1 deletion .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -2,4 +2,5 @@ __pycache__/
*.pyc
.coverage
.pytest_cache/
.env
.env
.DS_Store
22 changes: 16 additions & 6 deletions plumb/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -115,7 +115,7 @@ def _init_clone_setup(repo_root: Path, cfg: PlumbConfig) -> None:
hooks_dir = repo_root / ".git" / "hooks"
hooks_dir.mkdir(exist_ok=True)
hook_path = hooks_dir / "pre-commit"
hook_path.write_text("#!/bin/sh\nplumb hook\nexit $?\n")
hook_path.write_text('#!/bin/sh\n[ "$PLUMB_SKIP" = "1" ] && exit 0\nplumb hook\nexit $?\n')
hook_path.chmod(0o755)
post_commit_path = hooks_dir / "post-commit"
post_commit_path.write_text("#!/bin/sh\nplumb post-commit\n")
Expand Down Expand Up @@ -249,7 +249,7 @@ def init():
hooks_dir = repo_root / ".git" / "hooks"
hooks_dir.mkdir(exist_ok=True)
hook_path = hooks_dir / "pre-commit"
hook_path.write_text("#!/bin/sh\nplumb hook\nexit $?\n")
hook_path.write_text('#!/bin/sh\n[ "$PLUMB_SKIP" = "1" ] && exit 0\nplumb hook\nexit $?\n')
hook_path.chmod(0o755)
post_commit_path = hooks_dir / "post-commit"
post_commit_path.write_text("#!/bin/sh\nplumb post-commit\n")
Expand Down Expand Up @@ -806,11 +806,14 @@ def map_tests(dry_run):
console.print(f"Found {len(test_summaries)} test functions and {len(requirements)} requirements.")
console.print("Running LLM mapping...")

from plumb.programs import configure_dspy, run_chunked_mapper
from plumb.programs import configure_dspy, run_chunked_mapper, get_program_lm, get_program_config
from plumb.programs.test_mapper import TestMapper

configure_dspy()
mapper = TestMapper()
override_lm = get_program_lm("test_mapper")
prog_cfg = get_program_config("test_mapper") or {}
budget = prog_cfg.get("budget", 60000)

req_json = json.dumps([{"id": r["id"], "text": r["text"]} for r in requirements])
items = [(s["name"], json.dumps(s)) for s in test_summaries]
Expand All @@ -819,9 +822,16 @@ def _combine(chunk):
return json.dumps([json.loads(t) for _, t in chunk])

try:
mappings = run_chunked_mapper(
mapper, req_json, items, budget=60000, combine_fn=_combine,
)
if override_lm:
import dspy
with dspy.context(lm=override_lm):
mappings = run_chunked_mapper(
mapper, req_json, items, budget=budget, combine_fn=_combine,
)
else:
mappings = run_chunked_mapper(
mapper, req_json, items, budget=budget, combine_fn=_combine,
)
except Exception as e:
console.print(f"[red]Mapping failed: {e}[/red]")
raise SystemExit(1)
Expand Down
26 changes: 21 additions & 5 deletions plumb/coverage_reporter.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
from rich.table import Table

from plumb.config import load_config
from plumb.ignore import is_ignored, parse_plumbignore

PLUMB_MARKER_RE = re.compile(r'#\s*plumb:(req-[a-f0-9]+)')
FUNC_NAME_RE = re.compile(r'def test_req_([a-f0-9]+)_')
Expand All @@ -29,6 +30,7 @@ def run_pytest_coverage(repo_root: str | Path) -> dict | None:
result = subprocess.run(
[
sys.executable, "-m", "pytest",
"-m", "not slow",
"--cov=.",
f"--cov-report=json:{cov_json}",
"--cov-report=",
Expand Down Expand Up @@ -118,11 +120,14 @@ def _collect_source_summaries(repo_root: Path) -> dict[str, str]:
"""
import ast

ignore_patterns = parse_plumbignore(repo_root)
per_file: dict[str, str] = {}
for item in sorted(repo_root.rglob("*.py")):
rel = str(item.relative_to(repo_root))
if ".plumb" in rel or "test_" in item.name or rel.startswith("tests/"):
continue
if is_ignored(rel, ignore_patterns):
continue
try:
content = item.read_text()
except Exception:
Expand Down Expand Up @@ -324,11 +329,14 @@ def check_spec_to_code_coverage(
return (0, len(requirements))

# --- LLM mapping ---
from plumb.programs import configure_dspy, run_chunked_mapper
from plumb.programs import configure_dspy, run_chunked_mapper, get_program_lm, get_program_config
from plumb.programs.code_coverage_mapper import CodeCoverageMapper

configure_dspy()
mapper = CodeCoverageMapper()
override_lm = get_program_lm("code_coverage_mapper", repo_root)
prog_cfg = get_program_config("code_coverage_mapper", repo_root) or {}
budget = prog_cfg.get("budget", 60000)

if full_remap:
dirty_reqs = requirements
Expand All @@ -346,10 +354,18 @@ def check_spec_to_code_coverage(
def _combine(chunk):
return "\n\n".join(text for _, text in chunk)

results = run_chunked_mapper(
mapper, req_json, items, budget=60000,
combine_fn=_combine, merge_fn=merge_coverage_results,
)
if override_lm:
import dspy
with dspy.context(lm=override_lm):
results = run_chunked_mapper(
mapper, req_json, items, budget=budget,
combine_fn=_combine, merge_fn=merge_coverage_results,
)
else:
results = run_chunked_mapper(
mapper, req_json, items, budget=budget,
combine_fn=_combine, merge_fn=merge_coverage_results,
)

# Build fresh results dict from LLM output
fresh_results: dict[str, dict] = {}
Expand Down
9 changes: 7 additions & 2 deletions plumb/ignore.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,12 +50,17 @@ def is_ignored(filepath: str, patterns: list[str]) -> bool:
- Exact match: ``README.md``
- Glob matched against the basename: ``*.txt``
- Directory prefix (pattern ends with ``/``): ``docs/`` matches ``docs/foo``
- Glob directory prefix: ``.venv*/`` matches ``.venv3.10/foo``
"""
basename = Path(filepath).name
top_dir = filepath.split("/")[0]
for pat in patterns:
if pat.endswith("/"):
# Directory prefix — match if filepath starts with the prefix
if filepath == pat.rstrip("/") or filepath.startswith(pat):
prefix = pat.rstrip("/")
# Directory prefix — exact startswith or fnmatch on top directory
if filepath == prefix or filepath.startswith(pat):
return True
if fnmatch(top_dir, prefix):
return True
else:
# Exact full-path match or fnmatch against basename
Expand Down
77 changes: 54 additions & 23 deletions plumb/programs/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,63 +5,78 @@

import dspy
from dspy.adapters import XMLAdapter
from dspy.clients.base_lm import BaseLM

from plumb import PlumbAuthError, PlumbInferenceError

_configured = False

_NO_BACKEND_MSG = (
"No LLM backend available.\n"
"Option 1: Set ANTHROPIC_API_KEY in .env or environment (direct API, fastest)\n"
"Option 2: Install Claude Code CLI — https://claude.ai/code (uses your subscription)"
)

def get_lm() -> dspy.LM:
return dspy.LM("anthropic/claude-sonnet-4-20250514", max_tokens=28000)

def get_lm() -> BaseLM:
"""Return the best available LM: direct API if ANTHROPIC_API_KEY is set,
otherwise Claude Code CLI if available."""
if os.environ.get("ANTHROPIC_API_KEY"):
return dspy.LM("anthropic/claude-sonnet-4-20250514", max_tokens=28000)

from plumb.programs.claude_code_lm import ClaudeCodeLM, find_claude_cli

if find_claude_cli():
return ClaudeCodeLM(model="sonnet", max_tokens=28000)

raise PlumbAuthError(_NO_BACKEND_MSG)


def configure_dspy() -> None:
"""Lazy DSPy configuration. No-op if already configured.
Never call at import time — ANTHROPIC_API_KEY absence would break
Never call at import time — missing auth would break
non-LLM commands like plumb status."""
global _configured
if _configured:
return
from dotenv import load_dotenv

load_dotenv(override=False)
lm = get_lm()
dspy.configure(lm=lm, adapter=XMLAdapter())
_configured = True


def validate_api_access() -> None:
"""Check that ANTHROPIC_API_KEY is set and works. Loads .env first, then
falls back to exported environment variables. Performs a smoke test to
verify the key is valid. Raises PlumbAuthError if not found or invalid."""
"""Check that an LLM backend is available and working.

Tries ANTHROPIC_API_KEY first (direct API), then falls back to the
Claude Code CLI. Performs a smoke test to verify the backend works.
Raises PlumbAuthError if neither is available or working.
"""
from dotenv import load_dotenv

load_dotenv(override=False)
if not os.environ.get("ANTHROPIC_API_KEY"):
raise PlumbAuthError(
"ANTHROPIC_API_KEY is not set. "
"Plumb requires a valid Anthropic API key to analyze commits.\n"
"Set it in a .env file or export it: export ANTHROPIC_API_KEY=your-key-here"
)

# Smoke test: verify the key actually works
lm = get_lm()

lm = get_lm() # raises PlumbAuthError if no backend available

try:
response = lm("Reply with only the word: hello")
if not response:
raise PlumbAuthError("API returned empty response - key may be invalid")
raise PlumbAuthError("LLM returned empty response - backend may be misconfigured")
except PlumbAuthError:
raise
except Exception as e:
err_str = str(e).lower()
if "auth" in err_str or "api key" in err_str or "401" in err_str:
raise PlumbAuthError(
f"ANTHROPIC_API_KEY is invalid or rejected: {e}"
) from e
raise PlumbAuthError(
f"Failed to verify API access: {e}"
) from e
raise PlumbAuthError(f"Failed to verify LLM access: {e}") from e


def get_program_lm(program_name: str, repo_root: str | Path | None = None) -> dspy.LM | None:
"""Return a per-program LM override from config, or None for the default."""
def get_program_config(program_name: str, repo_root: str | Path | None = None) -> dict | None:
"""Return the raw program_models entry for a program, or None."""
from plumb.config import find_repo_root, load_config

if repo_root is None:
Expand All @@ -71,14 +86,29 @@ def get_program_lm(program_name: str, repo_root: str | Path | None = None) -> ds
cfg = load_config(repo_root)
if cfg is None:
return None
entry = cfg.program_models.get(program_name)
return cfg.program_models.get(program_name)


def get_program_lm(program_name: str, repo_root: str | Path | None = None) -> BaseLM | None:
"""Return a per-program LM override from config, or None for the default."""
entry = get_program_config(program_name, repo_root)
if entry is None:
return None
model = entry.get("model")
if not model:
return None
max_tokens = entry.get("max_tokens", 8192)
return dspy.LM(model, max_tokens=max_tokens)

if os.environ.get("ANTHROPIC_API_KEY"):
return dspy.LM(model, max_tokens=max_tokens)

from plumb.programs.claude_code_lm import ClaudeCodeLM, find_claude_cli

if find_claude_cli():
cli_model = model.removeprefix("anthropic/")
return ClaudeCodeLM(model=cli_model, max_tokens=max_tokens)

return None


def run_with_retries(fn, *args, max_retries: int = 2, **kwargs):
Expand All @@ -94,6 +124,7 @@ def run_with_retries(fn, *args, max_retries: int = 2, **kwargs):
raise PlumbAuthError(
f"API key is invalid or rejected: {e}"
) from e
print(f"[retry {attempt+1}/{max_retries+1}] {type(e).__name__}: {e}")
last_error = e
raise PlumbInferenceError(
f"LLM inference failed after {max_retries + 1} attempts: {last_error}"
Expand Down
Loading