#!/usr/bin/env python3
"""ACE-Tool CLI - Semantic code search and prompt enhancement."""

import argparse
import json
import sys
from pathlib import Path

import httpx

try:
    from .client import AceToolClient
    from .indexer import Indexer
    from .utils import load_env
    from .web_ui import run_interactive_enhance
except ImportError:
    from client import AceToolClient
    from indexer import Indexer
    from utils import load_env
    from web_ui import run_interactive_enhance

load_env()


def cmd_search_context(args):
    """Handle search_context command."""
    client = AceToolClient(args.api_url, args.token, args.endpoint)
    result = client.search_context(args.project_root, args.query)
    print(json.dumps(result, indent=2, ensure_ascii=False))


def cmd_enhance_prompt(args):
    """Handle enhance_prompt command."""
    client = AceToolClient(args.api_url, args.token, args.endpoint, args.reasoning_effort)

    history = args.history
    if args.history_file:
        history = Path(args.history_file).read_text(encoding="utf-8")

    if not args.no_interactive:
        result = run_interactive_enhance(
            client,
            args.prompt,
            history,
            args.port,
            auto_open_browser=not args.no_browser,
            project_root=args.project_root,
        )
        if result:
            print(json.dumps({"enhanced_prompt": result}, indent=2, ensure_ascii=False))
        else:
            print(json.dumps({"cancelled": True}), file=sys.stderr)
            sys.exit(1)
    else:
        result = client.enhance_prompt(args.prompt, history, args.project_root)
        if "error" in result:
            print(json.dumps(result), file=sys.stderr)
            sys.exit(1)
        print(json.dumps(result, indent=2, ensure_ascii=False))


def cmd_get_config(args):
    """Handle get_config command."""
    client = AceToolClient(args.api_url, args.token, args.endpoint)
    result = client.get_config()
    print(json.dumps(result, indent=2, ensure_ascii=False))


def cmd_index(args):
    """Handle index command."""
    client = AceToolClient(args.api_url, args.token, args.endpoint)
    if not client.base_url or not client.token:
        print(
            json.dumps(
                {
                    "error": "No authentication configured for indexing. Set up session.json, AUGMENT_SESSION_AUTH, or ACE_API_URL/ACE_API_TOKEN"
                }
            ),
            file=sys.stderr,
        )
        sys.exit(1)
    indexer = Indexer(args.project_root, client.base_url, client.token)
    blob_names = indexer.get_blob_names()
    result = {
        "total_blobs": len(blob_names),
        "last_indexed": indexer._index.last_indexed,
        "project_root": args.project_root,
        "effective_root": str(indexer.root),
    }
    print(json.dumps(result, indent=2, ensure_ascii=False))


