--- File: aggregate.py --- # aggregate.py from __future__ import annotations """ Note(Gemini): This module orchestrates the construction of the final Markdown context string. Instead of sending every file to the AI raw (which blows up tokens), this uses a pipeline: 1. Resolve paths (handles globs and absolute paths). 2. Build file items (raw content). 3. If 'summary_only' is true (which is the default behavior now), it pipes the files through summarize.py to generate a compacted view. This is essential for keeping prompt tokens low while giving the AI enough structural info to use the MCP tools to fetch only what it needs. """ import tomllib import re import glob from pathlib import Path, PureWindowsPath from typing import Any import summarize import project_manager from file_cache import ASTParser def find_next_increment(output_dir: Path, namespace: str) -> int: ... def is_absolute_with_drive(entry: str) -> bool: ... def resolve_paths(base_dir: Path, entry: str) -> list[Path]: ... def build_discussion_section(history: list[str]) -> str: ... def build_files_section(base_dir: Path, files: list[str | dict[str, Any]]) -> str: ... def build_screenshots_section(base_dir: Path, screenshots: list[str]) -> str: ... def build_file_items(base_dir: Path, files: list[str | dict[str, Any]]) -> list[dict[str, Any]]: """ Return a list of dicts describing each file, for use by ai_client when it wants to upload individual files rather than inline everything as markdown. Each dict has: path : Path (resolved absolute path) entry : str (original config entry string) content : str (file text, or error string) error : bool mtime : float (last modification time, for skip-if-unchanged optimization) tier : int | None (optional tier for context management) """ ... def build_summary_section(base_dir: Path, files: list[str | dict[str, Any]]) -> str: """ Build a compact summary section using summarize.py — one short block per file. Used as the initial block instead of full file contents. """ ... def _build_files_section_from_items(file_items: list[dict[str, Any]]) -> str: """Build the files markdown section from pre-read file items (avoids double I/O).""" ... def build_markdown_from_items(file_items: list[dict[str, Any]], screenshot_base_dir: Path, screenshots: list[str], history: list[str], summary_only: bool = False) -> str: """Build markdown from pre-read file items instead of re-reading from disk.""" ... def build_markdown_no_history(file_items: list[dict[str, Any]], screenshot_base_dir: Path, screenshots: list[str], summary_only: bool = False) -> str: """Build markdown with only files + screenshots (no history). Used for stable caching.""" ... def build_discussion_text(history: list[str]) -> str: """Build just the discussion history section text. Returns empty string if no history.""" ... def build_tier1_context(file_items: list[dict[str, Any]], screenshot_base_dir: Path, screenshots: list[str], history: list[str]) -> str: """ Tier 1 Context: Strategic/Orchestration. Full content for core conductor files and files with tier=1, summaries for others. """ ... def build_tier2_context(file_items: list[dict[str, Any]], screenshot_base_dir: Path, screenshots: list[str], history: list[str]) -> str: """ Tier 2 Context: Architectural/Tech Lead. Full content for all files (standard behavior). """ ... def build_tier3_context(file_items: list[dict[str, Any]], screenshot_base_dir: Path, screenshots: list[str], history: list[str], focus_files: list[str]) -> str: """ Tier 3 Context: Execution/Worker. Full content for focus_files and files with tier=3, summaries/skeletons for others. """ ... def build_markdown(base_dir: Path, files: list[str | dict[str, Any]], screenshot_base_dir: Path, screenshots: list[str], history: list[str], summary_only: bool = False) -> str: ... def run(config: dict[str, Any]) -> tuple[str, Path, list[dict[str, Any]]]: ... def main() -> None: # Load global config to find active project ... if __name__ == "__main__": main() --- File: ai_client.py --- # ai_client.py from __future__ import annotations """ Note(Gemini): Acts as the unified interface for multiple LLM providers (Anthropic, Gemini). Abstracts away the differences in how they handle tool schemas, history, and caching. For Anthropic: aggressively manages the ~200k token limit by manually culling stale [FILES UPDATED] entries and dropping the oldest message pairs. For Gemini: injects the initial context directly into system_instruction during chat creation to avoid massive history bloat. """ # ai_client.py import tomllib import json import sys import time import datetime import hashlib import difflib import threading import requests from typing import Optional, Callable, Any import os import project_manager import file_cache import mcp_client import anthropic from gemini_cli_adapter import GeminiCliAdapter from google import genai from google.genai import types from events import EventEmitter _provider: str = "gemini" _model: str = "gemini-2.5-flash-lite" _temperature: float = 0.0 _max_tokens: int = 8192 _history_trunc_limit: int = 8000 # Global event emitter for API lifecycle events events: EventEmitter = EventEmitter() def set_model_params(temp: float, max_tok: int, trunc_limit: int = 8000) -> None: ... def get_history_trunc_limit() -> int: ... def set_history_trunc_limit(val: int) -> None: ... _gemini_client: genai.Client | None = None _gemini_chat: Any = None _gemini_cache: Any = None _gemini_cache_md_hash: int | None = None _gemini_cache_created_at: float | None = None # Gemini cache TTL in seconds. Caches are created with this TTL and # proactively rebuilt at 90% of this value to avoid stale-reference errors. _GEMINI_CACHE_TTL: int = 3600 _anthropic_client: anthropic.Anthropic | None = None _anthropic_history: list[dict] = [] _anthropic_history_lock: threading.Lock = threading.Lock() _deepseek_client: Any = None _deepseek_history: list[dict] = [] _deepseek_history_lock: threading.Lock = threading.Lock() _send_lock: threading.Lock = threading.Lock() _gemini_cli_adapter: GeminiCliAdapter | None = None # Injected by gui.py - called when AI wants to run a command. # Signature: (script: str, base_dir: str) -> str | None confirm_and_run_callback: Callable[[str, str], str | None] | None = None # Injected by gui.py - called whenever a comms entry is appended. # Signature: (entry: dict) -> None comms_log_callback: Callable[[dict[str, Any]], None] | None = None # Injected by gui.py - called whenever a tool call completes. # Signature: (script: str, result: str) -> None tool_log_callback: Callable[[str, str], None] | None = None # Set by caller tiers before ai_client.send(); cleared in finally. # Safe — ai_client.send() calls are serialized by the MMA engine executor. current_tier: str | None = None # Increased to allow thorough code exploration before forcing a summary MAX_TOOL_ROUNDS: int = 10 # Maximum cumulative bytes of tool output allowed per send() call. # Prevents unbounded memory growth during long tool-calling loops. _MAX_TOOL_OUTPUT_BYTES: int = 500_000 # Maximum characters per text chunk sent to Anthropic. # Kept well under the ~200k token API limit. _ANTHROPIC_CHUNK_SIZE: int = 120_000 _SYSTEM_PROMPT: str = ( "You are a helpful coding assistant with access to a PowerShell tool and MCP tools (file access: read_file, list_directory, search_files, get_file_summary, web access: web_search, fetch_url). " "When calling file/directory tools, always use the 'path' parameter for the target path. " "When asked to create or edit files, prefer targeted edits over full rewrites. " "Always explain what you are doing before invoking the tool.\n\n" "When writing or rewriting large files (especially those containing quotes, backticks, or special characters), " "avoid python -c with inline strings. Instead: (1) write a .py helper script to disk using a PS here-string " "(@'...'@ for literal content), (2) run it with `python