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  # None = config default_model, else Gemma-4-E4B-it-GGUF

    # 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
    memory_incognito: bool = False           # memory off: load the embedder only on first use
    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).

        On Windows, raises PermissionError when the file cannot be read
        because another program holds it exclusively or has locked part of it,
        so the watcher retries it; any other failure is logged with its
        traceback.
        """
        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..."}

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 = [
    "faiss-cpu>=1.7.0",
    "pypdf>=3.0.0",
]

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

from gaia_agent_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