def main():
    parser = argparse.ArgumentParser(description="ACE-Tool CLI - Semantic code search and prompt enhancement")
    parser.add_argument("--api-url", help="API base URL")
    parser.add_argument("--token", help="API authentication token")
    parser.add_argument(
        "--endpoint",
        choices=["new", "old", "claude", "openai", "gemini", "codex"],
        help="Enhancer endpoint type",
    )

    subparsers = parser.add_subparsers(dest="command", required=True)

    p_search = subparsers.add_parser("search_context", help="Search codebase with natural language")
    p_search.add_argument("-p", "--project-root", required=True, help="Project root path")
    p_search.add_argument("-q", "--query", required=True, help="Natural language query")
    p_search.set_defaults(func=cmd_search_context)

    p_enhance = subparsers.add_parser("enhance_prompt", help="Enhance prompt with context")
    p_enhance.add_argument("-p", "--prompt", required=True, help="Original prompt")
    p_enhance.add_argument("-H", "--history", default="", help="Conversation history")
    p_enhance.add_argument("--history-file", help="File containing conversation history")
    p_enhance.add_argument("--project-root", help="Project root path (optional)")
    p_enhance.add_argument(
        "--reasoning-effort",
        default=None,
        metavar="LEVEL",
        help="Reasoning effort: none/minimal/low/medium/high/xhigh/max (default: none, or PROMPT_ENHANCER_REASONING_EFFORT; empty string omits the parameter)",
    )
    p_enhance.add_argument("--no-interactive", action="store_true", help="Disable web UI, output JSON directly")
    p_enhance.add_argument("--no-browser", action="store_true", help="Don't auto-open browser, just print URL")
    p_enhance.add_argument("--port", type=int, default=8765, help="Port for interactive web server (default: 8765)")
    p_enhance.set_defaults(func=cmd_enhance_prompt)

    p_config = subparsers.add_parser("get_config", help="Show current configuration")
    p_config.set_defaults(func=cmd_get_config)

    p_index = subparsers.add_parser("index", help="Index project for remote search")
    p_index.add_argument("-p", "--project-root", required=True, help="Project root path")
    p_index.set_defaults(func=cmd_index)

    args = parser.parse_args()
    try:
        args.func(args)
    except httpx.HTTPStatusError as e:
        print(json.dumps({"error": str(e), "status_code": e.response.status_code}), file=sys.stderr)
        sys.exit(1)
    except Exception as e:
        print(json.dumps({"error": str(e)}), file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

"""ACE-Tool API client for semantic search and prompt enhancement."""

import json
import os
import re
import uuid
from pathlib import Path

import httpx
from tenacity import retry, retry_if_not_exception_type, stop_after_attempt, wait_exponential

try:
    from .indexer import Indexer
    from .templates import (
        DEFAULT_CLAUDE_MODEL,
        DEFAULT_CODEX_MODEL,
        DEFAULT_GEMINI_MODEL,
        DEFAULT_MODEL,
        DEFAULT_OPENAI_MODEL,
        ENCODING_CHAIN,
        ENHANCE_PROMPT_TEMPLATE,
        ENV_ENHANCER_ENDPOINT,
        ENV_ENHANCER_ENDPOINT_LEGACY,
        ENV_ENHANCER_INCLUDE_SEARCH_CONTEXT,
        ENV_ENHANCER_REASONING_EFFORT,
        EXCLUDE_PATTERNS,
        ITERATIVE_ENHANCE_TEMPLATE,
        NO_RELEVANT_CODE_CONTEXT,
        RETRIEVAL_TIMEOUT,
        SEARCH_CONTEXT_CHAR_LIMIT,
        TEXT_EXTENSIONS,
        USER_AGENT,
    )
    from .utils import (
        build_api_url,
        detect_and_read,
        get_session_id,
        is_chinese_text,
        load_session_auth,
        parse_chat_history,
    )
except ImportError:
    from indexer import Indexer
    from templates import (
        DEFAULT_CLAUDE_MODEL,
        DEFAULT_CODEX_MODEL,
        DEFAULT_GEMINI_MODEL,
        DEFAULT_MODEL,
        DEFAULT_OPENAI_MODEL,
        ENCODING_CHAIN,
        ENHANCE_PROMPT_TEMPLATE,
        ENV_ENHANCER_ENDPOINT,
        ENV_ENHANCER_ENDPOINT_LEGACY,
        ENV_ENHANCER_INCLUDE_SEARCH_CONTEXT,
        ENV_ENHANCER_REASONING_EFFORT,
        EXCLUDE_PATTERNS,
        ITERATIVE_ENHANCE_TEMPLATE,
        NO_RELEVANT_CODE_CONTEXT,
        RETRIEVAL_TIMEOUT,
        SEARCH_CONTEXT_CHAR_LIMIT,
        TEXT_EXTENSIONS,
        USER_AGENT,
    )
    from utils import (
        build_api_url,
        detect_and_read,
        get_session_id,
        is_chinese_text,
        load_session_auth,
        parse_chat_history,
    )

import logging

log = logging.getLogger(__name__)

_ENHANCED_PROMPT_RE = re.compile(
    r"<augment-enhanced-prompt(?:\s+[^>]*)?>\s*(.*?)\s*</augment-enhanced-prompt\s*>",
    re.DOTALL,
)
_THIRD_PARTY_ENDPOINTS = frozenset({"claude", "openai", "gemini", "codex"})
_TRUE_ENV_VALUES = frozenset({"1", "true", "yes", "on"})


class AceToolClient:
    """Client for ACE-Tool API endpoints."""

    def __init__(
        self,
        base_url: str | None = None,
        token: str | None = None,
        endpoint: str | None = None,
        reasoning_effort: str | None = None,
    ):
        # Determine auth source priority: constructor params > session auth
        if base_url is not None or token is not None:
            # Constructor params provided (full or partial override)
            self.auth_source = "constructor"
            loaded_base_url, loaded_token, _ = load_session_auth()
            self.base_url = (base_url if base_url is not None else loaded_base_url or "").rstrip("/")
            self.token = token if token is not None else loaded_token or ""
        else:
            # No constructor params, use load_session_auth()
            loaded_base_url, loaded_token, auth_source = load_session_auth()
            self.base_url = (loaded_base_url or "").rstrip("/")
            self.token = loaded_token or ""
            self.auth_source = auth_source

        # Endpoint resolution: PROMPT_ENHANCER_ENDPOINT > ACE_ENHANCER_ENDPOINT > constructor arg > "new"
        env_endpoint = os.getenv(ENV_ENHANCER_ENDPOINT, "") or os.getenv(ENV_ENHANCER_ENDPOINT_LEGACY, "")
        resolved = env_endpoint or endpoint or "new"
        self.endpoint = resolved.lower()

        self.timeout = httpx.Timeout(180.0, connect=30.0)

        self.third_party_base_url = os.getenv("PROMPT_ENHANCER_BASE_URL", "").rstrip("/")
        self.third_party_token = os.getenv("PROMPT_ENHANCER_TOKEN", "")
        self.third_party_model = os.getenv("PROMPT_ENHANCER_MODEL", "")
        # Reasoning control: default none (enhancement needs no deep reasoning);
        # empty string sends no reasoning parameter at all.
        env_val = os.environ.get(ENV_ENHANCER_REASONING_EFFORT)
        if reasoning_effort is not None:
            self.reasoning_effort = reasoning_effort.strip().lower()
        elif env_val is not None:
            self.reasoning_effort = env_val.strip().lower()
        else:
            self.reasoning_effort = "none"

    def _get_headers(self, use_third_party: bool = False) -> dict:
        headers = {
            "Content-Type": "application/json",
            "User-Agent": USER_AGENT,
            "x-request-id": str(uuid.uuid4()),
            "x-request-session-id": get_session_id(),
        }
        if use_third_party and self.third_party_token:
            headers["Authorization"] = f"Bearer {self.third_party_token}"
        elif self.token:
            headers["Authorization"] = f"Bearer {self.token}"
        return headers

    def _is_third_party(self) -> bool:
        return self.endpoint in _THIRD_PARTY_ENDPOINTS

    def _should_include_search_context(self) -> bool:
        """Check if search context injection is enabled via env var."""
        val = os.environ.get(ENV_ENHANCER_INCLUDE_SEARCH_CONTEXT, "").strip().lower()
        return val in _TRUE_ENV_VALUES

    @staticmethod
    def _normalize_search_context(text: str) -> str:
        """Normalize search context: placeholder for empty, truncate if over limit."""
        stripped = text.strip()
        if not stripped or stripped == NO_RELEVANT_CODE_CONTEXT:
            return NO_RELEVANT_CODE_CONTEXT
        if len(stripped) > SEARCH_CONTEXT_CHAR_LIMIT:
            return stripped[:SEARCH_CONTEXT_CHAR_LIMIT] + "\n\n[codebase_context truncated for length]"
        return stripped

    @staticmethod
    def _build_prompt_with_search_context(original: str, ctx: str) -> str:
        """Wrap original prompt with codebase context in XML tags."""
        return f"<codebase_context>\n{ctx}\n</codebase_context>\n\n<original_request>\n{original}\n</original_request>"

    def _maybe_inject_search_context(self, endpoint: str, prompt: str, project_root: str | None) -> str:
        """Inject search context into prompt if enabled and applicable."""
        if not self._is_third_party() or not self._should_include_search_context():
            return prompt
        if not project_root:
            raise ValueError("project_root is required when search context injection is enabled")
        if not self.base_url or not self.token:
            raise ValueError(
                "No authentication configured for search context injection. Set up session.json, AUGMENT_SESSION_AUTH, or ACE_API_URL/ACE_API_TOKEN"
            )
        result = self._remote_search(project_root, prompt)
        raw_ctx = result.get("results", "")
        ctx = self._normalize_search_context(raw_ctx)
        return self._build_prompt_with_search_context(prompt, ctx)

    def _get_third_party_model(self) -> str:
        if self.third_party_model:
            return self.third_party_model
        return {
            "claude": DEFAULT_CLAUDE_MODEL,
            "openai": DEFAULT_OPENAI_MODEL,
            "gemini": DEFAULT_GEMINI_MODEL,
            "codex": DEFAULT_CODEX_MODEL,
        }.get(self.endpoint, DEFAULT_MODEL)

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
    def search_context(self, project_root: str, query: str) -> dict:
        """Search codebase: remote via ACE API if available, else local fallback."""
        if self.base_url and self.token:
            try:
                return self._remote_search(project_root, query)
            except httpx.HTTPStatusError as e:
                if e.response.status_code in (401, 403):
                    log.error("Auth failed (%d) for remote search: %s", e.response.status_code, e)
                else:
                    log.warning("Remote search failed, falling back to local: %s", e)
            except Exception as e:
                log.warning("Remote search failed, falling back to local: %s", e)
        return self._local_search(project_root, query)

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_not_exception_type(ValueError),
    )
    def enhance_prompt(
        self,
        prompt: str,
        conversation_history: str,
        project_root: str | None = None,
    ) -> dict:
        """Enhance prompt with codebase context and conversation history."""
        if self._is_third_party():
            return self._call_third_party_api(prompt, conversation_history, project_root)

        if not self.base_url:
            return {"enhanced_prompt": prompt, "note": "No API configured, returning original"}

        chat_history = parse_chat_history(conversation_history)

        if self.endpoint == "old":
            return self._call_old_endpoint(prompt, chat_history, project_root)
        return self._call_new_endpoint(prompt, chat_history, project_root)

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_not_exception_type(ValueError),
    )
    def iterative_enhance(
        self,
        original_prompt: str,
        previous_enhanced: str,
        current_prompt: str,
        conversation_history: str,
        project_root: str | None = None,
    ) -> dict:
        """Iteratively enhance an already-enhanced prompt, preserving user modifications."""
        iterative_prompt = ITERATIVE_ENHANCE_TEMPLATE.format(
            original_prompt=original_prompt,
            previous_enhanced=previous_enhanced,
            current_prompt=current_prompt,
        )

        if self._is_third_party():
            return self._call_third_party_api_raw(iterative_prompt, conversation_history, project_root)

        if not self.base_url:
            return {"enhanced_prompt": current_prompt, "note": "No API configured, returning current"}

        chat_history = parse_chat_history(conversation_history)

        if self.endpoint == "old":
            return self._call_old_endpoint_raw(iterative_prompt, chat_history, project_root)
        return self._call_new_endpoint_raw(iterative_prompt, chat_history, project_root)

    def _call_new_endpoint(self, prompt: str, chat_history: list[dict], project_root: str | None = None) -> dict:
        """Call /prompt-enhancer endpoint (new)."""
        context = self._get_retrieval_context(project_root, prompt)
        enriched_prompt = f"{context}{prompt}" if context else prompt
        payload = {
            "nodes": [{"id": 0, "type": 0, "text_node": {"content": enriched_prompt}}],
            "chat_history": chat_history,
            "conversation_id": None,
            "model": DEFAULT_MODEL,
            "mode": "CHAT",
        }

        data = self._post_json(
            build_api_url(self.base_url, "/prompt-enhancer"),
            payload,
            headers=self._get_headers(),
        )
        return {"enhanced_prompt": data.get("text", prompt)}

    def _call_new_endpoint_raw(
        self, raw_prompt: str, chat_history: list[dict], project_root: str | None = None
    ) -> dict:
        """Call /prompt-enhancer with pre-built prompt."""
        context = self._get_retrieval_context(project_root, raw_prompt)
        enriched_prompt = f"{context}{raw_prompt}" if context else raw_prompt
        payload = {
            "nodes": [{"id": 0, "type": 0, "text_node": {"content": enriched_prompt}}],
            "chat_history": chat_history,
            "conversation_id": None,
            "model": DEFAULT_MODEL,
            "mode": "CHAT",
        }

        data = self._post_json(
            build_api_url(self.base_url, "/prompt-enhancer"),
            payload,
            headers=self._get_headers(),
        )
        text = data.get("text", raw_prompt)
        return {"enhanced_prompt": self._extract_enhanced_prompt(text)}

    def _build_old_payload(
        self, message: str, chat_history: list[dict], language_guideline: str, blob_names: list[str] | None = None
    ) -> dict:
        """Build payload for old endpoint."""
        return {
            "model": DEFAULT_MODEL,
            "path": None,
            "prefix": None,
            "selected_code": None,
            "suffix": None,
            "message": message,
            "chat_history": chat_history,
            "lang": None,
            "blobs": {"checkpoint_id": None, "added_blobs": blob_names or [], "deleted_blobs": []},
            "user_guided_blobs": [],
            "context_code_exchange_request_id": None,
            "external_source_ids": [],
            "disable_auto_external_sources": None,
            "user_guidelines": language_guideline,
            "workspace_guidelines": "",
            "feature_detection_flags": {"support_parallel_tool_use": None},
            "third_party_override": None,
            "tool_definitions": [],
            "nodes": [{"id": 1, "type": 0, "text_node": {"content": message}}],
            "mode": "CHAT",
            "agent_memories": None,
            "persona_type": None,
            "rules": [],
            "silent": None,
            "enable_parallel_tool_use": None,
            "conversation_id": None,
            "system_prompt": None,
        }

    def _get_blob_names(self, project_root: str | None) -> list[str] | None:
        """Get blob_names from indexer if API and project_root are available."""
        if not project_root or not self.base_url or not self.token:
            return None
        try:
            indexer = Indexer(project_root, self.base_url, self.token)
            return indexer.get_blob_names()
        except httpx.HTTPStatusError as e:
            if e.response.status_code in (401, 403):
                log.error("Auth failed (%d) getting blob names: %s", e.response.status_code, e)
            else:
                log.warning("Failed to get blob names: %s", e)
            return None
        except Exception as e:
            log.warning("Failed to get blob names: %s", e)
            return None

    def _get_retrieval_context(self, project_root: str | None, query: str) -> str:
        """Get cloud retrieval context to inject into prompts for non-old endpoints."""
        if not project_root or not self.base_url or not self.token:
            return ""
        try:
            result = self._remote_search(project_root, query)
            context = result.get("results", "")
            if context:
                return f"\n\n<codebase-context>\n{context}\n</codebase-context>\n\n"
        except httpx.HTTPStatusError as e:
            if e.response.status_code in (401, 403):
                log.error("Auth failed (%d) for cloud retrieval: %s", e.response.status_code, e)
            else:
                log.warning("Cloud retrieval failed, proceeding without context: %s", e)
        except Exception as e:
            log.warning("Cloud retrieval failed, proceeding without context: %s", e)
        return ""

    def _call_old_endpoint(self, prompt: str, chat_history: list[dict], project_root: str | None = None) -> dict:
        """Call /chat-stream endpoint (old, streaming)."""
        final_prompt = ENHANCE_PROMPT_TEMPLATE.replace("{original_prompt}", prompt)
        language_guideline = (
            "Please respond in Chinese (Simplified Chinese). 请用中文回复。" if is_chinese_text(prompt) else ""
        )
        blob_names = self._get_blob_names(project_root)
        payload = self._build_old_payload(final_prompt, chat_history, language_guideline, blob_names)

        with httpx.Client(timeout=self.timeout) as client:
            resp = client.post(
                build_api_url(self.base_url, "/chat-stream"),
                headers=self._get_headers(),
                json=payload,
            )
            self._check_auth_error(resp.status_code)
            resp.raise_for_status()
            raw_text = self._parse_streaming_response(resp.text)
            enhanced = self._extract_enhanced_prompt(raw_text)
            enhanced = self._replace_tool_names(enhanced)
            return {"enhanced_prompt": enhanced}

    def _call_old_endpoint_raw(
        self, raw_prompt: str, chat_history: list[dict], project_root: str | None = None
    ) -> dict:
        """Call /chat-stream with pre-built prompt."""
        language_guideline = (
            "Please respond in Chinese (Simplified Chinese). 请用中文回复。" if is_chinese_text(raw_prompt) else ""
        )
        blob_names = self._get_blob_names(project_root)
        payload = self._build_old_payload(raw_prompt, chat_history, language_guideline, blob_names)

        with httpx.Client(timeout=self.timeout) as client:
            resp = client.post(
                build_api_url(self.base_url, "/chat-stream"),
                headers=self._get_headers(),
                json=payload,
            )
            self._check_auth_error(resp.status_code)
            resp.raise_for_status()
            raw_text = self._parse_streaming_response(resp.text)
            enhanced = self._extract_enhanced_prompt(raw_text)
            enhanced = self._replace_tool_names(enhanced)
            return {"enhanced_prompt": enhanced}

    def _parse_streaming_response(self, body: str) -> str:
        """Parse streaming response from /chat-stream endpoint."""
        combined = []
        for line in body.split("\n"):
            line = line.strip()
            if not line or line.startswith("data:"):
                line = line[5:].strip() if line.startswith("data:") else ""
            if not line or line == "[DONE]":
                continue
            try:
                data = json.loads(line)
                if text := data.get("text"):
                    combined.append(text)
            except json.JSONDecodeError:
                continue
        return "".join(combined) if combined else body

    def _extract_enhanced_prompt(self, text: str) -> str:
        """Extract enhanced prompt from XML-like response."""
        match = _ENHANCED_PROMPT_RE.search(text)
        if match:
            extracted = match.group(1).strip()
            if extracted:
                return extracted
        return text

    def _replace_tool_names(self, text: str) -> str:
        """Replace Augment-specific tool names with ace-tool names."""
        return text.replace("codebase-retrieval", "search_context").replace("codebase_retrieval", "search_context")

    def _post_json(
        self,
        url: str,
        payload: dict,
        *,
        headers: dict,
        provider: str = "API",
        timeout: httpx.Timeout | None = None,
    ) -> dict:
        with httpx.Client(timeout=timeout or self.timeout) as client:
            resp = client.post(url, headers=headers, json=payload)
            self._check_auth_error(resp.status_code, provider)
            resp.raise_for_status()
            return resp.json()

    def _enhanced_prompt_from_text(self, text: str, fallback: str) -> dict:
        enhanced = self._extract_enhanced_prompt(text) if text else fallback
        return {"enhanced_prompt": self._replace_tool_names(enhanced)}

    def _call_third_party_api(self, prompt: str, conversation_history: str, project_root: str | None = None) -> dict:
        """Call third-party API (Claude/OpenAI/Gemini/Codex)."""
        if not self.third_party_base_url or not self.third_party_token:
            raise ValueError(
                f"PROMPT_ENHANCER_BASE_URL and PROMPT_ENHANCER_TOKEN required for '{self.endpoint}' endpoint"
            )

        chat_history = parse_chat_history(conversation_history)
        model = self._get_third_party_model()
        injected_prompt = self._maybe_inject_search_context(self.endpoint, prompt, project_root)
        final_prompt = ENHANCE_PROMPT_TEMPLATE.replace("{original_prompt}", injected_prompt)
        language_hint = "\n\n请用中文回复。" if is_chinese_text(prompt) else ""
        full_prompt = f"{final_prompt}{language_hint}"

        return self._dispatch_third_party(full_prompt, chat_history, model)

    def _call_third_party_api_raw(
        self, raw_prompt: str, conversation_history: str, project_root: str | None = None
    ) -> dict:
        """Call third-party API with pre-built prompt."""
        if not self.third_party_base_url or not self.third_party_token:
            raise ValueError(
                f"PROMPT_ENHANCER_BASE_URL and PROMPT_ENHANCER_TOKEN required for '{self.endpoint}' endpoint"
            )

        chat_history = parse_chat_history(conversation_history)
        model = self._get_third_party_model()
        injected_prompt = self._maybe_inject_search_context(self.endpoint, raw_prompt, project_root)
        language_hint = "\n\n请用中文回复。" if is_chinese_text(raw_prompt) else ""
        full_prompt = f"{injected_prompt}{language_hint}"

        return self._dispatch_third_party(full_prompt, chat_history, model)

    def _dispatch_third_party(self, prompt: str, chat_history: list[dict], model: str) -> dict:
        """Dispatch to appropriate third-party API."""
        if self.endpoint == "claude":
            return self._call_claude_api(prompt, chat_history, model)
        elif self.endpoint == "openai":
            return self._call_openai_api(prompt, chat_history, model)
        elif self.endpoint == "gemini":
            return self._call_gemini_api(prompt, chat_history, model)
        elif self.endpoint == "codex":
            return self._call_codex_api(prompt, chat_history, model)
        return {"error": f"Unknown endpoint: {self.endpoint}"}

    @staticmethod
    def _extract_codex_output_text(api_response: dict) -> str:
        outputs = api_response.get("output", [])
        final = [o for o in outputs if o.get("type") == "message" and o.get("phase") == "final_answer"]
        candidates = final or [o for o in outputs if o.get("type") == "message"]
        text_parts, refusal_parts = [], []
        for msg in candidates:
            for part in msg.get("content") or []:
                t = part.get("type")
                if t == "output_text":
                    txt = (part.get("text") or "").strip()
                    if txt:
                        text_parts.append(txt)
                elif t == "refusal":
                    rf = (part.get("refusal") or "").strip()
                    if rf:
                        refusal_parts.append(rf)
        if text_parts:
            return "\n".join(text_parts)
        if refusal_parts:
            raise RuntimeError(f"Codex API refusal: {chr(10).join(refusal_parts)}")
        raise RuntimeError("Codex API returned no output_text content")

    def _call_codex_api(self, prompt: str, chat_history: list[dict], model: str) -> dict:
        input_items = []
        for msg in chat_history:
            input_items.append({"role": msg["role"], "content": msg["content"]})
        input_items.append({"role": "user", "content": prompt})

        payload = {"model": model, "input": input_items}
        if self.reasoning_effort:
            payload["reasoning"] = {"effort": self.reasoning_effort}
        url = build_api_url(self.third_party_base_url, "/v1/responses")

        data = self._post_json(url, payload, headers=self._get_headers(use_third_party=True), provider="Codex")
        text = self._extract_codex_output_text(data)
        return self._enhanced_prompt_from_text(text, prompt)

    def _call_claude_api(self, prompt: str, chat_history: list[dict], model: str) -> dict:
        """Call Claude API."""
        messages = chat_history + [{"role": "user", "content": prompt}]
        payload = {"model": model, "max_tokens": 4096, "messages": messages}
        # ponytail: adaptive only; legacy thinking-enabled/budget models unmapped —
        # unset effort if such an endpoint rejects adaptive.
        if self.reasoning_effort:
            if self.reasoning_effort == "none":
                payload["thinking"] = {"type": "disabled"}
            else:
                payload["thinking"] = {"type": "adaptive"}
                payload["output_config"] = {"effort": self.reasoning_effort}

        url = build_api_url(self.third_party_base_url, "/v1/messages")

        data = self._post_json(
            url,
            payload,
            headers={
                "Content-Type": "application/json",
                "x-api-key": self.third_party_token,
                "anthropic-version": "2023-06-01",
            },
            provider="Claude",
        )
        text = "".join(c.get("text", "") for c in data.get("content", []) if c.get("type") == "text")
        return self._enhanced_prompt_from_text(text, prompt)

    def _call_openai_api(self, prompt: str, chat_history: list[dict], model: str) -> dict:
        """Call OpenAI API."""
        messages = chat_history + [{"role": "user", "content": prompt}]
        payload = {"model": model, "messages": messages, "max_tokens": 4096}
        if self.reasoning_effort:
            payload["reasoning_effort"] = self.reasoning_effort

        url = build_api_url(self.third_party_base_url, "/v1/chat/completions")

        data = self._post_json(url, payload, headers=self._get_headers(use_third_party=True), provider="OpenAI")
        text = data.get("choices", [{}])[0].get("message", {}).get("content", "")
        return self._enhanced_prompt_from_text(text, prompt)

    def _call_gemini_api(self, prompt: str, chat_history: list[dict], model: str) -> dict:
        """Call Gemini API."""
        contents = []
        for msg in chat_history:
            role = "user" if msg["role"] == "user" else "model"
            contents.append({"role": role, "parts": [{"text": msg["content"]}]})
        contents.append({"role": "user", "parts": [{"text": prompt}]})

        payload = {"contents": contents, "generationConfig": {"maxOutputTokens": 4096}}
        # Gemini 3 has no "none" thinking level; "minimal" is its lowest.
        if self.reasoning_effort:
            level = "minimal" if self.reasoning_effort == "none" else self.reasoning_effort
            payload["generationConfig"]["thinkingConfig"] = {"thinkingLevel": level}

        url = build_api_url(self.third_party_base_url, f"/v1beta/models/{model}:generateContent")

        data = self._post_json(
            url,
            payload,
            headers={"Content-Type": "application/json", "x-goog-api-key": self.third_party_token},
            provider="Gemini",
        )
        text = data.get("candidates", [{}])[0].get("content", {}).get("parts", [{}])[0].get("text", "")
        return self._enhanced_prompt_from_text(text, prompt)

    def _check_auth_error(self, status: int, provider: str = "API"):
        if status == 401:
            raise httpx.HTTPStatusError(
                f"{provider} token invalid or expired",
                request=None,
                response=httpx.Response(status),
            )
        if status == 403:
            raise httpx.HTTPStatusError(
                f"{provider} access denied, token may be disabled",
                request=None,
                response=httpx.Response(status),
            )

    def _remote_search(self, project_root: str, query: str) -> dict:
        """Search via ACE codebase-retrieval API."""
        indexer = Indexer(project_root, self.base_url, self.token)
        blob_names = indexer.get_blob_names()

        url = build_api_url(self.base_url, "/agents/codebase-retrieval")
        payload = {
            "information_request": query,
            "blobs": {"checkpoint_id": None, "added_blobs": blob_names, "deleted_blobs": []},
            "dialog": [],
            "max_output_length": 0,
            "disable_codebase_retrieval": False,
            "enable_commit_retrieval": False,
        }
        timeout = httpx.Timeout(RETRIEVAL_TIMEOUT, connect=15.0)

        try:
            data = self._post_json(url, payload, headers=self._get_headers(), timeout=timeout)
        except httpx.HTTPStatusError as e:
            # Server lost blobs the local index marked as uploaded: rebuild, re-upload, retry once.
            if e.response.status_code != 400 or "unknown blob" not in e.response.text.lower():
                raise
            log.warning("Server reports unknown blobs; rebuilding index and retrying once")
            blob_names = indexer.force_rebuild()
            payload["blobs"]["added_blobs"] = blob_names
            data = self._post_json(url, payload, headers=self._get_headers(), timeout=timeout)

        return {
            "results": data.get("formatted_retrieval", ""),
            "query": query,
            "mode": "remote",
            "blob_count": len(blob_names),
        }

    def _local_search(self, project_root: str, query: str) -> dict:
        """Fallback local search using keyword matching."""
        results = []
        root = Path(project_root)
        keywords = [w.lower() for w in re.findall(r"\w+", query.lower()) if len(w) > 2]

        for file_path in root.rglob("*"):
            if not file_path.is_file():
                continue
            if file_path.suffix.lower() not in TEXT_EXTENSIONS:
                continue
            if any(p in EXCLUDE_PATTERNS for p in file_path.parts):
                continue
            if any(part.startswith(".") for part in file_path.parts[len(root.parts) :]):
                continue

            try:
                content = detect_and_read(file_path, ENCODING_CHAIN)
                if content is None:
                    continue
                content = content.lower()
                score = sum(1 for kw in keywords if kw in content)
                if score > 0:
                    results.append({"file": str(file_path.relative_to(root)), "score": score})
            except Exception:
                continue

        results.sort(key=lambda x: x["score"], reverse=True)
        return {"results": results[:10], "query": query, "mode": "local_fallback"}

    def get_config(self) -> dict:
        """Get current configuration."""
        return {
            "base_url": self.base_url or "(not configured)",
            "endpoint": self.endpoint,
            "endpoint_effective": self.endpoint,
            "endpoint_env_ready": bool(self.third_party_base_url and self.third_party_token)
            if self._is_third_party()
            else bool(self.base_url and self.token),
            "token_configured": bool(self.token),
            "third_party_configured": bool(self.third_party_base_url and self.third_party_token),
            "auth_source": self.auth_source,
            "search_context_injection": self._should_include_search_context(),
        }

