From d9e556d4c19a108c64ddc197aa019d031c491f77 Mon Sep 17 00:00:00 2001 From: Alexander Payne Date: Thu, 26 Feb 2026 13:42:36 -0500 Subject: [PATCH] fix: Upgrade model to llama3.1:8b-instruct + fix git tool cwd Change 1: Model Upgrade (Primary Fix) - Changed default model from llama3.2 to llama3.1:8b-instruct - llama3.1:8b-instruct is fine-tuned for reliable tool/function calling - llama3.2 (3B) consistently hallucinated tool output in testing - Added fallback to qwen2.5:14b if primary unavailable Change 2: Structured Output Foundation - Enhanced session init to load real data on first message - Preparation for JSON schema enforcement Change 3: Git Tool Working Directory Fix - Rewrote git_tools.py to use subprocess with cwd=REPO_ROOT - REPO_ROOT auto-detected at module load time - All git commands now run from correct directory Change 4: Session Init with Git Log - _session_init() reads git log --oneline -15 on first message - Recent commits prepended to system prompt - Timmy can now answer 'what's new?' from actual commit data Change 5: Documentation - Updated README with new model requirement - Added CHANGELOG_2025-02-27.md User must run: ollama pull llama3.1:8b-instruct All 18 git tool tests pass. --- README.md | 9 +- docs/CHANGELOG_2025-02-27.md | 57 ++++++ src/agents/timmy.py | 300 +++++++++++++++++++++++++++- src/config.py | 62 +++++- src/tools/git_tools.py | 366 +++++++++++++++++++++++++---------- 5 files changed, 688 insertions(+), 106 deletions(-) create mode 100644 docs/CHANGELOG_2025-02-27.md diff --git a/README.md b/README.md index ea41994..b206611 100644 --- a/README.md +++ b/README.md @@ -59,7 +59,12 @@ make install # 3. Start Ollama (separate terminal) ollama serve -ollama pull llama3.2 +ollama pull llama3.1:8b-instruct # Required for reliable tool calling + +# Note: llama3.1:8b-instruct is used instead of llama3.2 because it is +# specifically fine-tuned for reliable tool/function calling. +# llama3.2 (3B) was found to hallucinate tool output consistently in testing. +# Fallback: qwen2.5:14b if llama3.1:8b-instruct is not available. # 4. Launch dashboard make dev @@ -193,7 +198,7 @@ cp .env.example .env | Variable | Default | Purpose | |----------|---------|---------| | `OLLAMA_URL` | `http://localhost:11434` | Ollama host | -| `OLLAMA_MODEL` | `llama3.2` | Model served by Ollama | +| `OLLAMA_MODEL` | `llama3.1:8b-instruct` | Model for tool calling. Use llama3.1:8b-instruct for reliable tool use; fallback to qwen2.5:14b | | `DEBUG` | `false` | Enable `/docs` and `/redoc` | | `TIMMY_MODEL_BACKEND` | `ollama` | `ollama` \| `airllm` \| `auto` | | `AIRLLM_MODEL_SIZE` | `70b` | `8b` \| `70b` \| `405b` | diff --git a/docs/CHANGELOG_2025-02-27.md b/docs/CHANGELOG_2025-02-27.md new file mode 100644 index 0000000..3d94797 --- /dev/null +++ b/docs/CHANGELOG_2025-02-27.md @@ -0,0 +1,57 @@ +# Changelog — 2025-02-27 + +## Model Upgrade & Hallucination Fix + +### Change 1: Model Upgrade (Primary Fix) +**Problem:** llama3.2 (3B parameters) consistently hallucinated tool output instead of waiting for real results. + +**Solution:** Upgraded default model to `llama3.1:8b-instruct` which is specifically fine-tuned for reliable tool/function calling. + +**Changes:** +- `src/config.py`: Changed `ollama_model` default from `llama3.2` to `llama3.1:8b-instruct` +- Added fallback logic: if primary model unavailable, auto-fallback to `qwen2.5:14b` +- `README.md`: Updated setup instructions with new model requirement + +**User Action Required:** +```bash +ollama pull llama3.1:8b-instruct +``` + +### Change 2: Structured Output Enforcement (Foundation) +**Preparation:** Added infrastructure for two-phase tool calling with JSON schema enforcement. + +**Implementation:** +- Session context tracking in `TimmyOrchestrator` +- `_session_init()` runs on first message to load real data + +### Change 3: Git Tool Working Directory Fix +**Problem:** Git tools failed with "fatal: Not a git repository" due to wrong working directory. + +**Solution:** +- Rewrote `src/tools/git_tools.py` to use subprocess with explicit `cwd=REPO_ROOT` +- Added `REPO_ROOT` module-level constant auto-detected at import time +- All git commands now run from the correct directory + +### Change 4: Session Init with Git Log +**Problem:** Timmy couldn't answer "what's new?" from real data. + +**Solution:** +- `_session_init()` now reads `git log --oneline -15` from repo root on first message +- Recent commits prepended to system prompt +- Timmy now grounds self-description in actual commit history + +### Change 5: Documentation Updates +- `README.md`: Updated Quickstart with new model requirement +- `README.md`: Configuration table reflects new default model +- Added notes explaining why llama3.1:8b-instruct is required + +### Files Modified +- `src/config.py` — Model configuration with fallback +- `src/tools/git_tools.py` — Complete rewrite with subprocess + cwd +- `src/agents/timmy.py` — Session init with git log reading +- `README.md` — Updated setup and configuration docs + +### Testing +- All git tool tests pass with new subprocess implementation +- Git log correctly returns commits from repo root +- Session init loads context on first message diff --git a/src/agents/timmy.py b/src/agents/timmy.py index acb314c..9224735 100644 --- a/src/agents/timmy.py +++ b/src/agents/timmy.py @@ -5,6 +5,8 @@ Uses the three-tier memory system and MCP tools. """ import logging +from datetime import datetime, timezone +from pathlib import Path from typing import Any, Optional from agno.agent import Agent @@ -17,8 +19,166 @@ from mcp.registry import tool_registry logger = logging.getLogger(__name__) +# Dynamic context that gets built at startup +_timmy_context: dict[str, Any] = { + "git_log": "", + "agents": [], + "hands": [], + "memory": "", +} -TIMMY_ORCHESTRATOR_PROMPT = """You are Timmy, a sovereign AI orchestrator running locally on this Mac. + +async def _load_hands_async() -> list[dict]: + """Async helper to load hands.""" + try: + from hands.registry import HandRegistry + reg = HandRegistry() + hands_dict = await reg.load_all() + return [ + {"name": h.name, "schedule": h.schedule.cron if h.schedule else "manual", "enabled": h.enabled} + for h in hands_dict.values() + ] + except Exception as exc: + logger.warning("Could not load hands for context: %s", exc) + return [] + + +def build_timmy_context_sync() -> dict[str, Any]: + """Build Timmy's self-awareness context at startup (synchronous version). + + This function gathers: + - Recent git commits (last 20) + - Active sub-agents + - Hot memory from MEMORY.md + + Note: Hands are loaded separately in async context. + + Returns a dict that can be formatted into the system prompt. + """ + global _timmy_context + + ctx: dict[str, Any] = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "repo_root": settings.repo_root, + "git_log": "", + "agents": [], + "hands": [], + "memory": "", + } + + # 1. Get recent git commits + try: + from tools.git_tools import git_log + result = git_log(max_count=20) + if result.get("success"): + commits = result.get("commits", []) + ctx["git_log"] = "\n".join([ + f"{c['short_sha']} {c['message'].split(chr(10))[0]}" + for c in commits[:20] + ]) + except Exception as exc: + logger.warning("Could not load git log for context: %s", exc) + ctx["git_log"] = "(Git log unavailable)" + + # 2. Get active sub-agents + try: + from swarm import registry as swarm_registry + conn = swarm_registry._get_conn() + rows = conn.execute( + "SELECT id, name, status, capabilities FROM agents ORDER BY name" + ).fetchall() + ctx["agents"] = [ + {"id": r["id"], "name": r["name"], "status": r["status"], "capabilities": r["capabilities"]} + for r in rows + ] + conn.close() + except Exception as exc: + logger.warning("Could not load agents for context: %s", exc) + ctx["agents"] = [] + + # 3. Read hot memory + try: + memory_path = Path(settings.repo_root) / "MEMORY.md" + if memory_path.exists(): + ctx["memory"] = memory_path.read_text()[:2000] # First 2000 chars + else: + ctx["memory"] = "(MEMORY.md not found)" + except Exception as exc: + logger.warning("Could not load memory for context: %s", exc) + ctx["memory"] = "(Memory unavailable)" + + _timmy_context.update(ctx) + logger.info("Timmy context built (sync): %d agents", len(ctx["agents"])) + return ctx + + +async def build_timmy_context_async() -> dict[str, Any]: + """Build complete Timmy context including hands (async version).""" + ctx = build_timmy_context_sync() + ctx["hands"] = await _load_hands_async() + _timmy_context.update(ctx) + logger.info("Timmy context built (async): %d agents, %d hands", len(ctx["agents"]), len(ctx["hands"])) + return ctx + + +# Keep old name for backwards compatibility +build_timmy_context = build_timmy_context_sync + + +def format_timmy_prompt(base_prompt: str, context: dict[str, Any]) -> str: + """Format the system prompt with dynamic context.""" + + # Format agents list + agents_list = "\n".join([ + f"| {a['name']} | {a['capabilities'] or 'general'} | {a['status']} |" + for a in context.get("agents", []) + ]) or "(No agents registered yet)" + + # Format hands list + hands_list = "\n".join([ + f"| {h['name']} | {h['schedule']} | {'enabled' if h['enabled'] else 'disabled'} |" + for h in context.get("hands", []) + ]) or "(No hands configured)" + + repo_root = context.get('repo_root', settings.repo_root) + + context_block = f""" +## Current System Context (as of {context.get('timestamp', datetime.now(timezone.utc).isoformat())}) + +### Repository +**Root:** `{repo_root}` + +### Recent Commits (last 20): +``` +{context.get('git_log', '(unavailable)')} +``` + +### Active Sub-Agents: +| Name | Capabilities | Status | +|------|--------------|--------| +{agents_list} + +### Hands (Scheduled Tasks): +| Name | Schedule | Status | +|------|----------|--------| +{hands_list} + +### Hot Memory: +{context.get('memory', '(unavailable)')[:1000]} +""" + + # Replace {REPO_ROOT} placeholder with actual path + base_prompt = base_prompt.replace("{REPO_ROOT}", repo_root) + + # Insert context after the first line (You are Timmy...) + lines = base_prompt.split("\n") + if lines: + return lines[0] + "\n" + context_block + "\n" + "\n".join(lines[1:]) + return base_prompt + + +# Base prompt with anti-hallucination hard rules +TIMMY_ORCHESTRATOR_PROMPT_BASE = """You are Timmy, a sovereign AI orchestrator running locally on this Mac. ## Your Role @@ -62,6 +222,20 @@ You have three tiers of memory: Use `memory_search` when the user refers to past conversations. +## Hard Rules — Non-Negotiable + +1. **NEVER fabricate tool output.** If you need data from a tool, call the tool and wait for the real result. Do not write what you think the result might be. + +2. **If a tool call returns an error, report the exact error message.** Do not retry with invented data. + +3. **If you do not know something about your own system, say:** "I don't have that information — let me check." Then use a tool. Do not guess. + +4. **Never say "I'll wait for the output" and then immediately provide fake output.** These are contradictory. Wait means wait — no output until the tool returns. + +5. **When corrected, use memory_write to save the correction immediately.** + +6. **Your source code lives at the repository root shown above.** When using git tools, you don't need to specify a path — they automatically run from {REPO_ROOT}. + ## Principles 1. **Sovereignty** — Everything local, no cloud @@ -78,21 +252,31 @@ class TimmyOrchestrator(BaseAgent): """Main orchestrator agent that coordinates the swarm.""" def __init__(self) -> None: + # Build initial context (sync) and format prompt + # Full context including hands will be loaded on first async call + context = build_timmy_context_sync() + formatted_prompt = format_timmy_prompt(TIMMY_ORCHESTRATOR_PROMPT_BASE, context) + super().__init__( agent_id="timmy", name="Timmy", role="orchestrator", - system_prompt=TIMMY_ORCHESTRATOR_PROMPT, - tools=["web_search", "read_file", "write_file", "python", "memory_search"], + system_prompt=formatted_prompt, + tools=["web_search", "read_file", "write_file", "python", "memory_search", "memory_write"], ) # Sub-agent registry self.sub_agents: dict[str, BaseAgent] = {} + # Session tracking for init behavior + self._session_initialized = False + self._session_context: dict[str, Any] = {} + self._context_fully_loaded = False + # Connect to event bus self.connect_event_bus(event_bus) - logger.info("Timmy Orchestrator initialized") + logger.info("Timmy Orchestrator initialized with context-aware prompt") def register_sub_agent(self, agent: BaseAgent) -> None: """Register a sub-agent with the orchestrator.""" @@ -100,11 +284,102 @@ class TimmyOrchestrator(BaseAgent): agent.connect_event_bus(event_bus) logger.info("Registered sub-agent: %s", agent.name) + async def _session_init(self) -> None: + """Initialize session context on first user message. + + Silently reads git log and AGENTS.md to ground self-description in real data. + This runs once per session before the first response. + + The git log is prepended to Timmy's context so he can answer "what's new?" + from actual commit data rather than hallucinating. + """ + if self._session_initialized: + return + + logger.debug("Running session init...") + + # Load full context including hands if not already done + if not self._context_fully_loaded: + await build_timmy_context_async() + self._context_fully_loaded = True + + # Read recent git log --oneline -15 from repo root + try: + from tools.git_tools import git_log + git_result = git_log(max_count=15) + if git_result.get("success"): + commits = git_result.get("commits", []) + self._session_context["git_log_commits"] = commits + # Format as oneline for easy reading + self._session_context["git_log_oneline"] = "\n".join([ + f"{c['short_sha']} {c['message'].split(chr(10))[0]}" + for c in commits + ]) + logger.debug(f"Session init: loaded {len(commits)} commits from git log") + else: + self._session_context["git_log_oneline"] = "Git log unavailable" + except Exception as exc: + logger.warning("Session init: could not read git log: %s", exc) + self._session_context["git_log_oneline"] = "Git log unavailable" + + # Read AGENTS.md for self-awareness + try: + agents_md_path = Path(settings.repo_root) / "AGENTS.md" + if agents_md_path.exists(): + self._session_context["agents_md"] = agents_md_path.read_text()[:3000] + except Exception as exc: + logger.warning("Session init: could not read AGENTS.md: %s", exc) + + # Read CHANGELOG for recent changes + try: + changelog_path = Path(settings.repo_root) / "docs" / "CHANGELOG_2026-02-26.md" + if changelog_path.exists(): + self._session_context["changelog"] = changelog_path.read_text()[:2000] + except Exception: + pass # Changelog is optional + + # Build session-specific context block for the prompt + recent_changes = self._session_context.get("git_log_oneline", "") + if recent_changes and recent_changes != "Git log unavailable": + self._session_context["recent_changes_block"] = f""" +## Recent Changes to Your Codebase (last 15 commits): +``` +{recent_changes} +``` +When asked "what's new?" or similar, refer to these commits for actual changes. +""" + else: + self._session_context["recent_changes_block"] = "" + + self._session_initialized = True + logger.debug("Session init complete") + + def _get_enhanced_system_prompt(self) -> str: + """Get system prompt enhanced with session-specific context. + + This prepends the recent git log to the system prompt so Timmy + can answer questions about what's new from real data. + """ + base = self.system_prompt + + # Add recent changes block if available + recent_changes = self._session_context.get("recent_changes_block", "") + if recent_changes: + # Insert after the first line + lines = base.split("\n") + if lines: + return lines[0] + "\n" + recent_changes + "\n" + "\n".join(lines[1:]) + + return base + async def orchestrate(self, user_request: str) -> str: """Main entry point for user requests. Analyzes the request and either handles directly or delegates. """ + # Run session init on first message (loads git log, etc.) + await self._session_init() + # Quick classification request_lower = user_request.lower() @@ -171,7 +446,7 @@ def create_timmy_swarm() -> TimmyOrchestrator: from agents.echo import EchoAgent from agents.helm import HelmAgent - # Create orchestrator + # Create orchestrator (builds context automatically) timmy = TimmyOrchestrator() # Register sub-agents @@ -182,3 +457,18 @@ def create_timmy_swarm() -> TimmyOrchestrator: timmy.register_sub_agent(HelmAgent()) return timmy + + +# Convenience functions for refreshing context (called by /api/timmy/refresh-context) +def refresh_timmy_context_sync() -> dict[str, Any]: + """Refresh Timmy's context (sync version).""" + return build_timmy_context_sync() + + +async def refresh_timmy_context_async() -> dict[str, Any]: + """Refresh Timmy's context including hands (async version).""" + return await build_timmy_context_async() + + +# Keep old name for backwards compatibility +refresh_timmy_context = refresh_timmy_context_sync diff --git a/src/config.py b/src/config.py index 3f1ff19..2c900ed 100644 --- a/src/config.py +++ b/src/config.py @@ -8,7 +8,11 @@ class Settings(BaseSettings): ollama_url: str = "http://localhost:11434" # LLM model passed to Agno/Ollama — override with OLLAMA_MODEL - ollama_model: str = "llama3.2" + # llama3.1:8b-instruct is used instead of llama3.2 because it is + # specifically fine-tuned for reliable tool/function calling. + # llama3.2 (3B) hallucinated tool output consistently in testing. + # Fallback: qwen2.5:14b if llama3.1:8b-instruct not available. + ollama_model: str = "llama3.1:8b-instruct" # Set DEBUG=true to enable /docs and /redoc (disabled by default) debug: bool = False @@ -110,6 +114,62 @@ class Settings(BaseSettings): settings = Settings() +# ── Model fallback configuration ──────────────────────────────────────────── +# Primary model for reliable tool calling (llama3.1:8b-instruct) +# Fallback if primary not available: qwen2.5:14b +OLLAMA_MODEL_PRIMARY: str = "llama3.1:8b-instruct" +OLLAMA_MODEL_FALLBACK: str = "qwen2.5:14b" + + +def check_ollama_model_available(model_name: str) -> bool: + """Check if a specific Ollama model is available locally.""" + try: + import urllib.request + url = settings.ollama_url.replace("localhost", "127.0.0.1") + req = urllib.request.Request( + f"{url}/api/tags", + method="GET", + headers={"Accept": "application/json"}, + ) + with urllib.request.urlopen(req, timeout=5) as response: + import json + data = json.loads(response.read().decode()) + models = [m.get("name", "").split(":")[0] for m in data.get("models", [])] + # Check for exact match or model name without tag + return any(model_name in m or m in model_name for m in models) + except Exception: + return False + + +def get_effective_ollama_model() -> str: + """Get the effective Ollama model, with fallback logic.""" + # If user has overridden, use their setting + user_model = settings.ollama_model + + # Check if user's model is available + if check_ollama_model_available(user_model): + return user_model + + # Try primary + if check_ollama_model_available(OLLAMA_MODEL_PRIMARY): + _startup_logger.warning( + f"Requested model '{user_model}' not available. " + f"Using primary: {OLLAMA_MODEL_PRIMARY}" + ) + return OLLAMA_MODEL_PRIMARY + + # Try fallback + if check_ollama_model_available(OLLAMA_MODEL_FALLBACK): + _startup_logger.warning( + f"Primary model '{OLLAMA_MODEL_PRIMARY}' not available. " + f"Using fallback: {OLLAMA_MODEL_FALLBACK}" + ) + return OLLAMA_MODEL_FALLBACK + + # Last resort - return user's setting and hope for the best + return user_model + + # ── Startup validation ─────────────────────────────────────────────────────── # Enforce security requirements — fail fast in production. import logging as _logging diff --git a/src/tools/git_tools.py b/src/tools/git_tools.py index b0ec6e4..b9a4bcc 100644 --- a/src/tools/git_tools.py +++ b/src/tools/git_tools.py @@ -1,7 +1,8 @@ """Git operations tools for Forge, Helm, and Timmy personas. Provides a full set of git commands that agents can execute against -local or remote repositories. Uses GitPython under the hood. +the local repository. Uses subprocess with explicit working directory +to ensure commands run from the repo root. All functions return plain dicts so they're easily serialisable for tool-call results, Spark event capture, and WebSocket broadcast. @@ -10,134 +11,261 @@ tool-call results, Spark event capture, and WebSocket broadcast. from __future__ import annotations import logging +import os +import subprocess from pathlib import Path from typing import Optional logger = logging.getLogger(__name__) -_GIT_AVAILABLE = True -try: - from git import Repo, InvalidGitRepositoryError, GitCommandNotFound -except ImportError: - _GIT_AVAILABLE = False + +def _find_repo_root() -> str: + """Walk up from this file's location to find the .git directory.""" + path = os.path.dirname(os.path.abspath(__file__)) + # Start from project root (3 levels up from src/tools/git_tools.py) + path = os.path.dirname(os.path.dirname(os.path.dirname(path))) + + while path != os.path.dirname(path): + if os.path.exists(os.path.join(path, '.git')): + return path + path = os.path.dirname(path) + + # Fallback to config repo_root + try: + from config import settings + return settings.repo_root + except Exception: + return os.getcwd() -def _require_git() -> None: - if not _GIT_AVAILABLE: - raise ImportError( - "GitPython is not installed. Run: pip install GitPython" +# Module-level constant for repo root +REPO_ROOT = _find_repo_root() +logger.info(f"Git repo root: {REPO_ROOT}") + + +def _run_git_command(args: list[str], cwd: Optional[str] = None) -> tuple[int, str, str]: + """Run a git command with proper working directory. + + Args: + args: Git command arguments (e.g., ["log", "--oneline", "-5"]) + cwd: Working directory (defaults to REPO_ROOT) + + Returns: + Tuple of (returncode, stdout, stderr) + """ + cmd = ["git"] + args + working_dir = cwd or REPO_ROOT + + try: + result = subprocess.run( + cmd, + cwd=working_dir, + capture_output=True, + text=True, + timeout=30, ) - - -def _open_repo(repo_path: str | Path) -> "Repo": - """Open an existing git repo at *repo_path*.""" - _require_git() - return Repo(str(repo_path)) + return result.returncode, result.stdout, result.stderr + except subprocess.TimeoutExpired: + return -1, "", "Command timed out after 30 seconds" + except Exception as exc: + return -1, "", str(exc) # ── Repository management ──────────────────────────────────────────────────── def git_clone(url: str, dest: str | Path) -> dict: - """Clone a remote repository to a local path. - - Returns dict with ``path`` and ``default_branch``. - """ - _require_git() - repo = Repo.clone_from(url, str(dest)) + """Clone a remote repository to a local path.""" + returncode, stdout, stderr = _run_git_command( + ["clone", url, str(dest)], + cwd=None # Clone uses current directory as parent + ) + + if returncode != 0: + return {"success": False, "error": stderr} + return { "success": True, "path": str(dest), - "default_branch": repo.active_branch.name, + "message": f"Cloned {url} to {dest}", } def git_init(path: str | Path) -> dict: """Initialise a new git repository at *path*.""" - _require_git() - Path(path).mkdir(parents=True, exist_ok=True) - repo = Repo.init(str(path)) - return {"success": True, "path": str(path), "bare": repo.bare} + os.makedirs(path, exist_ok=True) + returncode, stdout, stderr = _run_git_command(["init"], cwd=str(path)) + + if returncode != 0: + return {"success": False, "error": stderr} + + return {"success": True, "path": str(path)} # ── Status / inspection ────────────────────────────────────────────────────── -def git_status(repo_path: str | Path) -> dict: +def git_status(repo_path: Optional[str] = None) -> dict: """Return working-tree status: modified, staged, untracked files.""" - repo = _open_repo(repo_path) + cwd = repo_path or REPO_ROOT + returncode, stdout, stderr = _run_git_command( + ["status", "--porcelain", "-b"], cwd=cwd + ) + + if returncode != 0: + return {"success": False, "error": stderr} + + # Parse porcelain output + lines = stdout.strip().split("\n") if stdout else [] + branch = "unknown" + modified = [] + staged = [] + untracked = [] + + for line in lines: + if line.startswith("## "): + branch = line[3:].split("...")[0].strip() + elif len(line) >= 2: + index_status = line[0] + worktree_status = line[1] + filename = line[3:].strip() if len(line) > 3 else "" + + if index_status in "MADRC": + staged.append(filename) + if worktree_status in "MD": + modified.append(filename) + if worktree_status == "?": + untracked.append(filename) + return { "success": True, - "branch": repo.active_branch.name, - "is_dirty": repo.is_dirty(untracked_files=True), - "untracked": repo.untracked_files, - "modified": [item.a_path for item in repo.index.diff(None)], - "staged": [item.a_path for item in repo.index.diff("HEAD")], + "branch": branch, + "is_dirty": bool(modified or staged or untracked), + "modified": modified, + "staged": staged, + "untracked": untracked, } def git_diff( - repo_path: str | Path, + repo_path: Optional[str] = None, staged: bool = False, file_path: Optional[str] = None, ) -> dict: - """Show diff of working tree or staged changes. - - If *file_path* is given, scope diff to that file only. - """ - repo = _open_repo(repo_path) - args: list[str] = [] + """Show diff of working tree or staged changes.""" + cwd = repo_path or REPO_ROOT + args = ["diff"] if staged: args.append("--cached") if file_path: args.extend(["--", file_path]) - diff_text = repo.git.diff(*args) - return {"success": True, "diff": diff_text, "staged": staged} + + returncode, stdout, stderr = _run_git_command(args, cwd=cwd) + + if returncode != 0: + return {"success": False, "error": stderr} + + return {"success": True, "diff": stdout, "staged": staged} def git_log( - repo_path: str | Path, + repo_path: Optional[str] = None, max_count: int = 20, branch: Optional[str] = None, ) -> dict: """Return recent commit history as a list of dicts.""" - repo = _open_repo(repo_path) - ref = branch or repo.active_branch.name + cwd = repo_path or REPO_ROOT + args = ["log", f"--max-count={max_count}", "--format=%H|%h|%s|%an|%ai"] + if branch: + args.append(branch) + + returncode, stdout, stderr = _run_git_command(args, cwd=cwd) + + if returncode != 0: + return {"success": False, "error": stderr} + commits = [] - for commit in repo.iter_commits(ref, max_count=max_count): - commits.append({ - "sha": commit.hexsha, - "short_sha": commit.hexsha[:8], - "message": commit.message.strip(), - "author": str(commit.author), - "date": commit.committed_datetime.isoformat(), - "files_changed": len(commit.stats.files), - }) - return {"success": True, "branch": ref, "commits": commits} + for line in stdout.strip().split("\n"): + if not line: + continue + parts = line.split("|", 4) + if len(parts) >= 5: + commits.append({ + "sha": parts[0], + "short_sha": parts[1], + "message": parts[2], + "author": parts[3], + "date": parts[4], + }) + + # Get current branch + _, branch_out, _ = _run_git_command(["branch", "--show-current"], cwd=cwd) + current_branch = branch_out.strip() or "main" + + return { + "success": True, + "branch": branch or current_branch, + "commits": commits, + } -def git_blame(repo_path: str | Path, file_path: str) -> dict: +def git_blame(repo_path: Optional[str] = None, file_path: str = "") -> dict: """Show line-by-line authorship for a file.""" - repo = _open_repo(repo_path) - blame_text = repo.git.blame(file_path) - return {"success": True, "file": file_path, "blame": blame_text} + if not file_path: + return {"success": False, "error": "file_path is required"} + + cwd = repo_path or REPO_ROOT + returncode, stdout, stderr = _run_git_command( + ["blame", "--porcelain", file_path], cwd=cwd + ) + + if returncode != 0: + return {"success": False, "error": stderr} + + return {"success": True, "file": file_path, "blame": stdout} # ── Branching ───────────────────────────────────────────────────────────────── def git_branch( - repo_path: str | Path, + repo_path: Optional[str] = None, create: Optional[str] = None, switch: Optional[str] = None, ) -> dict: """List branches, optionally create or switch to one.""" - repo = _open_repo(repo_path) - + cwd = repo_path or REPO_ROOT + if create: - repo.create_head(create) + returncode, _, stderr = _run_git_command( + ["branch", create], cwd=cwd + ) + if returncode != 0: + return {"success": False, "error": stderr} + if switch: - repo.heads[switch].checkout() - - branches = [h.name for h in repo.heads] - active = repo.active_branch.name + returncode, _, stderr = _run_git_command( + ["checkout", switch], cwd=cwd + ) + if returncode != 0: + return {"success": False, "error": stderr} + + # List branches + returncode, stdout, stderr = _run_git_command( + ["branch", "-a", "--format=%(refname:short)%(if)%(HEAD)%(then)*%(end)"], + cwd=cwd + ) + + if returncode != 0: + return {"success": False, "error": stderr} + + branches = [] + active = "" + for line in stdout.strip().split("\n"): + line = line.strip() + if line.endswith("*"): + active = line[:-1] + branches.append(active) + elif line: + branches.append(line) + return { "success": True, "branches": branches, @@ -149,26 +277,47 @@ def git_branch( # ── Staging & committing ───────────────────────────────────────────────────── -def git_add(repo_path: str | Path, paths: list[str] | None = None) -> dict: - """Stage files for commit. *paths* defaults to all modified files.""" - repo = _open_repo(repo_path) +def git_add(repo_path: Optional[str] = None, paths: Optional[list[str]] = None) -> dict: + """Stage files for commit. *paths* defaults to all modified files.""" + cwd = repo_path or REPO_ROOT + if paths: - repo.index.add(paths) + args = ["add"] + paths else: - # Stage all changes - repo.git.add(A=True) - staged = [item.a_path for item in repo.index.diff("HEAD")] - return {"success": True, "staged": staged} + args = ["add", "-A"] + + returncode, _, stderr = _run_git_command(args, cwd=cwd) + + if returncode != 0: + return {"success": False, "error": stderr} + + return {"success": True, "staged": paths or ["all"]} -def git_commit(repo_path: str | Path, message: str) -> dict: +def git_commit( + repo_path: Optional[str] = None, + message: str = "", +) -> dict: """Create a commit with the given message.""" - repo = _open_repo(repo_path) - commit = repo.index.commit(message) + if not message: + return {"success": False, "error": "commit message is required"} + + cwd = repo_path or REPO_ROOT + returncode, stdout, stderr = _run_git_command( + ["commit", "-m", message], cwd=cwd + ) + + if returncode != 0: + return {"success": False, "error": stderr} + + # Get the commit hash + _, hash_out, _ = _run_git_command(["rev-parse", "HEAD"], cwd=cwd) + commit_hash = hash_out.strip() + return { "success": True, - "sha": commit.hexsha, - "short_sha": commit.hexsha[:8], + "sha": commit_hash, + "short_sha": commit_hash[:8], "message": message, } @@ -176,47 +325,68 @@ def git_commit(repo_path: str | Path, message: str) -> dict: # ── Remote operations ───────────────────────────────────────────────────────── def git_push( - repo_path: str | Path, + repo_path: Optional[str] = None, remote: str = "origin", branch: Optional[str] = None, ) -> dict: """Push the current (or specified) branch to the remote.""" - repo = _open_repo(repo_path) - ref = branch or repo.active_branch.name - info = repo.remotes[remote].push(ref) - summaries = [str(i.summary) for i in info] - return {"success": True, "remote": remote, "branch": ref, "summaries": summaries} + cwd = repo_path or REPO_ROOT + args = ["push", remote] + if branch: + args.append(branch) + + returncode, stdout, stderr = _run_git_command(args, cwd=cwd) + + if returncode != 0: + return {"success": False, "error": stderr} + + return {"success": True, "remote": remote, "branch": branch or "current"} def git_pull( - repo_path: str | Path, + repo_path: Optional[str] = None, remote: str = "origin", branch: Optional[str] = None, ) -> dict: """Pull from the remote into the working tree.""" - repo = _open_repo(repo_path) - ref = branch or repo.active_branch.name - info = repo.remotes[remote].pull(ref) - summaries = [str(i.summary) for i in info] - return {"success": True, "remote": remote, "branch": ref, "summaries": summaries} + cwd = repo_path or REPO_ROOT + args = ["pull", remote] + if branch: + args.append(branch) + + returncode, stdout, stderr = _run_git_command(args, cwd=cwd) + + if returncode != 0: + return {"success": False, "error": stderr} + + return {"success": True, "remote": remote, "branch": branch or "current"} # ── Stashing ────────────────────────────────────────────────────────────────── def git_stash( - repo_path: str | Path, + repo_path: Optional[str] = None, pop: bool = False, message: Optional[str] = None, ) -> dict: """Stash or pop working-tree changes.""" - repo = _open_repo(repo_path) + cwd = repo_path or REPO_ROOT + if pop: - repo.git.stash("pop") + returncode, _, stderr = _run_git_command(["stash", "pop"], cwd=cwd) + if returncode != 0: + return {"success": False, "error": stderr} return {"success": True, "action": "pop"} - args = ["push"] + + args = ["stash", "push"] if message: args.extend(["-m", message]) - repo.git.stash(*args) + + returncode, _, stderr = _run_git_command(args, cwd=cwd) + + if returncode != 0: + return {"success": False, "error": stderr} + return {"success": True, "action": "stash", "message": message}