ChatAgent

@dataclass
class ChatAgentConfig:
    """Configuration for ChatAgent."""

    # LLM settings
    use_claude: bool = False
    use_chatgpt: bool = False
    claude_model: str = "claude-sonnet-4-20250514"
    # None = fall back to LEMONADE_BASE_URL env var, then the default
    base_url: Optional[str] = None
    model_id: Optional[str] = None  # Default: Qwen3.5-35B

    # Execution
    max_steps: int = 10
    streaming: bool = False

    # Debug/output
    debug: bool = False
    debug_prompts: bool = False     # Backward-compatible alias for show_prompts
    show_prompts: bool = False
    show_stats: bool = False
    silent_mode: bool = False
    output_dir: Optional[str] = None

    # RAG settings
    rag_documents: List[str] = field(default_factory=list)
    library_documents: List[str] = field(default_factory=list)  # UI-attached docs
    watch_directories: List[str] = field(default_factory=list)
    chunk_size: int = 500
    chunk_overlap: int = 100
    max_chunks: int = 5
    use_llm_chunking: bool = False  # False = fast heuristic

    # Security
    allowed_paths: Optional[List[str]] = None

    # Agent UI integration
    ui_session_id: Optional[str] = None      # resume an Agent-UI session
    enable_sd_tools: bool = False            # Stable Diffusion image generation

class ChatAgent(
    Agent,
    RAGToolsMixin,
    FileToolsMixin,
    ShellToolsMixin,
    FileSearchToolsMixin,
    FileIOToolsMixin,
    VLMToolsMixin,
    ScreenshotToolsMixin,
    SDToolsMixin,
    MCPClientMixin,
):
    """
    Chat agent with RAG, file operations, and shell capabilities.
    """

    def __init__(self, config: Optional[ChatAgentConfig] = None):
        """Initialize Chat Agent with config."""
        pass

    def load_session(self, session_id: str) -> bool:
        """Load saved session, restoring indexed docs and history."""
        pass

    def save_current_session(self) -> bool:
        """Save current session state."""
        pass

    def reindex_file(self, file_path: str) -> None:
        """Reindex modified/created file (auto-called by file watcher)."""
        pass

    def stop_watching(self) -> None:
        """Stop all file system observers."""
        pass

@tool
def query_documents(query: str, max_chunks: int = 5) -> Dict[str, Any]:
    """Search all indexed documents for relevant content."""
    pass

@tool
def index_document(file_path: str) -> Dict[str, Any]:
    """Index a document for RAG search."""
    pass

@tool
def search_file(file_pattern: str, directory: str = None) -> Dict[str, Any]:
    """
    Search for files matching pattern across drives.
    Phase 1: Common locations (Documents, Downloads, Desktop)
    Phase 2: Deep search entire drive if not found
    """
    pass

@tool
def read_file(file_path: str) -> Dict[str, Any]:
    """Read file content with security validation."""
    pass

@tool
def run_shell_command(command: str, timeout: int = 30) -> Dict[str, Any]:
    """Execute shell command and return output."""
    pass

# System prompt teaches agent this workflow:
# 1. Check if relevant documents indexed
# 2. If NO:
#    a. Extract key terms from question
#    b. Search for files: search_file(file_pattern="key terms")
#    c. Index found files: index_document(file_path)
#    d. Provide status: "Found and indexed X file(s)"
#    e. Query to answer: query_specific_file(...)
# 3. If YES: query directly

User: "what is the vision of the oil & gas regulator?"

Agent: {"tool": "list_indexed_documents", "tool_args": {}}
Result: {"documents": [], "count": 0}

Agent: {"tool": "search_file", "tool_args": {"file_pattern": "oil gas regulator"}}
Result: {"files": ["/docs/Oil-Gas-Manual.pdf"], "count": 1}

Agent: {"tool": "index_document", "tool_args": {"file_path": "/docs/Oil-Gas-Manual.pdf"}}
Result: {"status": "success", "chunks": 150}

Agent: {"tool": "query_specific_file", "tool_args": {"file_path": "/docs/Oil-Gas-Manual.pdf", "query": "vision"}}
Result: {"chunks": ["The vision is to be recognized..."], "scores": [0.92]}

Agent: {"answer": "According to the Oil & Gas Manual, the vision is..."}

class FileChangeHandler(FileSystemEventHandler):
    """Handler for file system changes to trigger re-indexing."""

    def __init__(self, agent):
        self.agent = agent
        self.last_indexed = {}  # LRU cache
        self.debounce_time = 2.0  # seconds

    def on_modified(self, event):
        """Handle file modification."""
        if not event.is_directory and self._should_index(event.src_path):
            self._schedule_reindex(event.src_path)

    def _schedule_reindex(self, file_path: str):
        """Debounced reindexing with LRU eviction."""
        current_time = time.time()
        last_time = self.last_indexed.get(file_path, 0)

        if current_time - last_time > self.debounce_time:
            self.last_indexed[file_path] = current_time
            self.agent.reindex_file(file_path)

            # LRU eviction (max 1000 entries)
            if len(self.last_indexed) > 1000:
                oldest = sorted(self.last_indexed.items(), key=lambda x: x[1])[:100]
                for path, _ in oldest:
                    del self.last_indexed[path]