"""Code indexer: scan, hash, chunk, upload to ACE batch-upload API."""

import fnmatch
import gzip
import hashlib
import json
import logging
import os
import shutil
import time
from dataclasses import asdict, dataclass, field
from pathlib import Path

import httpx

try:
    from .templates import (
        BINARY_EXTENSIONS,
        ENCODING_CHAIN,
        EXCLUDE_PATTERNS,
        INDEX_DIR,
        INDEX_FILE,
        MAX_BATCH_SIZE,
        MAX_BLOB_SIZE,
        MAX_LINES_PER_BLOB,
        TEXT_EXTENSIONS,
        UPLOAD_BATCH_COUNT,
        USER_AGENT,
    )
    from .utils import build_api_url, detect_and_read, get_session_id, sanitize_content
except ImportError:
    from templates import (
        BINARY_EXTENSIONS,
        ENCODING_CHAIN,
        EXCLUDE_PATTERNS,
        INDEX_DIR,
        INDEX_FILE,
        MAX_BATCH_SIZE,
        MAX_BLOB_SIZE,
        MAX_LINES_PER_BLOB,
        TEXT_EXTENSIONS,
        UPLOAD_BATCH_COUNT,
        USER_AGENT,
    )
    from utils import build_api_url, detect_and_read, get_session_id, sanitize_content

log = logging.getLogger(__name__)


class IndexRebuildError(ValueError):
    """Forced re-upload failed. Subclasses ValueError so tenacity-wrapped callers do not retry the full re-upload."""


@dataclass
class BlobEntry:
    path: str
    blob_name: str
    mtime: float
    size: int


@dataclass
class ProjectIndex:
    entries: dict[str, BlobEntry] = field(default_factory=dict)
    last_indexed: float = 0.0


class Indexer:
    def __init__(self, project_root: str, base_url: str, token: str):
        self.root = self._resolve_root(project_root)
        self.base_url = base_url.rstrip("/")
        self.token = token
        self.index_path = self.root / INDEX_DIR / INDEX_FILE
        self._index: ProjectIndex | None = None
        self._gitignore_patterns: list[str] = []
        self._child_cache_dirs: set[Path] = set()
        self._load_ignore_patterns()

    @staticmethod
    def _resolve_root(project_root: str) -> Path:
        """Use the nearest ancestor with an existing index cache as the effective root."""
        requested = Path(project_root).resolve()
        if (requested / INDEX_DIR / INDEX_FILE).is_file():
            return requested
        home = Path.home().resolve()
        for parent in requested.parents:
            # Home and filesystem-root caches are too broad to inherit implicitly.
            if parent == home or parent.parent == parent:
                break
            if (parent / INDEX_DIR / INDEX_FILE).is_file():
                log.info("Inheriting index cache from ancestor: %s", parent)
                return parent
        return requested

    def get_blob_names(self) -> list[str]:
        """Main entry: load/build index, upload pending, return blob_names."""
        self._load_index()
        old_entries = dict(self._index.entries)
        changed = self._scan_and_update()
        if changed:
            if self._upload_pending():
                if self._save_index():
                    self._absorb_child_caches()
            else:
                self._index.entries = old_entries
        elif self.index_path.is_file():
            self._absorb_child_caches()
        return list(self._index.entries.keys())

    def force_rebuild(self) -> list[str]:
        """Discard local index state, re-upload every blob (server lost our data)."""
        self._index = ProjectIndex()
        self._scan_and_update()
        if not self._upload_pending():
            raise IndexRebuildError("Failed to re-upload blobs after index rebuild")
        if self._save_index():
            self._absorb_child_caches()
        return list(self._index.entries.keys())

    def _absorb_child_caches(self):
        """Delete child .ace-tool caches superseded by this root's index."""
        own = self.index_path.parent
        for cache_dir in self._child_cache_dirs:
            if cache_dir == own or cache_dir.is_symlink():
                continue
            if not self._covered_by_scan(cache_dir.parent):
                continue
            try:
                cache_dir.resolve().relative_to(self.root)
                if not (cache_dir / INDEX_FILE).is_file():
                    continue
                shutil.rmtree(cache_dir)
                log.info("Absorbed child cache: %s", cache_dir)
            except (OSError, ValueError) as e:
                log.warning("Failed to remove child cache %s: %s", cache_dir, e)

    def _covered_by_scan(self, owner: Path) -> bool:
        """True if the cache owner's subtree is included in this root's scan."""
        rel_parts = owner.relative_to(self.root).parts
        if any(p in EXCLUDE_PATTERNS for p in rel_parts):
            return False
        if any(part.startswith(".") for part in rel_parts):
            return False
        return not self._is_gitignored(owner)

    # --- Index persistence ---

    def _load_index(self):
        if self._index is not None:
            return
        if self.index_path.exists():
            try:
                with gzip.open(self.index_path, "rt", encoding="utf-8") as f:
                    data = json.load(f)
                entries = {k: BlobEntry(**v) for k, v in data.get("entries", {}).items()}
                self._index = ProjectIndex(entries=entries, last_indexed=data.get("last_indexed", 0.0))
                return
            except Exception as e:
                log.warning("Failed to load index, rebuilding: %s", e)
        self._index = ProjectIndex()

    def _save_index(self) -> bool:
        self._index.last_indexed = time.time()
        # Per-process tmp: concurrent same-root writers each publish a fully
        # written file, so a torn index can never be renamed into place.
        tmp = self.index_path.parent / f"{INDEX_FILE}.{os.getpid()}.tmp"
        data = {
            "entries": {k: asdict(v) for k, v in self._index.entries.items()},
            "last_indexed": self._index.last_indexed,
        }
        try:
            self.index_path.parent.mkdir(parents=True, exist_ok=True)
            with gzip.open(tmp, "wt", encoding="utf-8") as f:
                json.dump(data, f)
            for attempt in range(3):
                try:
                    tmp.replace(self.index_path)
                    break
                except PermissionError:
                    # Windows refuses to replace a target a concurrent reader
                    # briefly holds open; genuine ACL errors still propagate.
                    if attempt == 2:
                        raise
                    time.sleep(0.05 * (2**attempt))
        except FileNotFoundError as e:
            # A concurrent ancestor index absorbed this cache dir mid-write.
            # Abandon persistence; the next run here inherits the ancestor root.
            log.warning("Index dir vanished during save (absorbed by ancestor?): %s", e)
            return False
        return True

    # --- Scanning ---

    def _scan_and_update(self) -> bool:
        self._child_cache_dirs = set()
        current_files: dict[str, Path] = {}
        for fp in self._walk_files():
            rel = fp.relative_to(self.root).as_posix()
            current_files[rel] = fp

        old_entries = self._index.entries
        new_entries: dict[str, BlobEntry] = {}
        pending_blobs: list[dict] = []
        changed = False

        # Group old entries by base path (strip chunk suffix) for cache hit
        old_by_path: dict[str, list[BlobEntry]] = {}
        for entry in old_entries.values():
            base = entry.path.split("#chunk", 1)[0]
            old_by_path.setdefault(base, []).append(entry)

        for rel, fp in current_files.items():
            try:
                stat = fp.stat()
            except OSError:
                continue
            mtime, size = stat.st_mtime, stat.st_size

            cached_list = old_by_path.get(rel)
            if cached_list and cached_list[0].mtime == mtime and cached_list[0].size == size:
                for cached in cached_list:
                    new_entries[cached.blob_name] = cached
                continue

            blobs = self._process_file(fp, rel)
            if not blobs:
                continue
            changed = True
            for blob_name, path_label, content in blobs:
                new_entries[blob_name] = BlobEntry(path=path_label, blob_name=blob_name, mtime=mtime, size=size)
                if blob_name not in old_entries:
                    pending_blobs.append({"path": path_label, "content": content, "blob_name": blob_name})

        if set(new_entries.keys()) != set(old_entries.keys()):
            changed = True

        self._index.entries = new_entries
        self._pending = pending_blobs
        return changed

    def _walk_files(self):
        for fp in self.root.rglob("*"):
            try:
                if not fp.is_file():
                    continue
            except OSError:
                continue
            rel_parts = fp.relative_to(self.root).parts
            if fp.name == INDEX_FILE and fp.parent.name == INDEX_DIR:
                self._child_cache_dirs.add(fp.parent)
            if any(p in EXCLUDE_PATTERNS for p in rel_parts):
                continue
            if any(part.startswith(".") and part != "." for part in rel_parts):
                continue
            if fp.suffix.lower() in BINARY_EXTENSIONS:
                continue
            if fp.suffix.lower() not in TEXT_EXTENSIONS:
                continue
            try:
                if fp.stat().st_size > MAX_BLOB_SIZE:
                    continue
            except OSError:
                continue
            if self._is_gitignored(fp):
                continue
            yield fp

    def _load_ignore_patterns(self):
        seen = set()
        for filename in (".gitignore", ".aceignore"):
            path = self.root / filename
            if path.exists():
                try:
                    for line in path.read_text(encoding="utf-8", errors="ignore").splitlines():
                        line = line.strip()
                        if line and not line.startswith("#") and line not in seen:
                            seen.add(line)
                            self._gitignore_patterns.append(line)
                except Exception:
                    pass

    def _is_gitignored(self, fp: Path) -> bool:
        rel = fp.relative_to(self.root).as_posix()
        name = fp.name
        for pat in self._gitignore_patterns:
            if fnmatch.fnmatch(name, pat) or fnmatch.fnmatch(rel, pat):
                return True
            clean = pat.rstrip("/")
            if clean in rel.split("/") or rel.startswith(clean + "/"):
                return True
        return False

    # --- File processing ---

    def _process_file(self, fp: Path, rel: str) -> list[tuple[str, str, str]]:
        """Read, sanitize, chunk, hash. Returns [(blob_name, path_label, content), ...]"""
        content = detect_and_read(fp, ENCODING_CHAIN)
        if content is None:
            return []
        content = sanitize_content(content)
        if not content.strip():
            return []

        lines = content.split("\n")
        if len(lines) <= MAX_LINES_PER_BLOB:
            blob_name = self._hash_blob(rel, content)
            return [(blob_name, rel, content)]

        # Chunk large files
        chunks = []
        total_chunks = (len(lines) + MAX_LINES_PER_BLOB - 1) // MAX_LINES_PER_BLOB
        for i in range(total_chunks):
            start = i * MAX_LINES_PER_BLOB
            end = min(start + MAX_LINES_PER_BLOB, len(lines))
            chunk_content = "\n".join(lines[start:end])
            chunk_label = f"{rel}#chunk{i + 1}of{total_chunks}"
            blob_name = self._hash_blob(chunk_label, chunk_content)
            chunks.append((blob_name, chunk_label, chunk_content))
        return chunks

    @staticmethod
    def _hash_blob(path: str, content: str) -> str:
        h = hashlib.sha256()
        h.update(path.encode("utf-8"))
        h.update(content.encode("utf-8"))
        return h.hexdigest()

    # --- Upload ---

    def _upload_pending(self) -> bool:
        if not self._pending or not self.base_url:
            return True

        batches = self._make_batches(self._pending)
        headers = {
            "Content-Type": "application/json",
            "User-Agent": USER_AGENT,
            "Authorization": f"Bearer {self.token}",
            "x-request-session-id": get_session_id(),
        }

        all_ok = True
        for batch in batches:
            payload = {"blobs": [{"path": b["path"], "content": b["content"]} for b in batch]}
            if not self._upload_batch_with_retry(headers, payload):
                all_ok = False

        self._pending = []
        return all_ok

    def _make_batches(self, blobs: list[dict]) -> list[list[dict]]:
        batches = []
        current_batch = []
        current_size = 0
        for b in blobs:
            item_size = len(b["content"].encode("utf-8"))
            if current_batch and (
                len(current_batch) >= UPLOAD_BATCH_COUNT or current_size + item_size > MAX_BATCH_SIZE
            ):
                batches.append(current_batch)
                current_batch = []
                current_size = 0
            current_batch.append(b)
            current_size += item_size
        if current_batch:
            batches.append(current_batch)
        return batches

    def _upload_batch_with_retry(self, headers: dict, payload: dict, max_retries: int = 3) -> bool:
        url = build_api_url(self.base_url, "/batch-upload")
        for attempt in range(max_retries):
            try:
                with httpx.Client(timeout=httpx.Timeout(60.0, connect=15.0)) as client:
                    resp = client.post(url, headers=headers, json=payload)
                    if resp.status_code in (401, 403):
                        log.error("Auth failed (%d) uploading blobs, aborting", resp.status_code)
                        return False
                    if resp.status_code == 429:
                        retry_after = int(resp.headers.get("Retry-After", "5"))
                        log.warning("Rate limited, waiting %ds", retry_after)
                        time.sleep(retry_after)
                        continue
                    if resp.status_code >= 500:
                        wait = 2**attempt
                        log.warning("Server error %d, retrying in %ds", resp.status_code, wait)
                        time.sleep(wait)
                        continue
                    resp.raise_for_status()
                    return True
            except httpx.TransportError as e:
                if attempt < max_retries - 1:
                    time.sleep(2**attempt)
                    log.warning("Transport error, retrying: %s", e)
                else:
                    log.error("Upload failed after retries: %s", e)
                    return False
        return False

"""Prompt templates and constants for ACE-Tool CLI."""

USER_AGENT = "augment.cli/0.29.0"
DEFAULT_MODEL = "claude-sonnet-4-5"

# Default models for third-party APIs
DEFAULT_CLAUDE_MODEL = "sonnet-4-6-20250929"
DEFAULT_OPENAI_MODEL = "gpt-5.4"
DEFAULT_CODEX_MODEL = "gpt-5.4"
DEFAULT_GEMINI_MODEL = "gemini-3-flash-preview"

# Environment variable names
ENV_ENHANCER_ENDPOINT = "PROMPT_ENHANCER_ENDPOINT"
ENV_ENHANCER_ENDPOINT_LEGACY = "ACE_ENHANCER_ENDPOINT"
ENV_ENHANCER_INCLUDE_SEARCH_CONTEXT = "PROMPT_ENHANCER_INCLUDE_SEARCH_CONTEXT"
ENV_ENHANCER_REASONING_EFFORT = "PROMPT_ENHANCER_REASONING_EFFORT"

# Search context injection
SEARCH_CONTEXT_CHAR_LIMIT = 12000
NO_RELEVANT_CODE_CONTEXT = "No relevant code context found for your query."

ENHANCE_PROMPT_TEMPLATE = """⚠️ NO TOOLS ALLOWED ⚠️

Here is an instruction that I'd like to give you, but it needs to be improved. Rewrite and enhance this instruction to make it clearer, more specific, less ambiguous, and correct any mistakes. Do not use any tools: reply immediately with your answer, even if you're not sure. Consider the context of our conversation history when enhancing the prompt. If there is code in triple backticks (```) consider whether it is a code sample and should remain unchanged.Reply with the following format:

### BEGIN RESPONSE ###
Here is an enhanced version of the original instruction that is more specific and clear:
<augment-enhanced-prompt>enhanced prompt goes here</augment-enhanced-prompt>

### END RESPONSE ###

Here is my original instruction:

{original_prompt}"""

ITERATIVE_ENHANCE_TEMPLATE = """⚠️ NO TOOLS ALLOWED ⚠️

You are performing an ITERATIVE ENHANCEMENT on an already-enhanced prompt. The user has reviewed and possibly edited the previous enhancement. Your task is to further refine and optimize while PRESERVING the user's modifications and intent.

**Context:**
- Original prompt: {original_prompt}
- Previous enhancement: {previous_enhanced}
- Current version (user may have edited): {current_prompt}

**Instructions:**
1. Identify what the user changed from the previous enhancement (their edits reflect their intent)
2. PRESERVE the user's modifications - do not revert their changes
3. Further optimize clarity, specificity, and correctness
4. If the user made no changes, provide alternative improvements or deeper refinement
5. Do not use any tools: reply immediately

Reply with the following format:

### BEGIN RESPONSE ###
<augment-enhanced-prompt>iteratively enhanced prompt goes here</augment-enhanced-prompt>
### END RESPONSE ###"""

TEXT_EXTENSIONS = {
    ".py",
    ".js",
    ".ts",
    ".jsx",
    ".tsx",
    ".mjs",
    ".cjs",
    ".java",
    ".go",
    ".rs",
    ".cpp",
    ".c",
    ".cc",
    ".h",
    ".hpp",
    ".hxx",
    ".cs",
    ".rb",
    ".php",
    ".swift",
    ".kt",
    ".kts",
    ".scala",
    ".clj",
    ".cljs",
    ".lua",
    ".dart",
    ".m",
    ".mm",
    ".pl",
    ".pm",
    ".r",
    ".R",
    ".jl",
    ".ex",
    ".exs",
    ".erl",
    ".hs",
    ".zig",
    ".v",
    ".nim",
    ".f90",
    ".f95",
    ".groovy",
    ".gradle",
    ".sol",
    ".move",
    ".md",
    ".mdx",
    ".txt",
    ".json",
    ".jsonc",
    ".json5",
    ".yaml",
    ".yml",
    ".toml",
    ".xml",
    ".ini",
    ".conf",
    ".cfg",
    ".properties",
    ".html",
    ".htm",
    ".css",
    ".scss",
    ".sass",
    ".less",
    ".styl",
    ".vue",
    ".svelte",
    ".astro",
    ".ejs",
    ".hbs",
    ".pug",
    ".jade",
    ".jinja",
    ".jinja2",
    ".erb",
    ".sql",
    ".sh",
    ".bash",
    ".zsh",
    ".fish",
    ".ps1",
    ".bat",
    ".cmd",
    ".graphql",
    ".gql",
    ".proto",
    ".prisma",
}

EXCLUDE_PATTERNS = {
    ".venv",
    "venv",
    ".env",
    "env",
    "node_modules",
    "vendor",
    ".pnpm",
    ".yarn",
    "bower_components",
    ".git",
    ".svn",
    ".hg",
    "__pycache__",
    ".pytest_cache",
    ".mypy_cache",
    ".tox",
    ".ruff_cache",
    "dist",
    "build",
    "target",
    "out",
    "bin",
    "obj",
    ".next",
    ".nuxt",
    ".output",
    ".vercel",
    ".netlify",
    ".turbo",
    ".parcel-cache",
    ".cache",
    ".temp",
    ".tmp",
    "coverage",
    ".nyc_output",
    "htmlcov",
    ".idea",
    ".vscode",
    ".vs",
    ".ace-tool",
}

# Indexing
MAX_BLOB_SIZE = 128 * 1024  # 128KB
MAX_BATCH_SIZE = 1 * 1024 * 1024  # 1MB
MAX_LINES_PER_BLOB = 800
UPLOAD_BATCH_COUNT = 30
RETRIEVAL_TIMEOUT = 60.0
INDEX_DIR = ".ace-tool"
INDEX_FILE = "index.json.gz"
ENCODING_CHAIN = ["utf-8", "gbk", "gb18030", "cp1252"]

BINARY_EXTENSIONS = {
    ".png",
    ".jpg",
    ".jpeg",
    ".gif",
    ".bmp",
    ".ico",
    ".svg",
    ".webp",
    ".mp3",
    ".mp4",
    ".avi",
    ".mov",
    ".wav",
    ".ogg",
    ".flv",
    ".zip",
    ".tar",
    ".gz",
    ".bz2",
    ".7z",
    ".rar",
    ".xz",
    ".exe",
    ".dll",
    ".so",
    ".dylib",
    ".o",
    ".a",
    ".lib",
    ".pdf",
    ".doc",
    ".docx",
    ".xls",
    ".xlsx",
    ".ppt",
    ".pptx",
    ".woff",
    ".woff2",
    ".ttf",
    ".otf",
    ".eot",
    ".pyc",
    ".pyo",
    ".class",
    ".jar",
    ".db",
    ".sqlite",
    ".sqlite3",
    ".bin",
    ".dat",
    ".pak",
    ".bundle",
}

"""Utility functions for ACE-Tool CLI."""

import os
import re
import uuid
from pathlib import Path

_SESSION_ID: str | None = None

_VERSION_SUFFIX_RE = re.compile(r"/v\d[A-Za-z0-9_-]*$")
_VERSION_PREFIX_RE = re.compile(r"^/v\d[A-Za-z0-9_-]*(?=/|$)")


def _has_version_suffix(url: str) -> tuple[bool, int]:
    match = _VERSION_SUFFIX_RE.search(url.rstrip("/"))
    return (True, match.start()) if match else (False, -1)


def _strip_version_prefix(path: str) -> str:
    return _VERSION_PREFIX_RE.sub("", path, count=1)


def build_api_url(base_url: str, path: str) -> str:
    base_url = base_url.rstrip("/")
    if not path.startswith("/"):
        path = "/" + path
    has_ver, ver_idx = _has_version_suffix(base_url)
    if has_ver:
        stripped = _strip_version_prefix(path)
        return base_url + stripped
    return base_url + path


def load_env():
    """Load environment variables from the skill root .env file."""
    env_path = Path(__file__).resolve().parent.parent / ".env"
    if env_path.exists():
        with open(env_path) as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith("#") and "=" in line:
                    key, value = line.split("=", 1)
                    value = value.strip()
                    # Strip inline comments (unquoted # preceded by whitespace)
                    if value and value[0] not in ('"', "'"):
                        value = re.split(r"\s+#", value, maxsplit=1)[0].strip()
                    else:
                        value = value.strip("\"'")
                    os.environ.setdefault(key.strip(), value)


def get_session_id() -> str:
    """Get or create persistent session ID."""
    global _SESSION_ID
    if _SESSION_ID is None:
        _SESSION_ID = str(uuid.uuid4())
    return _SESSION_ID


def is_chinese_text(text: str) -> bool:
    """Detect if text is primarily Chinese."""
    chinese_chars = re.findall(r"[\u4e00-\u9fa5]", text)
    if not chinese_chars:
        return False
    if len(chinese_chars) >= 3:
        return True
    non_ws = len([c for c in text if not c.isspace()])
    return non_ws > 0 and len(chinese_chars) / non_ws >= 0.1


def parse_chat_history(conversation_history: str) -> list[dict]:
    """Parse conversation history into ChatMessage format."""
    messages = []
    current_role = None
    current_lines = []

    user_prefixes = ["User:", "用户:"]
    assistant_prefixes = ["AI:", "Assistant:", "助手:"]

    for line in conversation_history.split("\n"):
        trimmed = line.strip()
        if not trimmed:
            if current_role:
                current_lines.append("")
            continue

        role_found = None
        content = None
        for prefix in user_prefixes:
            if trimmed.startswith(prefix):
                role_found = "user"
                content = trimmed[len(prefix) :].strip()
                break
        if not role_found:
            for prefix in assistant_prefixes:
                if trimmed.startswith(prefix):
                    role_found = "assistant"
                    content = trimmed[len(prefix) :].strip()
                    break

        if role_found:
            if current_role:
                messages.append({"role": current_role, "content": "\n".join(current_lines)})
            current_role = role_found
            current_lines = [content]
        elif current_role:
            current_lines.append(line)

    if current_role:
        messages.append({"role": current_role, "content": "\n".join(current_lines)})

    return messages


def detect_and_read(file_path: Path, encoding_chain: list[str]) -> str | None:
    """Try multiple encodings to read a file. Returns None for binary/unreadable."""
    try:
        raw = file_path.read_bytes()
    except OSError:
        return None
    if b"\x00" in raw[:8192]:
        return None
    for enc in encoding_chain:
        try:
            return raw.decode(enc)
        except (UnicodeDecodeError, LookupError):
            continue
    return None


def sanitize_content(content: str) -> str:
    """Normalize line endings and remove null bytes."""
    return content.replace("\r\n", "\n").replace("\r", "\n").replace("\x00", "")


def _validate_session_data(data: dict) -> tuple[str | None, str | None]:
    """Validate and extract session data fields.

    Returns (tenant_url, access_token) or (None, None) if invalid.
    """
    if not isinstance(data, dict):
        return None, None
    access_token = data.get("accessToken", "")
    tenant_url = data.get("tenantURL", "")
    if isinstance(access_token, str) and access_token.strip() and isinstance(tenant_url, str) and tenant_url.strip():
        return tenant_url.rstrip("/"), access_token
    return None, None


def load_session_auth() -> tuple[str | None, str | None, str]:
    """Load authentication from session.json, AUGMENT_SESSION_AUTH, or legacy env vars.

    Returns:
        (base_url, token, source) where source is one of:
        - "session.json"
        - "AUGMENT_SESSION_AUTH"
        - "ACE_API_TOKEN"
        - "none"
    """
    import json
    import logging

    log = logging.getLogger(__name__)

    # Try session.json first
    session_path = Path.home() / ".augment" / "session.json"
    if session_path.exists():
        try:
            content = session_path.read_text(encoding="utf-8-sig")
            data = json.loads(content)
            tenant_url, access_token = _validate_session_data(data)
            if tenant_url and access_token:
                log.debug("Auth source: session.json")
                return tenant_url, access_token, "session.json"
        except (OSError, json.JSONDecodeError, UnicodeDecodeError) as e:
            log.debug("Failed to load session.json: %s", e)

    # Try AUGMENT_SESSION_AUTH env var
    env_session = os.getenv("AUGMENT_SESSION_AUTH", "")
    if env_session:
        try:
            data = json.loads(env_session)
            tenant_url, access_token = _validate_session_data(data)
            if tenant_url and access_token:
                log.debug("Auth source: AUGMENT_SESSION_AUTH")
                return tenant_url, access_token, "AUGMENT_SESSION_AUTH"
        except (json.JSONDecodeError, TypeError) as e:
            log.debug("Failed to parse AUGMENT_SESSION_AUTH: %s", e)

    # Fallback to legacy env vars
    legacy_url = os.getenv("ACE_API_URL", "").rstrip("/")
    legacy_token = os.getenv("ACE_API_TOKEN", "")
    if legacy_url or legacy_token:
        log.debug("Auth source: ACE_API_TOKEN (legacy)")
        return legacy_url or None, legacy_token or None, "ACE_API_TOKEN"

    log.debug("Auth source: none")
    return None, None, "none"

"""Web UI for interactive prompt enhancement."""

import contextlib
import json
import os
import queue
import socket
import sys
import threading
import time
import uuid
import webbrowser
from http.server import BaseHTTPRequestHandler, HTTPServer
from typing import TYPE_CHECKING
from urllib.parse import parse_qs, urlparse

if TYPE_CHECKING:
    try:
        from .client import AceToolClient
    except ImportError:
        from client import AceToolClient

_SESSIONS: dict = {}
_RESULT_QUEUE: queue.Queue | None = None