def _validate_and_open_file(self, file_path: str, mode: str = "r"):
    """
    Safely open file with TOCTOU protection.

    Uses O_NOFOLLOW flag to reject symlinks.
    """
    import stat

    # Determine flags
    flags = os.O_RDONLY if 'r' in mode else os.O_WRONLY | os.O_CREAT

    # CRITICAL: Add O_NOFOLLOW
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW

    # Open file descriptor
    try:
        fd = os.open(file_path, flags)
    except OSError as e:
        if e.errno == 40:  # ELOOP (symlink)
            raise PermissionError(f"Symlinks not allowed: {file_path}")
        raise

    # Verify it's a regular file
    file_stat = os.fstat(fd)
    if not stat.S_ISREG(file_stat.st_mode):
        os.close(fd)
        raise PermissionError(f"Not a regular file: {file_path}")

    # Validate against allowed paths
    real_path = Path(os.readlink(f"/proc/self/fd/{fd}")).resolve()
    if not self._is_path_allowed(real_path):
        os.close(fd)
        raise PermissionError(f"Access denied: {real_path}")

    return os.fdopen(fd, mode)

# tests/agents/test_chat_agent.py

def test_chat_agent_initialization(tmp_path):
    """Test ChatAgent initializes correctly."""
    config = ChatAgentConfig(
        rag_documents=[],
        allowed_paths=[str(tmp_path)],
        silent_mode=True
    )
    agent = ChatAgent(config)
    assert agent is not None

def test_smart_discovery_workflow(tmp_path):
    """Test smart discovery finds and indexes files."""
    # Create test document
    doc = tmp_path / "manual.pdf"
    doc.write_text("The vision is...")

    config = ChatAgentConfig(
        allowed_paths=[str(tmp_path)],
        silent_mode=True
    )
    agent = ChatAgent(config)

    # Query should trigger discovery
    result = agent.process_query("What is the vision in the manual?")

    assert result["status"] == "success"
    assert agent.rag.indexed_files
    assert str(doc) in agent.rag.indexed_files

def test_file_security_symlink_rejection(tmp_path):
    """Test that symlinks are rejected."""
    real_file = tmp_path / "real.txt"
    real_file.write_text("content")

    symlink = tmp_path / "link.txt"
    symlink.symlink_to(real_file)

    config = ChatAgentConfig(allowed_paths=[str(tmp_path)], silent_mode=True)
    agent = ChatAgent(config)

    with pytest.raises(PermissionError, match="Symlinks not allowed"):
        agent._validate_and_open_file(str(symlink))

def test_session_persistence(tmp_path):
    """Test session save/load."""
    config = ChatAgentConfig(
        rag_documents=[],
        allowed_paths=[str(tmp_path)],
        silent_mode=True
    )
    agent = ChatAgent(config)

    # Index a document
    doc = tmp_path / "test.txt"
    doc.write_text("test content")
    agent.rag.index_document(str(doc))

    # Save session
    assert agent.save_current_session()
    session_id = agent.current_session.session_id

    # Load in new agent
    agent2 = ChatAgent(config)
    assert agent2.load_session(session_id)
    assert str(doc) in agent2.indexed_files

[project.optional-dependencies]
rag = [
    "sentence-transformers>=2.0.0",
    "faiss-cpu>=1.7.0",
    "pypdf>=3.0.0",
]

chat = [
    "watchdog>=2.1.0",  # File monitoring
    "aiohttp>=3.8.0",   # Async HTTP
]

from gaia.agents.chat.agent import ChatAgent, ChatAgentConfig

# Configure agent
config = ChatAgentConfig(
    rag_documents=["./docs/manual.pdf"],
    chunk_size=500,
    max_chunks=5
)

# Create agent
agent = ChatAgent(config)

# Query document
result = agent.process_query("What are the safety guidelines?")
print(result["result"])

config = ChatAgentConfig(
    watch_directories=["./watched_dir"],
    rag_documents=[]  # Will auto-index files found in watched_dir
)

agent = ChatAgent(config)

# Files in watched_dir are auto-indexed when created/modified
# Query them naturally
result = agent.process_query("Search the latest report")

config = ChatAgentConfig()
agent = ChatAgent(config)

# Smart search: Phase 1 (fast) → Phase 2 (deep)
result = agent.process_query("Find the oil and gas manual on my computer")

# Agent will:
# 1. Search common locations (Documents, Downloads)
# 2. If not found, deep search all drives
# 3. Present results with numbered list
# 4. Auto-index selected file