HTML_TEMPLATE = """<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Prompt Enhancer - ACE Tool</title>
<style>
*{margin:0;padding:0;box-sizing:border-box}
body{font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,sans-serif;background:#f5f5f5;min-height:100vh;padding:20px;display:flex;align-items:center;justify-content:center}
.container{background:#fff;border-radius:8px;box-shadow:0 2px 8px rgba(0,0,0,0.1);border:1px solid #e0e0e0;max-width:1000px;width:100%;overflow:hidden}
.header{background:#fff;color:#333;padding:30px;text-align:center;border-bottom:1px solid #e0e0e0}
.header h1{font-size:24px;font-weight:600;margin-bottom:8px;display:flex;align-items:center;justify-content:center;gap:10px}
.header p{font-size:14px;color:#666}
.countdown{margin-top:12px;padding:8px 16px;background:#f0f0f0;border-radius:6px;display:inline-block;font-size:13px;font-weight:500;color:#555}
.countdown.warning{background:#fff3cd;color:#856404}
.countdown.danger{background:#f8d7da;color:#721c24;animation:pulse 1s ease-in-out infinite}
@keyframes pulse{0%,100%{opacity:1}50%{opacity:0.7}}
.content{padding:30px}
.section{margin-bottom:25px}
.section-title{font-size:14px;font-weight:600;color:#333;margin-bottom:10px;text-transform:uppercase;letter-spacing:0.5px}
.editor-wrapper{position:relative}
textarea{width:100%;min-height:350px;padding:16px;border:2px solid #e0e0e0;border-radius:8px;font-family:'SF Mono',Monaco,Menlo,Consolas,monospace;font-size:14px;line-height:1.6;resize:vertical;background:#fafafa}
textarea:focus{outline:none;border-color:#333;background:#fff}
.char-count{position:absolute;bottom:12px;right:12px;background:rgba(255,255,255,0.9);padding:4px 10px;border-radius:12px;font-size:12px;color:#666;pointer-events:none;box-shadow:0 2px 8px rgba(0,0,0,0.1)}
.info-box{background:#f9f9f9;border-left:4px solid #333;padding:15px;border-radius:4px;margin-bottom:20px}
.info-box p{font-size:14px;color:#555;line-height:1.6}
.buttons{display:flex;gap:12px;justify-content:flex-end;margin-top:25px;flex-wrap:wrap}
button{padding:12px 28px;border:none;border-radius:8px;font-size:15px;font-weight:600;cursor:pointer;transition:all 0.3s;display:flex;align-items:center;gap:8px}
button:focus-visible{outline:2px solid #333;outline-offset:2px}
.send-btn{background:#333;color:#fff}
.send-btn:hover:not(:disabled){background:#000}
.send-btn:disabled{background:#ccc;cursor:not-allowed}
.cancel-btn{background:#fff;color:#666;border:2px solid #e0e0e0}
.cancel-btn:hover{background:#f5f5f5;border-color:#ccc}
.re-enhance-btn{background:#fff;color:#333;border:2px solid #333}
.re-enhance-btn:hover:not(:disabled){background:#f5f5f5}
.re-enhance-btn:disabled{background:#f5f5f5;color:#ccc;border-color:#e0e0e0;cursor:not-allowed}
.status{margin-top:20px;padding:15px;border-radius:8px;display:none;animation:slideIn 0.3s ease}
@keyframes slideIn{from{opacity:0;transform:translateY(-10px)}to{opacity:1;transform:translateY(0)}}
.status.success{background:#d4edda;color:#155724;border-left:4px solid #28a745;display:block}
.status.error{background:#f8d7da;color:#721c24;border-left:4px solid #dc3545;display:block}
.loading{display:none;text-align:center;padding:40px}
.loading.active{display:block}
.spinner{border:3px solid #f3f3f3;border-top:3px solid #333;border-radius:50%;width:40px;height:40px;animation:spin 1s linear infinite;margin:0 auto 15px}
@keyframes spin{0%{transform:rotate(0deg)}100%{transform:rotate(360deg)}}
.keyboard-hint{font-size:12px;color:#999;text-align:center;margin-top:15px}
.keyboard-hint kbd{background:#f5f5f5;border:1px solid #ddd;border-radius:4px;padding:2px 6px;font-family:monospace;font-size:11px}
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1><svg width="24" height="24" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" aria-hidden="true"><path d="M12 2L2 7l10 5 10-5-10-5z"/><path d="M2 17l10 5 10-5"/><path d="M2 12l10 5 10-5"/></svg>Prompt Enhancer</h1>
<p>Review and refine your enhanced prompt</p>
<div class="countdown" id="countdown" aria-live="polite">Loading...</div>
</div>
<div class="content">
<div class="loading" id="loading" role="status" aria-live="polite"><div class="spinner" aria-hidden="true"></div><p>Loading your enhanced prompt...</p></div>
<div id="mainContent" style="display:none">
<div class="info-box"><p><strong>Tip:</strong> AI has enhanced your prompt. You can edit it below, then click "Send Enhanced" to continue. Use "Regenerate" to get a completely new enhancement, or "Refine" to improve the current version while keeping your edits.</p></div>
<div class="section">
<div class="section-title">Enhanced Prompt</div>
<div class="editor-wrapper">
<textarea id="promptText" aria-label="Enhanced prompt" placeholder="Your enhanced prompt will appear here..." spellcheck="false"></textarea>
<div class="char-count" id="charCount">0 chars</div>
</div>
</div>
<div class="buttons">
<button class="cancel-btn" onclick="endConversation()">Cancel</button>
<button class="re-enhance-btn" id="regenerateBtn" onclick="regenerate()" title="Discard current and generate a new enhancement from scratch">Regenerate</button>
<button class="re-enhance-btn" id="refineBtn" onclick="refine()" title="Further improve current version while preserving your edits">Refine</button>
<button class="cancel-btn" onclick="useOriginal()">Use Original</button>
<button class="send-btn" id="sendBtn" onclick="sendPrompt()">Send Enhanced</button>
</div>
<div class="keyboard-hint">Shortcuts: <kbd>Ctrl</kbd>+<kbd>Enter</kbd> Send | <kbd>Esc</kbd> Cancel</div>
<div id="status" class="status" role="alert" aria-live="assertive"></div>
</div>
</div>
</div>
<script>
const urlParams=new URLSearchParams(window.location.search);
const sessionId=urlParams.get('session');
const promptText=document.getElementById('promptText');
const charCount=document.getElementById('charCount');
const loading=document.getElementById('loading');
const mainContent=document.getElementById('mainContent');
const countdownEl=document.getElementById('countdown');
let countdownInterval=null,sessionCreatedAt=null,sessionTimeoutMs=null;

function updateCharCount(){charCount.textContent=promptText.value.length+' chars'}
promptText.addEventListener('input',updateCharCount);

function formatTime(ms){const s=Math.floor(ms/1000),m=Math.floor(s/60);return m+':'+(s%60).toString().padStart(2,'0')}
function updateCountdown(){
if(!sessionCreatedAt||!sessionTimeoutMs)return;
const remaining=sessionTimeoutMs-(Date.now()-sessionCreatedAt);
if(remaining<=0){countdownEl.textContent='Timed out';countdownEl.className='countdown danger';clearInterval(countdownInterval);return}
countdownEl.className='countdown'+(remaining<60000?' danger':remaining<180000?' warning':'');
countdownEl.textContent='Remaining: '+formatTime(remaining);
}
function startCountdown(createdAt,timeoutMs){sessionCreatedAt=createdAt;sessionTimeoutMs=timeoutMs;updateCountdown();countdownInterval=setInterval(updateCountdown,1000)}

document.addEventListener('keydown',e=>{if((e.ctrlKey||e.metaKey)&&e.key==='Enter'){e.preventDefault();sendPrompt()}else if(e.key==='Escape'){e.preventDefault();endConversation()}});

if(!sessionId){loading.style.display='none';mainContent.style.display='block';showStatus('Error: No session ID','error')}
else{loading.classList.add('active');
fetch('/api/session?session='+encodeURIComponent(sessionId)).then(r=>r.json()).then(data=>{
if(data.error)throw new Error(data.error);
promptText.value=data.enhancedPrompt;updateCharCount();
loading.classList.remove('active');mainContent.style.display='block';promptText.focus();
if(data.createdAt&&data.timeoutMs)startCountdown(data.createdAt,data.timeoutMs);
}).catch(err=>{loading.classList.remove('active');mainContent.style.display='block';showStatus('Load failed: '+err.message,'error')})}

function regenerate(){
const btn=document.getElementById('regenerateBtn'),refineBtn=document.getElementById('refineBtn'),sendBtn=document.getElementById('sendBtn');
btn.disabled=refineBtn.disabled=sendBtn.disabled=true;btn.innerHTML='<div class="spinner" style="width:16px;height:16px;border-width:2px;margin:0"></div> Regenerating...';
fetch('/api/regenerate',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({sessionId})})
.then(r=>r.json()).then(data=>{if(data.error)throw new Error(data.error);promptText.value=data.enhancedPrompt;updateCharCount();showStatus('Regenerated!','success');btn.disabled=refineBtn.disabled=sendBtn.disabled=false;btn.innerHTML='Regenerate'})
.catch(err=>{showStatus('Failed: '+err.message,'error');btn.disabled=refineBtn.disabled=sendBtn.disabled=false;btn.innerHTML='Regenerate'})}

function refine(){
const content=promptText.value.trim();if(!content){showStatus('Please enter content','error');return}
const btn=document.getElementById('refineBtn'),regenerateBtn=document.getElementById('regenerateBtn'),sendBtn=document.getElementById('sendBtn');
btn.disabled=regenerateBtn.disabled=sendBtn.disabled=true;btn.innerHTML='<div class="spinner" style="width:16px;height:16px;border-width:2px;margin:0"></div> Refining...';
fetch('/api/refine',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({sessionId,currentPrompt:content})})
.then(r=>r.json()).then(data=>{if(data.error)throw new Error(data.error);promptText.value=data.enhancedPrompt;updateCharCount();showStatus('Refined!','success');btn.disabled=regenerateBtn.disabled=sendBtn.disabled=false;btn.innerHTML='Refine'})
.catch(err=>{showStatus('Failed: '+err.message,'error');btn.disabled=regenerateBtn.disabled=sendBtn.disabled=false;btn.innerHTML='Refine'})}

function sendPrompt(){
const content=promptText.value.trim();if(!content){showStatus('Please enter content','error');return}
const btn=document.getElementById('sendBtn'),regenerateBtn=document.getElementById('regenerateBtn'),refineBtn=document.getElementById('refineBtn');
btn.disabled=regenerateBtn.disabled=refineBtn.disabled=true;btn.innerHTML='<div class="spinner" style="width:16px;height:16px;border-width:2px;margin:0"></div> Sending...';
fetch('/api/submit',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({sessionId,content,action:'send'})})
.then(r=>r.json()).then(data=>{if(data.error)throw new Error(data.error);showStatus('Sent! Closing...','success');setTimeout(()=>window.close(),1500)})
.catch(err=>{showStatus('Failed: '+err.message,'error');btn.disabled=regenerateBtn.disabled=refineBtn.disabled=false;btn.innerHTML='Send Enhanced'})}

function useOriginal(){if(confirm('Use original prompt?')){
fetch('/api/submit',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({sessionId,content:'',action:'use_original'})})
.then(r=>r.json()).then(data=>{if(data.error)throw new Error(data.error);showStatus('Using original...','success');setTimeout(()=>window.close(),1000)})
.catch(err=>showStatus('Failed: '+err.message,'error'))}}

function endConversation(){if(confirm('Cancel enhancement?')){
fetch('/api/submit',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({sessionId,content:'',action:'cancel'})})
.then(r=>r.json()).then(data=>{if(data.error)throw new Error(data.error);showStatus('Cancelled','success');setTimeout(()=>window.close(),1000)})
.catch(err=>showStatus('Failed: '+err.message,'error'))}}

function showStatus(msg,type){const s=document.getElementById('status');s.textContent=msg;s.className='status '+type}
</script>
</body>
</html>"""


class EnhanceRequestHandler(BaseHTTPRequestHandler):
    client: "AceToolClient" = None
    original_prompt: str = ""
    conversation_history: str = ""

    def log_message(self, fmt, *args):
        pass

    def handle(self):
        with contextlib.suppress(ConnectionAbortedError, ConnectionResetError, BrokenPipeError):
            super().handle()

    def do_GET(self):
        parsed = urlparse(self.path)
        if parsed.path == "/enhance":
            self.send_response(200)
            self.send_header("Content-Type", "text/html; charset=utf-8")
            self.end_headers()
            self.wfile.write(HTML_TEMPLATE.encode("utf-8"))
        elif parsed.path == "/api/session":
            self._handle_get_session(parsed.query)
        else:
            self.send_error(404)

    def do_POST(self):
        parsed = urlparse(self.path)
        length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(length).decode("utf-8")

        if parsed.path == "/api/submit":
            self._handle_submit(body)
        elif parsed.path == "/api/regenerate":
            self._handle_regenerate(body)
        elif parsed.path == "/api/refine":
            self._handle_refine(body)
        else:
            self.send_error(404)

    def _send_json(self, data: dict, status: int = 200):
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        self.wfile.write(json.dumps(data).encode("utf-8"))

    def _handle_get_session(self, query: str):
        params = parse_qs(query)
        session_id = params.get("session", [None])[0]
        if not session_id or session_id not in _SESSIONS:
            self._send_json({"error": "Session not found"}, 404)
            return
        session = _SESSIONS[session_id]
        self._send_json(
            {
                "enhancedPrompt": session["enhanced"],
                "status": session["status"],
                "createdAt": session["created_at"],
                "timeoutMs": session["timeout_ms"],
            }
        )

    def _handle_submit(self, body: str):
        try:
            data = json.loads(body)
        except json.JSONDecodeError:
            self._send_json({"error": "Invalid JSON"}, 400)
            return

        session_id = data.get("sessionId")
        action = data.get("action", "send")
        content = data.get("content", "")

        if not session_id or session_id not in _SESSIONS:
            self._send_json({"error": "Session not found"}, 404)
            return

        session = _SESSIONS[session_id]
        if session["status"] != "pending":
            self._send_json({"error": "Session already completed"}, 400)
            return

        session["status"] = "completed"

        if action == "cancel":
            result = None
        elif action == "use_original":
            result = session["original"]
        else:
            result = content if content else session["enhanced"]

        if _RESULT_QUEUE:
            _RESULT_QUEUE.put(result)

        self._send_json({"success": True})

    def _handle_regenerate(self, body: str):
        """Regenerate: discard current and generate new enhancement from original prompt."""
        try:
            data = json.loads(body)
        except json.JSONDecodeError:
            self._send_json({"error": "Invalid JSON"}, 400)
            return

        session_id = data.get("sessionId")
        if not session_id or session_id not in _SESSIONS:
            self._send_json({"error": "Session not found"}, 404)
            return

        session = _SESSIONS[session_id]
        try:
            result = self.client.enhance_prompt(
                session["original"],
                self.conversation_history,
                session.get("project_root"),
            )
            new_enhanced = result.get("enhanced_prompt", session["original"])
            session["previous_enhanced"] = session["enhanced"]
            session["enhanced"] = new_enhanced
            session["regenerate_count"] = session.get("regenerate_count", 0) + 1
            self._send_json(
                {
                    "enhancedPrompt": new_enhanced,
                    "regenerateCount": session["regenerate_count"],
                }
            )
        except Exception as e:
            self._send_json({"error": str(e)}, 500)

    def _handle_refine(self, body: str):
        """Refine: iteratively improve current version while preserving user edits."""
        try:
            data = json.loads(body)
        except json.JSONDecodeError:
            self._send_json({"error": "Invalid JSON"}, 400)
            return

        session_id = data.get("sessionId")
        current_prompt = data.get("currentPrompt", "")

        if not session_id or session_id not in _SESSIONS:
            self._send_json({"error": "Session not found"}, 404)
            return

        session = _SESSIONS[session_id]
        try:
            previous_enhanced = session.get("previous_enhanced") or session["enhanced"]
            result = self.client.iterative_enhance(
                original_prompt=session["original"],
                previous_enhanced=previous_enhanced,
                current_prompt=current_prompt,
                conversation_history=self.conversation_history,
                project_root=session.get("project_root"),
            )
            new_enhanced = result.get("enhanced_prompt", current_prompt)
            session["previous_enhanced"] = session["enhanced"]
            session["enhanced"] = new_enhanced
            session["refine_count"] = session.get("refine_count", 0) + 1
            self._send_json(
                {
                    "enhancedPrompt": new_enhanced,
                    "refineCount": session["refine_count"],
                }
            )
        except Exception as e:
            self._send_json({"error": str(e)}, 500)


def run_interactive_enhance(
    client: "AceToolClient",
    prompt: str,
    history: str,
    port: int = 8765,
    auto_open_browser: bool = True,
    project_root: str | None = None,
) -> str | None:
    """Run interactive web-based prompt enhancement."""
    global _RESULT_QUEUE

    print("Enhancing prompt...", file=sys.stderr)
    result = client.enhance_prompt(prompt, history, project_root)

    if "error" in result:
        print(f"Enhancement error: {result['error']}", file=sys.stderr)
        return None

    enhanced = result.get("enhanced_prompt", prompt)

    session_id = str(uuid.uuid4())
    created_at = int(time.time() * 1000)
    timeout_ms = 5 * 60 * 1000

    _SESSIONS[session_id] = {
        "original": prompt,
        "enhanced": enhanced,
        "previous_enhanced": None,
        "regenerate_count": 0,
        "refine_count": 0,
        "status": "pending",
        "created_at": created_at,
        "timeout_ms": timeout_ms,
        "project_root": project_root,
    }

    _RESULT_QUEUE = queue.Queue()

    EnhanceRequestHandler.client = client
    EnhanceRequestHandler.original_prompt = prompt
    EnhanceRequestHandler.conversation_history = history

    for p in range(port, port + 100):
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.bind(("127.0.0.1", p))
                port = p
                break
        except OSError:
            continue

    server = HTTPServer(("127.0.0.1", port), EnhanceRequestHandler)
    server.timeout = 0.5

    shutdown_flag = threading.Event()
    server_ready = threading.Event()

    def serve():
        server_ready.set()
        while not shutdown_flag.is_set():
            server.handle_request()

    thread = threading.Thread(target=serve, daemon=True)
    thread.start()
    server_ready.wait(timeout=2)

    url = f"http://127.0.0.1:{port}/enhance?session={session_id}"
    print(f"Opening browser: {url}", file=sys.stderr)

    def open_browser():
        import subprocess

        if sys.platform == "win32":
            # Try multiple methods on Windows
            methods = [
                lambda: os.startfile(url),
                lambda: subprocess.run(["cmd", "/c", "start", "", url], check=True, shell=False),
                lambda: subprocess.run(["powershell", "-Command", f"Start-Process '{url}'"], check=True),
                lambda: webbrowser.open(url),
            ]
        elif sys.platform == "darwin":
            methods = [
                lambda: subprocess.run(["open", url], check=True),
                lambda: webbrowser.open(url),
            ]
        else:
            methods = [
                lambda: subprocess.run(["xdg-open", url], check=True),
                lambda: webbrowser.open(url),
            ]

        for method in methods:
            try:
                method()
                return True
            except Exception as e:
                print(f"Browser open method failed: {e}", file=sys.stderr)
                continue
        return False

    if auto_open_browser:
        if not open_browser():
            print(f"Could not open browser. Please open manually: {url}", file=sys.stderr)
    else:
        print(f"Browser auto-open disabled. Please open: {url}", file=sys.stderr)

    print("Waiting for user action...", file=sys.stderr)

    try:
        result = _RESULT_QUEUE.get(timeout=timeout_ms / 1000)
    except queue.Empty:
        result = None
        print("Timeout waiting for user action", file=sys.stderr)

    shutdown_flag.set()
    thread.join(timeout=1)
    server.server_close()
    del _SESSIONS[session_id]
    _RESULT_QUEUE = None

    return result

"""ACE-Tool CLI package."""

from .client import AceToolClient
from .indexer import Indexer
from .utils import detect_and_read, get_session_id, is_chinese_text, load_env, parse_chat_history, sanitize_content
from .web_ui import run_interactive_enhance

# Auto-load environment variables on import. Client/indexer/web_ui only read env
# in methods, not at import time, so this can sit after the imports.
load_env()

__all__ = [
    "AceToolClient",
    "Indexer",
    "load_env",
    "get_session_id",
    "is_chinese_text",
    "parse_chat_history",
    "detect_and_read",
    "sanitize_content",
    "run_interactive_enhance",
]

"""Entry point for running as module: python -m scripts"""

from .ace_cli import main

if __name__ == "__main__":
    main()

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))

"""Tests for authentication upgrade: load_session_auth() function."""

import json
import os
from unittest.mock import patch

try:
    from scripts import utils
    from scripts.utils import load_env, load_session_auth
except ImportError:
    import utils
    from utils import load_env, load_session_auth


class TestLoadSessionAuth:
    """Test load_session_auth() authentication fallback chain."""

    def test_session_json_valid(self, tmp_path):
        """Test loading from valid session.json file."""
        session_file = tmp_path / ".augment" / "session.json"
        session_file.parent.mkdir(parents=True)
        session_data = {"accessToken": "token_from_file", "tenantURL": "https://file.example.com/", "scopes": ["email"]}
        session_file.write_text(json.dumps(session_data), encoding="utf-8")

        with patch("pathlib.Path.home", return_value=tmp_path):
            base_url, token, source = load_session_auth()

        assert base_url == "https://file.example.com"
        assert token == "token_from_file"
        assert source == "session.json"

    def test_augment_session_auth_env(self):
        """Test loading from AUGMENT_SESSION_AUTH environment variable."""
        env_data = {"accessToken": "token_from_env", "tenantURL": "https://env.example.com/"}
        with (
            patch.dict(os.environ, {"AUGMENT_SESSION_AUTH": json.dumps(env_data)}, clear=False),
            patch("pathlib.Path.exists", return_value=False),
        ):
            base_url, token, source = load_session_auth()

        assert base_url == "https://env.example.com"
        assert token == "token_from_env"
        assert source == "AUGMENT_SESSION_AUTH"

    def test_legacy_env_fallback(self, tmp_path):
        """Test fallback to legacy ACE_API_URL and ACE_API_TOKEN."""
        env_vars = {
            "ACE_API_URL": "https://legacy.example.com/",
            "ACE_API_TOKEN": "legacy_token",
            "AUGMENT_SESSION_AUTH": "",
        }
        with patch("pathlib.Path.home", return_value=tmp_path), patch.dict(os.environ, env_vars, clear=False):
            base_url, token, source = load_session_auth()

        assert base_url == "https://legacy.example.com"
        assert token == "legacy_token"
        assert source == "ACE_API_TOKEN"

    def test_all_missing_returns_none(self, tmp_path):
        """Test returns (None, None, 'none') when all sources missing."""
        env_vars = {"ACE_API_URL": "", "ACE_API_TOKEN": "", "AUGMENT_SESSION_AUTH": ""}
        with patch("pathlib.Path.home", return_value=tmp_path), patch.dict(os.environ, env_vars, clear=False):
            base_url, token, source = load_session_auth()

        assert base_url is None
        assert token is None
        assert source == "none"

    def test_session_json_file_not_found(self, tmp_path):
        """Test graceful handling when session.json doesn't exist."""
        env_data = {"accessToken": "env_token", "tenantURL": "https://env.example.com/"}
        with (
            patch("pathlib.Path.home", return_value=tmp_path),
            patch.dict(os.environ, {"AUGMENT_SESSION_AUTH": json.dumps(env_data)}, clear=False),
        ):
            base_url, token, source = load_session_auth()

        assert base_url == "https://env.example.com"
        assert token == "env_token"
        assert source == "AUGMENT_SESSION_AUTH"

    def test_session_json_invalid_json(self, tmp_path):
        """Test fallback when session.json contains invalid JSON."""
        session_file = tmp_path / ".augment" / "session.json"
        session_file.parent.mkdir(parents=True)
        session_file.write_text("not valid json{", encoding="utf-8")

        env_data = {"accessToken": "fallback_token", "tenantURL": "https://fallback.example.com/"}
        with (
            patch("pathlib.Path.home", return_value=tmp_path),
            patch.dict(os.environ, {"AUGMENT_SESSION_AUTH": json.dumps(env_data)}, clear=False),
        ):
            base_url, token, source = load_session_auth()

        assert base_url == "https://fallback.example.com"
        assert token == "fallback_token"
        assert source == "AUGMENT_SESSION_AUTH"

    def test_session_json_missing_fields(self, tmp_path):
        """Test fallback when session.json missing required fields."""
        session_file = tmp_path / ".augment" / "session.json"
        session_file.parent.mkdir(parents=True)
        session_data = {"scopes": ["email"]}  # Missing accessToken and tenantURL
        session_file.write_text(json.dumps(session_data), encoding="utf-8")

        env_vars = {"ACE_API_URL": "https://legacy.example.com/", "ACE_API_TOKEN": "legacy_token"}
        with patch("pathlib.Path.home", return_value=tmp_path), patch.dict(os.environ, env_vars, clear=True):
            base_url, token, source = load_session_auth()

        assert base_url == "https://legacy.example.com"
        assert token == "legacy_token"
        assert source == "ACE_API_TOKEN"

    def test_session_json_empty_strings(self, tmp_path):
        """Test rejection of empty string values."""
        session_file = tmp_path / ".augment" / "session.json"
        session_file.parent.mkdir(parents=True)
        session_data = {"accessToken": "", "tenantURL": "   "}
        session_file.write_text(json.dumps(session_data), encoding="utf-8")

        env_vars = {"ACE_API_URL": "https://legacy.example.com/", "ACE_API_TOKEN": "legacy_token"}
        with patch("pathlib.Path.home", return_value=tmp_path), patch.dict(os.environ, env_vars, clear=True):
            base_url, token, source = load_session_auth()

        assert base_url == "https://legacy.example.com"
        assert token == "legacy_token"
        assert source == "ACE_API_TOKEN"

    def test_session_json_with_bom(self, tmp_path):
        """Test UTF-8 BOM handling."""
        session_file = tmp_path / ".augment" / "session.json"
        session_file.parent.mkdir(parents=True)
        session_data = {"accessToken": "bom_token", "tenantURL": "https://bom.example.com/"}
        session_file.write_bytes(b"\xef\xbb\xbf" + json.dumps(session_data).encode("utf-8"))

        with patch("pathlib.Path.home", return_value=tmp_path):
            base_url, token, source = load_session_auth()

        assert base_url == "https://bom.example.com"
        assert token == "bom_token"
        assert source == "session.json"

    def test_url_trailing_slash_stripped(self, tmp_path):
        """Test that trailing slashes are removed from URLs."""
        session_file = tmp_path / ".augment" / "session.json"
        session_file.parent.mkdir(parents=True)
        session_data = {"accessToken": "token", "tenantURL": "https://example.com///"}
        session_file.write_text(json.dumps(session_data), encoding="utf-8")

        with patch("pathlib.Path.home", return_value=tmp_path):
            base_url, token, source = load_session_auth()

        assert base_url == "https://example.com"
        assert token == "token"
        assert source == "session.json"


class TestLoadEnv:
    """Test .env loading behavior."""

    def test_loads_only_skill_root_env(self, tmp_path, monkeypatch):
        """Test loading only the skill root .env file."""
        cwd_env = tmp_path / ".env"
        cwd_env.write_text("UNRELATED=value\nPROMPT_ENHANCER_ENDPOINT=wrong\n", encoding="utf-8")

        skill_root = tmp_path / "skill"
        scripts_dir = skill_root / "scripts"
        scripts_dir.mkdir(parents=True)
        skill_env = skill_root / ".env"
        skill_env.write_text("PROMPT_ENHANCER_ENDPOINT=openai\n", encoding="utf-8")

        monkeypatch.chdir(tmp_path)
        monkeypatch.setattr(utils, "__file__", str(scripts_dir / "utils.py"))
        monkeypatch.delenv("UNRELATED", raising=False)
        monkeypatch.delenv("PROMPT_ENHANCER_ENDPOINT", raising=False)

        load_env()

        assert "UNRELATED" not in os.environ
        assert os.environ["PROMPT_ENHANCER_ENDPOINT"] == "openai"

"""CLI command behavior: index auth resolution, enhance error exit codes."""

import json
import os
from types import SimpleNamespace
from unittest.mock import MagicMock, patch

import pytest

try:
    from scripts import ace_cli
except ImportError:
    import ace_cli


def _index_args(project_root):
    return SimpleNamespace(api_url=None, token=None, endpoint=None, project_root=str(project_root))


class TestIndexAuth:
    """Test cmd_index resolves auth via the full session chain, not legacy vars only."""

    def test_index_accepts_augment_session_auth(self, tmp_path):
        """Regression: index must work with AUGMENT_SESSION_AUTH (new session format)."""
        env_data = {"accessToken": "session_token", "tenantURL": "https://session.example.com/"}
        env_vars = {"AUGMENT_SESSION_AUTH": json.dumps(env_data), "ACE_API_URL": "", "ACE_API_TOKEN": ""}

        captured = {}

        class FakeIndexer:
            def __init__(self, project_root, base_url, token):
                captured["base_url"] = base_url
                captured["token"] = token
                self.root = project_root
                self._index = SimpleNamespace(last_indexed=0.0)

            def get_blob_names(self):
                return ["blob1", "blob2"]

        with (
            patch("pathlib.Path.home", return_value=tmp_path),
            patch.dict(os.environ, env_vars, clear=False),
            patch.object(ace_cli, "Indexer", FakeIndexer),
        ):
            ace_cli.cmd_index(_index_args(tmp_path))

        assert captured["base_url"] == "https://session.example.com"
        assert captured["token"] == "session_token"

    def test_index_errors_when_no_auth(self, tmp_path, capsys):
        """Index exits with a clear error when no auth source is configured."""
        env_vars = {"AUGMENT_SESSION_AUTH": "", "ACE_API_URL": "", "ACE_API_TOKEN": ""}

        with (
            patch("pathlib.Path.home", return_value=tmp_path),
            patch.dict(os.environ, env_vars, clear=False),
            pytest.raises(SystemExit) as exc,
        ):
            ace_cli.cmd_index(_index_args(tmp_path))

        assert exc.value.code == 1
        err = json.loads(capsys.readouterr().err)
        assert "authentication" in err["error"].lower()


class TestCmdEnhancePromptError:
    """cmd_enhance_prompt exits 1 when enhancement returns an error dict."""

    def test_error_exits_with_code_1(self):
        args = MagicMock()
        args.no_interactive = True
        args.prompt = "test"
        args.history = ""
        args.history_file = None
        args.project_root = None
        args.api_url = None
        args.token = None
        args.endpoint = "new"

        with (
            patch.dict(os.environ, {"ACE_API_URL": "", "ACE_API_TOKEN": ""}, clear=False),
            patch.object(ace_cli, "AceToolClient") as mock_client,
        ):
            mock_client.return_value.enhance_prompt.return_value = {"error": "test error"}
            with pytest.raises(SystemExit) as exc_info:
                ace_cli.cmd_enhance_prompt(args)
            assert exc_info.value.code == 1

"""Tests for AceToolClient authentication integration."""

import json
import os
from unittest.mock import patch

try:
    from scripts.client import AceToolClient
except ImportError:
    from client import AceToolClient


class TestClientAuthIntegration:
    """Test AceToolClient uses load_session_auth() correctly."""

    def test_constructor_params_override_all(self, tmp_path):
        """Test constructor parameters take highest priority."""
        session_file = tmp_path / ".augment" / "session.json"
        session_file.parent.mkdir(parents=True)
        session_data = {"accessToken": "file_token", "tenantURL": "https://file.example.com/"}
        session_file.write_text(json.dumps(session_data), encoding="utf-8")

        env_vars = {"ACE_API_URL": "https://env.example.com/", "ACE_API_TOKEN": "env_token"}

        with patch("pathlib.Path.home", return_value=tmp_path), patch.dict(os.environ, env_vars, clear=False):
            client = AceToolClient(base_url="https://constructor.example.com", token="constructor_token")

        assert client.base_url == "https://constructor.example.com"
        assert client.token == "constructor_token"
        assert client.auth_source == "constructor"

    def test_session_json_used_when_no_constructor_params(self, tmp_path):
        """Test session.json used when constructor params not provided."""
        session_file = tmp_path / ".augment" / "session.json"
        session_file.parent.mkdir(parents=True)
        session_data = {"accessToken": "session_token", "tenantURL": "https://session.example.com/"}
        session_file.write_text(json.dumps(session_data), encoding="utf-8")

        with patch("pathlib.Path.home", return_value=tmp_path):
            client = AceToolClient()

        assert client.base_url == "https://session.example.com"
        assert client.token == "session_token"
        assert client.auth_source == "session.json"

    def test_augment_session_auth_fallback(self, tmp_path):
        """Test AUGMENT_SESSION_AUTH used when session.json missing."""
        env_data = {"accessToken": "env_token", "tenantURL": "https://env.example.com/"}
        env_vars = {"AUGMENT_SESSION_AUTH": json.dumps(env_data)}

        with patch("pathlib.Path.home", return_value=tmp_path), patch.dict(os.environ, env_vars, clear=False):
            client = AceToolClient()

        assert client.base_url == "https://env.example.com"
        assert client.token == "env_token"
        assert client.auth_source == "AUGMENT_SESSION_AUTH"

    def test_legacy_env_vars_lowest_priority(self, tmp_path):
        """Test legacy ACE_API_* vars used as last resort."""
        env_vars = {
            "ACE_API_URL": "https://legacy.example.com/",
            "ACE_API_TOKEN": "legacy_token",
            "AUGMENT_SESSION_AUTH": "",
        }

        with patch("pathlib.Path.home", return_value=tmp_path), patch.dict(os.environ, env_vars, clear=False):
            client = AceToolClient()

        assert client.base_url == "https://legacy.example.com"
        assert client.token == "legacy_token"
        assert client.auth_source == "ACE_API_TOKEN"

    def test_partial_constructor_params(self, tmp_path):
        """Test partial constructor params override only those fields."""
        session_file = tmp_path / ".augment" / "session.json"
        session_file.parent.mkdir(parents=True)
        session_data = {"accessToken": "session_token", "tenantURL": "https://session.example.com/"}
        session_file.write_text(json.dumps(session_data), encoding="utf-8")

        with patch("pathlib.Path.home", return_value=tmp_path):
            client = AceToolClient(token="override_token")

        # Only token overridden, base_url from session.json
        assert client.base_url == "https://session.example.com"
        assert client.token == "override_token"
        assert client.auth_source == "constructor"

    def test_no_auth_configured(self, tmp_path):
        """Test client handles missing authentication gracefully."""
        env_vars = {"ACE_API_URL": "", "ACE_API_TOKEN": "", "AUGMENT_SESSION_AUTH": ""}

        with patch("pathlib.Path.home", return_value=tmp_path), patch.dict(os.environ, env_vars, clear=False):
            client = AceToolClient()

        assert client.base_url == ""
        assert client.token == ""
        assert client.auth_source == "none"

    def test_auth_source_in_get_config(self, tmp_path):
        """Test get_config() includes auth_source field."""
        session_file = tmp_path / ".augment" / "session.json"
        session_file.parent.mkdir(parents=True)
        session_data = {"accessToken": "token", "tenantURL": "https://example.com/"}
        session_file.write_text(json.dumps(session_data), encoding="utf-8")

        with patch("pathlib.Path.home", return_value=tmp_path):
            client = AceToolClient()
            config = client.get_config()

        assert "auth_source" in config
        assert config["auth_source"] == "session.json"

"""Endpoint routing, third-party detection, hard errors, and config output."""

import os
from unittest.mock import patch

import pytest
from client import AceToolClient


class TestEndpointResolution:
    """Resolution order: PROMPT_ENHANCER_ENDPOINT > ACE_ENHANCER_ENDPOINT > constructor > default."""

    def test_env_overrides_constructor(self):
        with patch.dict(
            os.environ,
            {
                "PROMPT_ENHANCER_ENDPOINT": "claude",
                "ACE_ENHANCER_ENDPOINT": "",
            },
            clear=False,
        ):
            c = AceToolClient(endpoint="openai")
            assert c.endpoint == "claude"

    def test_new_env_wins_over_legacy_and_constructor(self):
        with patch.dict(
            os.environ,
            {
                "PROMPT_ENHANCER_ENDPOINT": "claude",
                "ACE_ENHANCER_ENDPOINT": "openai",
            },
            clear=False,
        ):
            c = AceToolClient(endpoint="gemini")
            assert c.endpoint == "claude"

    def test_legacy_env_used_when_new_absent(self):
        with patch.dict(
            os.environ,
            {
                "PROMPT_ENHANCER_ENDPOINT": "",
                "ACE_ENHANCER_ENDPOINT": "gemini",
            },
            clear=False,
        ):
            c = AceToolClient()
            assert c.endpoint == "gemini"

    def test_constructor_arg_used_when_no_env(self):
        with patch.dict(
            os.environ,
            {
                "PROMPT_ENHANCER_ENDPOINT": "",
                "ACE_ENHANCER_ENDPOINT": "",
            },
            clear=False,
        ):
            c = AceToolClient(endpoint="openai")
            assert c.endpoint == "openai"

    def test_default_when_nothing_set(self):
        with patch.dict(
            os.environ,
            {
                "PROMPT_ENHANCER_ENDPOINT": "",
                "ACE_ENHANCER_ENDPOINT": "",
            },
            clear=False,
        ):
            c = AceToolClient()
            assert c.endpoint == "new"


class TestIsThirdParty:
    def _client(self, endpoint):
        with patch.dict(
            os.environ,
            {
                "PROMPT_ENHANCER_ENDPOINT": "",
                "ACE_ENHANCER_ENDPOINT": "",
            },
            clear=False,
        ):
            return AceToolClient(endpoint=endpoint)

    def test_codex_is_third_party(self):
        assert self._client("codex")._is_third_party() is True

    def test_claude_is_third_party(self):
        assert self._client("claude")._is_third_party() is True

    def test_new_is_not_third_party(self):
        assert self._client("new")._is_third_party() is False


class TestHardError:
    """Unconfigured third-party endpoints raise immediately, no silent fallback."""

    def _assert_raises(self, endpoint):
        with patch.dict(
            os.environ,
            {
                "PROMPT_ENHANCER_ENDPOINT": "",
                "ACE_ENHANCER_ENDPOINT": "",
                "PROMPT_ENHANCER_BASE_URL": "",
                "PROMPT_ENHANCER_TOKEN": "",
            },
            clear=False,
        ):
            c = AceToolClient(endpoint=endpoint)
            with pytest.raises(ValueError, match="PROMPT_ENHANCER_BASE_URL"):
                c.enhance_prompt("test", "")

    def test_unconfigured_claude_raises(self):
        self._assert_raises("claude")

    def test_unconfigured_codex_raises(self):
        self._assert_raises("codex")


class TestGetConfigExtended:
    def test_get_config_has_extended_fields(self):
        with patch.dict(
            os.environ,
            {
                "ACE_API_URL": "http://test",
                "ACE_API_TOKEN": "tok",
                "PROMPT_ENHANCER_ENDPOINT": "claude",
                "ACE_ENHANCER_ENDPOINT": "",
                "PROMPT_ENHANCER_INCLUDE_SEARCH_CONTEXT": "1",
                "PROMPT_ENHANCER_BASE_URL": "http://third",
                "PROMPT_ENHANCER_TOKEN": "tp",
            },
            clear=False,
        ):
            config = AceToolClient().get_config()
            assert "endpoint_effective" in config
            assert "endpoint_env_ready" in config
            assert "search_context_injection" in config

"""Codex endpoint: Responses API output parsing and model resolution."""

import os
from unittest.mock import patch

import pytest
from client import AceToolClient


class TestExtractCodexOutputText:
    def _extract(self, api_response):
        return AceToolClient._extract_codex_output_text(api_response)

    def test_final_answer_priority(self):
        resp = {
            "output": [
                {"type": "message", "phase": "thinking", "content": [{"type": "output_text", "text": "thinking text"}]},
                {
                    "type": "message",
                    "phase": "final_answer",
                    "content": [{"type": "output_text", "text": "final text"}],
                },
            ]
        }
        assert self._extract(resp) == "final text"

    def test_multi_part_concat(self):
        resp = {
            "output": [
                {
                    "type": "message",
                    "phase": "final_answer",
                    "content": [
                        {"type": "output_text", "text": "part1"},
                        {"type": "output_text", "text": "part2"},
                    ],
                },
            ]
        }
        assert self._extract(resp) == "part1\npart2"

    def test_refusal_raises(self):
        resp = {
            "output": [
                {"type": "message", "content": [{"type": "refusal", "refusal": "I cannot do that"}]},
            ]
        }
        with pytest.raises(RuntimeError, match="Codex API refusal"):
            self._extract(resp)

    def test_refusal_plus_text_returns_text(self):
        resp = {
            "output": [
                {
                    "type": "message",
                    "content": [
                        {"type": "refusal", "refusal": "refused"},
                        {"type": "output_text", "text": "actual output"},
                    ],
                },
            ]
        }
        assert self._extract(resp) == "actual output"

    def test_no_output_raises(self):
        with pytest.raises(RuntimeError, match="no output_text"):
            self._extract({"output": []})

    def test_empty_text_ignored(self):
        resp = {
            "output": [
                {
                    "type": "message",
                    "content": [
                        {"type": "output_text", "text": "  "},
                        {"type": "output_text", "text": "real"},
                    ],
                },
            ]
        }
        assert self._extract(resp) == "real"

    def test_null_content_raises(self):
        resp = {"output": [{"type": "message", "content": None}]}
        with pytest.raises(RuntimeError, match="no output_text"):
            self._extract(resp)


class TestCodexModelResolution:
    def test_codex_default_model(self):
        with patch.dict(
            os.environ,
            {
                "PROMPT_ENHANCER_ENDPOINT": "",
                "ACE_ENHANCER_ENDPOINT": "",
                "PROMPT_ENHANCER_MODEL": "",
            },
            clear=False,
        ):
            c = AceToolClient(endpoint="codex")
            from templates import DEFAULT_CODEX_MODEL

            assert c._get_third_party_model() == DEFAULT_CODEX_MODEL

"""Indexer ignore patterns: .gitignore + .aceignore merge behavior."""

from indexer import Indexer


class TestLoadIgnorePatterns:
    def test_reads_gitignore(self, tmp_path):
        (tmp_path / ".gitignore").write_text("*.pyc\n__pycache__\n", encoding="utf-8")
        idx = Indexer(str(tmp_path), "http://fake.url", "fake-token")
        assert "*.pyc" in idx._gitignore_patterns
        assert "__pycache__" in idx._gitignore_patterns

    def test_reads_aceignore(self, tmp_path):
        (tmp_path / ".aceignore").write_text("secret/\n*.log\n", encoding="utf-8")
        idx = Indexer(str(tmp_path), "http://fake.url", "fake-token")
        assert "secret/" in idx._gitignore_patterns
        assert "*.log" in idx._gitignore_patterns

    def test_merges_both_files(self, tmp_path):
        (tmp_path / ".gitignore").write_text("*.pyc\n", encoding="utf-8")
        (tmp_path / ".aceignore").write_text("*.log\n", encoding="utf-8")
        idx = Indexer(str(tmp_path), "http://fake.url", "fake-token")
        assert "*.pyc" in idx._gitignore_patterns
        assert "*.log" in idx._gitignore_patterns

    def test_skips_comments_and_empty_lines(self, tmp_path):
        (tmp_path / ".aceignore").write_text("# comment\n\n  \nvalid_pattern\n", encoding="utf-8")
        idx = Indexer(str(tmp_path), "http://fake.url", "fake-token")
        assert "valid_pattern" in idx._gitignore_patterns
        assert "# comment" not in idx._gitignore_patterns

    def test_no_files_empty_patterns(self, tmp_path):
        idx = Indexer(str(tmp_path), "http://fake.url", "fake-token")
        assert idx._gitignore_patterns == []

    def test_aceignore_only(self, tmp_path):
        (tmp_path / ".aceignore").write_text("build/\n", encoding="utf-8")
        idx = Indexer(str(tmp_path), "http://fake.url", "fake-token")
        assert "build/" in idx._gitignore_patterns

"""Reasoning effort control: resolution order and per-endpoint payload injection."""

import os
from unittest.mock import patch

from client import AceToolClient

_BASE_ENV = {
    "PROMPT_ENHANCER_ENDPOINT": "",
    "ACE_ENHANCER_ENDPOINT": "",
    "PROMPT_ENHANCER_BASE_URL": "https://h",
    "PROMPT_ENHANCER_TOKEN": "t",
    "PROMPT_ENHANCER_INCLUDE_SEARCH_CONTEXT": "",
    "PROMPT_ENHANCER_REASONING_EFFORT": "",
}


def _make(endpoint, effort_env=None, effort_arg="__unset__"):
    env = dict(_BASE_ENV)
    if effort_env is None:
        env.pop("PROMPT_ENHANCER_REASONING_EFFORT")
    else:
        env["PROMPT_ENHANCER_REASONING_EFFORT"] = effort_env
    with patch.dict(os.environ, env, clear=False):
        kwargs = {} if effort_arg == "__unset__" else {"reasoning_effort": effort_arg}
        return AceToolClient(endpoint=endpoint, **kwargs)


def _captured(client, endpoint):
    """Call the endpoint-specific API with a stubbed _post_json; return its payload."""
    responses = {
        "openai": {"choices": [{"message": {"content": "<augment-enhanced-prompt>x</augment-enhanced-prompt>"}}]},
        "codex": {"output": [{"type": "message", "content": [{"type": "output_text", "text": "x"}]}]},
        "claude": {"content": [{"type": "text", "text": "x"}]},
        "gemini": {"candidates": [{"content": {"parts": [{"text": "x"}]}}]},
    }
    captured = {}

    def fake_post(url, payload, **kw):
        captured.update(payload)
        return responses[endpoint]

    client._post_json = fake_post
    getattr(client, f"_call_{endpoint}_api")("test prompt", [], client._get_third_party_model())
    return captured


class TestEffortResolution:
    def test_default_is_none(self):
        assert _make("openai").reasoning_effort == "none"

    def test_env_override_normalized(self):
        assert _make("openai", effort_env=" HIGH ").reasoning_effort == "high"

    def test_constructor_wins_over_env(self):
        assert _make("openai", effort_env="high", effort_arg="low").reasoning_effort == "low"

    def test_empty_env_sends_nothing(self):
        assert _make("openai", effort_env="").reasoning_effort == ""


class TestPayloadInjection:
    def test_openai_default(self):
        p = _captured(_make("openai"), "openai")
        assert p["reasoning_effort"] == "none"
        assert p["max_tokens"] == 4096

    def test_openai_explicit(self):
        p = _captured(_make("openai", effort_env="high"), "openai")
        assert p["reasoning_effort"] == "high"

    def test_openai_empty_omits_key(self):
        p = _captured(_make("openai", effort_env=""), "openai")
        assert "reasoning_effort" not in p

    def test_codex_default(self):
        p = _captured(_make("codex"), "codex")
        assert p["reasoning"] == {"effort": "none"}

    def test_codex_empty_omits_key(self):
        p = _captured(_make("codex", effort_env=""), "codex")
        assert "reasoning" not in p

    def test_claude_default_disabled(self):
        p = _captured(_make("claude"), "claude")
        assert p["thinking"] == {"type": "disabled"}
        assert "output_config" not in p

    def test_claude_explicit_adaptive_with_effort(self):
        p = _captured(_make("claude", effort_env="high"), "claude")
        assert p["thinking"] == {"type": "adaptive"}
        assert p["output_config"] == {"effort": "high"}

    def test_claude_empty_omits_key(self):
        p = _captured(_make("claude", effort_env=""), "claude")
        assert "thinking" not in p

    def test_gemini_none_maps_to_minimal(self):
        p = _captured(_make("gemini"), "gemini")
        assert p["generationConfig"]["thinkingConfig"] == {"thinkingLevel": "minimal"}

    def test_gemini_explicit_level(self):
        p = _captured(_make("gemini", effort_env="low"), "gemini")
        assert p["generationConfig"]["thinkingConfig"] == {"thinkingLevel": "low"}

    def test_gemini_empty_omits_key(self):
        p = _captured(_make("gemini", effort_env=""), "gemini")
        assert "thinkingConfig" not in p["generationConfig"]

"""Search context injection: env toggle, normalization, prompt wrapping, guards."""

import os
from unittest.mock import patch

import pytest
from client import AceToolClient


class TestShouldIncludeSearchContext:
    def _check(self, env_val, expected):
        with patch.dict(
            os.environ,
            {
                "PROMPT_ENHANCER_ENDPOINT": "",
                "ACE_ENHANCER_ENDPOINT": "",
                "PROMPT_ENHANCER_INCLUDE_SEARCH_CONTEXT": env_val,
            },
            clear=False,
        ):
            assert AceToolClient()._should_include_search_context() is expected

    def test_true_values(self):
        for val in ("1", "true", "yes", "on", " True ", " YES ", " ON "):
            self._check(val, True)

    def test_false_values(self):
        for val in ("0", "false", "no", "off", "", " "):
            self._check(val, False)


class TestNormalizeSearchContext:
    def _normalize(self, text):
        return AceToolClient._normalize_search_context(text)

    def test_empty_returns_placeholder(self):
        from templates import NO_RELEVANT_CODE_CONTEXT

        assert self._normalize("") == NO_RELEVANT_CODE_CONTEXT

    def test_whitespace_returns_placeholder(self):
        from templates import NO_RELEVANT_CODE_CONTEXT

        assert self._normalize("   ") == NO_RELEVANT_CODE_CONTEXT

    def test_normal_text_passthrough(self):
        assert self._normalize("some code context") == "some code context"

    def test_exact_limit_no_truncation(self):
        from templates import SEARCH_CONTEXT_CHAR_LIMIT

        result = self._normalize("a" * SEARCH_CONTEXT_CHAR_LIMIT)
        assert len(result) == SEARCH_CONTEXT_CHAR_LIMIT
        assert "[codebase_context truncated" not in result

    def test_over_limit_truncated(self):
        from templates import SEARCH_CONTEXT_CHAR_LIMIT

        result = self._normalize("a" * (SEARCH_CONTEXT_CHAR_LIMIT + 1))
        assert "[codebase_context truncated for length]" in result


class TestBuildPromptWithSearchContext:
    def _build(self, original, ctx):
        return AceToolClient._build_prompt_with_search_context(original, ctx)

    def test_contains_codebase_context_tag(self):
        result = self._build("my prompt", "some context")
        assert "<codebase_context>" in result
        assert "</codebase_context>" in result

    def test_contains_original_request_tag(self):
        result = self._build("my prompt", "some context")
        assert "<original_request>" in result
        assert "</original_request>" in result

    def test_original_and_context_preserved(self):
        result = self._build("my prompt", "some context")
        assert "my prompt" in result
        assert "some context" in result


class TestMaybeInjectSearchContext:
    def test_missing_project_root_raises(self):
        with patch.dict(
            os.environ,
            {
                "PROMPT_ENHANCER_ENDPOINT": "claude",
                "ACE_ENHANCER_ENDPOINT": "",
                "PROMPT_ENHANCER_INCLUDE_SEARCH_CONTEXT": "1",
                "PROMPT_ENHANCER_BASE_URL": "https://api.example.com",
                "PROMPT_ENHANCER_TOKEN": "tok",
                "ACE_API_URL": "https://ace.example.com",
                "ACE_API_TOKEN": "ace-tok",
            },
            clear=False,
        ):
            c = AceToolClient(endpoint="claude")
            with pytest.raises(ValueError, match="project"):
                c._maybe_inject_search_context("claude", "test prompt", None)

    def test_non_third_party_returns_original(self):
        with patch.dict(
            os.environ,
            {
                "PROMPT_ENHANCER_ENDPOINT": "",
                "ACE_ENHANCER_ENDPOINT": "",
                "PROMPT_ENHANCER_INCLUDE_SEARCH_CONTEXT": "1",
            },
            clear=False,
        ):
            c = AceToolClient(endpoint="new")
            assert c._maybe_inject_search_context("new", "test prompt", "/some/path") == "test prompt"

    def test_disabled_returns_original(self):
        with patch.dict(
            os.environ,
            {
                "PROMPT_ENHANCER_ENDPOINT": "claude",
                "ACE_ENHANCER_ENDPOINT": "",
                "PROMPT_ENHANCER_INCLUDE_SEARCH_CONTEXT": "0",
            },
            clear=False,
        ):
            c = AceToolClient(endpoint="claude")
            assert c._maybe_inject_search_context("claude", "test prompt", "/some/path") == "test prompt"

"""Regression tests for diagnose fixes: URL consistency, unknown-blobs self-heal, cache hierarchy."""

import gzip
import json
import os
import shutil
import sys
from pathlib import Path
from types import SimpleNamespace

import httpx
import pytest

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))

import ace_cli  # noqa: E402
from client import AceToolClient, build_api_url  # noqa: E402
from indexer import Indexer, IndexRebuildError, ProjectIndex  # noqa: E402
from templates import INDEX_DIR, INDEX_FILE  # noqa: E402
from utils import build_api_url as utils_build_api_url  # noqa: E402


def _write_index(root: Path):
    p = root / INDEX_DIR / INDEX_FILE
    p.parent.mkdir(parents=True, exist_ok=True)
    with gzip.open(p, "wt", encoding="utf-8") as f:
        json.dump({"entries": {}, "last_indexed": 1.0}, f)
    return p


# --- Problem 1: URL construction consistency ---


def test_build_api_url_lives_in_utils_and_reexported():
    assert build_api_url is utils_build_api_url


@pytest.mark.parametrize(
    "base,path,expected",
    [
        ("https://api.example.com", "/v1/messages", "https://api.example.com/v1/messages"),
        ("https://api.example.com/v1", "/v1/messages", "https://api.example.com/v1/messages"),
        ("https://proxy.com/v1beta", "/v1/messages", "https://proxy.com/v1beta/messages"),
        ("https://api.example.com/vertex", "/v1/messages", "https://api.example.com/vertex/v1/messages"),
        ("https://api.example.com", "v1/messages", "https://api.example.com/v1/messages"),
        ("https://api.example.com/v1/", "/v1/messages", "https://api.example.com/v1/messages"),
        ("https://h/v1", "/batch-upload", "https://h/v1/batch-upload"),
    ],
)
def test_build_api_url_version_handling(base, path, expected):
    assert build_api_url(base, path) == expected


def test_upload_url_uses_build_api_url(tmp_path, monkeypatch):
    captured = {}

    class FakeResp:
        status_code = 200

        def raise_for_status(self):
            pass

    class FakeClient:
        def __init__(self, *a, **k):
            pass

        def __enter__(self):
            return self

        def __exit__(self, *a):
            return False

        def post(self, url, **kw):
            captured["url"] = url
            return FakeResp()

    monkeypatch.setattr("indexer.httpx.Client", FakeClient)
    idx = Indexer(str(tmp_path), "https://h/v1", "tok")
    assert idx._upload_batch_with_retry({}, {"blobs": []})
    assert captured["url"] == build_api_url("https://h/v1", "/batch-upload")


# --- Problem 2: 400 unknown blobs self-heal ---


def test_unknown_blobs_triggers_rebuild_and_single_retry(monkeypatch):
    c = AceToolClient(base_url="https://h", token="tok")
    calls = {"post": 0, "rebuild": 0}

    monkeypatch.setattr("client.Indexer.__init__", lambda self, *a, **k: None)
    monkeypatch.setattr("client.Indexer.get_blob_names", lambda self: ["stale"])

    def fake_rebuild(self):
        calls["rebuild"] += 1
        return ["fresh"]

    monkeypatch.setattr("client.Indexer.force_rebuild", fake_rebuild, raising=False)

    def fake_post(url, payload, *, headers, provider="API", timeout=None):
        calls["post"] += 1
        if calls["post"] == 1:
            resp = httpx.Response(400, text="unknown blobs: ['stale']")
            raise httpx.HTTPStatusError("400 Bad Request", request=None, response=resp)
        assert payload["blobs"]["added_blobs"] == ["fresh"]
        return {"formatted_retrieval": "ok"}

    monkeypatch.setattr(c, "_post_json", fake_post)

    result = c._remote_search("/proj", "query")
    assert result["results"] == "ok"
    assert calls["rebuild"] == 1
    assert calls["post"] == 2


def test_unknown_blobs_no_infinite_retry(monkeypatch):
    c = AceToolClient(base_url="https://h", token="tok")
    calls = {"post": 0}

    monkeypatch.setattr("client.Indexer.__init__", lambda self, *a, **k: None)
    monkeypatch.setattr("client.Indexer.get_blob_names", lambda self: ["stale"])
    monkeypatch.setattr("client.Indexer.force_rebuild", lambda self: ["fresh"], raising=False)

    def fake_post(url, payload, *, headers, provider="API", timeout=None):
        calls["post"] += 1
        resp = httpx.Response(400, text="unknown blobs")
        raise httpx.HTTPStatusError("400 Bad Request", request=None, response=resp)

    monkeypatch.setattr(c, "_post_json", fake_post)

    with pytest.raises(httpx.HTTPStatusError):
        c._remote_search("/proj", "query")
    assert calls["post"] == 2


def test_other_400_errors_not_swallowed(monkeypatch):
    c = AceToolClient(base_url="https://h", token="tok")
    calls = {"post": 0, "rebuild": 0}

    monkeypatch.setattr("client.Indexer.__init__", lambda self, *a, **k: None)
    monkeypatch.setattr("client.Indexer.get_blob_names", lambda self: ["a"])
    monkeypatch.setattr(
        "client.Indexer.force_rebuild",
        lambda self: calls.__setitem__("rebuild", calls["rebuild"] + 1) or [],
        raising=False,
    )

    def fake_post(url, payload, *, headers, provider="API", timeout=None):
        calls["post"] += 1
        resp = httpx.Response(400, text="malformed request")
        raise httpx.HTTPStatusError("400 Bad Request", request=None, response=resp)

    monkeypatch.setattr(c, "_post_json", fake_post)

    with pytest.raises(httpx.HTTPStatusError):
        c._remote_search("/proj", "query")
    assert calls["post"] == 1
    assert calls["rebuild"] == 0


# --- Problem 3: cache hierarchy ---


def test_child_inherits_nearest_ancestor_cache(tmp_path):
    parent = tmp_path / "parent"
    child = parent / "sub"
    child.mkdir(parents=True)
    _write_index(parent)

    idx = Indexer(str(child), "https://h", "tok")
    assert idx.root == parent.resolve()
    assert idx.index_path == parent.resolve() / INDEX_DIR / INDEX_FILE


def test_own_cache_preferred_over_ancestor(tmp_path):
    parent = tmp_path / "p2"
    child = parent / "sub"
    child.mkdir(parents=True)
    _write_index(parent)
    _write_index(child)

    idx = Indexer(str(child), "https://h", "tok")
    assert idx.root == child.resolve()


def test_no_cache_anywhere_uses_project_root(tmp_path):
    proj = tmp_path / "fresh"
    proj.mkdir()

    idx = Indexer(str(proj), "https://h", "tok")
    assert idx.root == proj.resolve()


def test_nearest_ancestor_wins_over_farther(tmp_path):
    outer = tmp_path / "repo"
    nearest = outer / "packages"
    child = nearest / "service"
    child.mkdir(parents=True)
    _write_index(outer)
    _write_index(nearest)

    idx = Indexer(str(child), "https://h", "tok")
    assert idx.root == nearest.resolve()


def test_home_cache_not_inherited(tmp_path, monkeypatch):
    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    _write_index(tmp_path)
    proj = tmp_path / "work" / "proj"
    proj.mkdir(parents=True)

    idx = Indexer(str(proj), "https://h", "tok")
    assert idx.root == proj.resolve()


def test_parent_indexing_absorbs_child_cache(tmp_path):
    root = tmp_path / "proj"
    sub = root / "pkg"
    sub.mkdir(parents=True)
    (root / "m.py").write_text("a = 1\n", encoding="utf-8")
    (sub / "n.py").write_text("b = 2\n", encoding="utf-8")
    _write_index(root)
    _write_index(sub)

    idx = Indexer(str(root), "", "")  # empty base_url: upload step is a no-op
    names = idx.get_blob_names()

    assert not (sub / INDEX_DIR).exists()
    assert (root / INDEX_DIR / INDEX_FILE).exists()
    assert len(names) == 2


# --- Review fixes: absorb boundaries, retry amplification, CLI transparency ---


def test_child_cache_kept_when_root_never_saves_index(tmp_path):
    root = tmp_path / "empty-root"
    sub = root / "pkg"
    sub.mkdir(parents=True)
    _write_index(sub)  # only a child cache, no eligible source files anywhere

    idx = Indexer(str(root), "", "")
    names = idx.get_blob_names()

    assert names == []
    assert not (root / INDEX_DIR / INDEX_FILE).exists()
    assert (sub / INDEX_DIR / INDEX_FILE).exists()


def test_ignored_subtree_cache_not_absorbed(tmp_path):
    root = tmp_path / "proj2"
    ignored = root / "legacy"
    ignored.mkdir(parents=True)
    (root / "m.py").write_text("a = 1\n", encoding="utf-8")
    (ignored / "old.py").write_text("z = 9\n", encoding="utf-8")
    (root / ".gitignore").write_text("legacy/\n", encoding="utf-8")
    _write_index(ignored)

    idx = Indexer(str(root), "", "")
    names = idx.get_blob_names()

    assert (ignored / INDEX_DIR / INDEX_FILE).exists()  # not covered by parent scan
    assert len(names) == 1  # only m.py


def test_force_rebuild_raises_on_upload_failure(tmp_path, monkeypatch):
    (tmp_path / "a.py").write_text("x = 1\n", encoding="utf-8")
    idx = Indexer(str(tmp_path), "https://h", "tok")
    monkeypatch.setattr(idx, "_upload_pending", lambda: False)

    with pytest.raises(IndexRebuildError):
        idx.force_rebuild()


def test_rebuild_failure_not_retried_by_enhance(monkeypatch):
    monkeypatch.delenv("PROMPT_ENHANCER_ENDPOINT", raising=False)
    monkeypatch.delenv("ACE_ENHANCER_ENDPOINT", raising=False)
    c = AceToolClient(base_url="https://h", token="tok", endpoint="claude")
    calls = {"n": 0}

    def fake_third_party(prompt, history, project_root=None):
        calls["n"] += 1
        raise IndexRebuildError("rebuild upload failed")

    monkeypatch.setattr(c, "_call_third_party_api", fake_third_party)

    with pytest.raises(IndexRebuildError):
        c.enhance_prompt("p", "", "/proj")
    assert calls["n"] == 1  # ValueError subclass: excluded from tenacity retry


def test_cmd_index_reports_effective_root(tmp_path, monkeypatch, capsys):
    parent = tmp_path / "top"
    child = parent / "sub"
    child.mkdir(parents=True)
    _write_index(parent)

    monkeypatch.setattr(
        ace_cli,
        "AceToolClient",
        lambda *a: SimpleNamespace(base_url="https://h", token="tok"),
    )

    def fake_names(self):
        self._index = ProjectIndex()
        return []

    monkeypatch.setattr(ace_cli.Indexer, "get_blob_names", fake_names)

    ace_cli.cmd_index(SimpleNamespace(api_url=None, token=None, endpoint=None, project_root=str(child)))
    out = json.loads(capsys.readouterr().out)
    assert out["effective_root"] == str(parent.resolve())
    assert out["project_root"] == str(child)


# --- Concurrency race: tolerate-and-converge (absorb vs concurrent child save) ---


def test_save_index_tolerates_dir_deleted_mid_write(tmp_path, monkeypatch):
    (tmp_path / "a.py").write_text("x = 1\n", encoding="utf-8")
    idx = Indexer(str(tmp_path), "", "")
    idx._load_index()
    idx._scan_and_update()

    real_mkdir = Path.mkdir

    def mkdir_then_vanish(self, *a, **k):
        real_mkdir(self, *a, **k)
        shutil.rmtree(self)  # concurrent ancestor absorbs the dir right after mkdir

    monkeypatch.setattr(Path, "mkdir", mkdir_then_vanish)

    assert idx._save_index() is False  # abandoned, not crashed
    assert not idx.index_path.exists()


def test_save_index_returns_true_on_success(tmp_path):
    (tmp_path / "a.py").write_text("x = 1\n", encoding="utf-8")
    idx = Indexer(str(tmp_path), "", "")
    idx._load_index()
    idx._scan_and_update()

    assert idx._save_index() is True
    assert idx.index_path.is_file()


def test_no_absorb_when_save_abandoned(tmp_path, monkeypatch):
    root = tmp_path / "proj3"
    sub = root / "pkg"
    sub.mkdir(parents=True)
    (root / "m.py").write_text("a = 1\n", encoding="utf-8")
    (sub / "n.py").write_text("b = 2\n", encoding="utf-8")
    _write_index(root)
    _write_index(sub)

    idx = Indexer(str(root), "", "")
    monkeypatch.setattr(idx, "_save_index", lambda: False)  # own cache absorbed by grandparent
    idx.get_blob_names()

    assert (sub / INDEX_DIR / INDEX_FILE).exists()  # no cascading deletion


def test_force_rebuild_skips_absorb_when_save_abandoned(tmp_path, monkeypatch):
    (tmp_path / "a.py").write_text("x = 1\n", encoding="utf-8")
    sub = tmp_path / "pkg"
    sub.mkdir()
    (sub / "n.py").write_text("b = 2\n", encoding="utf-8")
    _write_index(sub)

    idx = Indexer(str(tmp_path), "", "")
    monkeypatch.setattr(idx, "_save_index", lambda: False)
    names = idx.force_rebuild()  # upload is a no-op (empty base_url) -> success

    assert names  # in-memory result still returned
    assert (sub / INDEX_DIR / INDEX_FILE).exists()


def test_absorb_skips_child_on_rmtree_failure(tmp_path, monkeypatch):
    root = tmp_path / "proj4"
    sub = root / "pkg"
    sub.mkdir(parents=True)
    (root / "m.py").write_text("a = 1\n", encoding="utf-8")
    (sub / "n.py").write_text("b = 2\n", encoding="utf-8")
    _write_index(root)
    _write_index(sub)

    def locked_rmtree(p):
        raise PermissionError(13, "file in use")  # Windows open-handle / racy delete

    monkeypatch.setattr("indexer.shutil.rmtree", locked_rmtree)
    idx = Indexer(str(root), "", "")
    names = idx.get_blob_names()  # must not raise

    assert (sub / INDEX_DIR / INDEX_FILE).exists()  # skipped, retried next run
    assert len(names) == 2


def test_save_index_tmp_is_per_process(tmp_path, monkeypatch):
    captured = {}
    real_replace = Path.replace

    def spy_replace(self, target):
        captured["tmp"] = self.name
        return real_replace(self, target)

    monkeypatch.setattr(Path, "replace", spy_replace)
    (tmp_path / "a.py").write_text("x = 1\n", encoding="utf-8")
    idx = Indexer(str(tmp_path), "", "")
    idx._load_index()
    idx._scan_and_update()

    assert idx._save_index() is True
    assert captured["tmp"] == f"{INDEX_FILE}.{os.getpid()}.tmp"  # no shared-name torn writes
    assert list((tmp_path / INDEX_DIR).glob("*.tmp")) == []  # renamed away, no litter


def test_save_index_retries_transient_replace_permission_error(tmp_path, monkeypatch):
    attempts = {"n": 0}
    real_replace = Path.replace

    def briefly_locked_replace(self, target):
        attempts["n"] += 1
        if attempts["n"] == 1:
            raise PermissionError(5, "target open in a concurrent reader")  # WinError 5
        return real_replace(self, target)

    monkeypatch.setattr(Path, "replace", briefly_locked_replace)
    monkeypatch.setattr("indexer.time.sleep", lambda s: None)
    (tmp_path / "a.py").write_text("x = 1\n", encoding="utf-8")
    idx = Indexer(str(tmp_path), "", "")
    idx._load_index()
    idx._scan_and_update()

    assert idx._save_index() is True
    assert attempts["n"] == 2
    assert idx.index_path.is_file()


def test_save_index_propagates_persistent_permission_error(tmp_path, monkeypatch):
    def always_locked_replace(self, target):
        raise PermissionError(5, "acl denies write")

    monkeypatch.setattr(Path, "replace", always_locked_replace)
    monkeypatch.setattr("indexer.time.sleep", lambda s: None)
    (tmp_path / "a.py").write_text("x = 1\n", encoding="utf-8")
    idx = Indexer(str(tmp_path), "", "")
    idx._load_index()
    idx._scan_and_update()

    with pytest.raises(PermissionError):  # genuine ACL problems must surface
        idx._save_index()