Brainhair - AI Chat Assistant

Version: 1.0 Port: 5050 Database: PostgreSQL Repository: hivematrix-brainhair

Table of Contents

1. Overview
2. Architecture
3. Features
4. Database Schema
5. API Reference
6. Claude Code Integration
7. Tools System
8. Approval Workflow
9. PHI/CJIS Filtering
10. Contract Alignment
11. Configuration
12. Session Management
13. Frontend Integration
14. Development
15. Monitoring
16. Troubleshooting
17. Security

---

Overview

Brainhair is HiveMatrix's conversational AI assistant service, powered by Claude Code. It provides natural language access to all HiveMatrix data and services, enabling technicians to interact with tickets, companies, billing data, knowledge base articles, and device information through conversational chat.

Purpose

Brainhair serves as the intelligent assistant that: - Provides natural language queries across all HiveMatrix services - Executes safe read operations automatically - Requests approval for write/modify operations - Filters sensitive PHI/CJIS data for compliance - Maintains conversation context and history - Assists with contract analysis and billing alignment - Integrates with Datto RMM for device management

Key Technologies

• Claude Code CLI: Anthropic's official CLI for Claude integration
• Microsoft Presidio: PII/PHI detection and anonymization
• PostgreSQL: Session and message persistence
• Flask 3.0.0: Web framework
• Polling Architecture: Better proxy compatibility than SSE
• JWT Authentication: Service-to-service secure communication

---

Architecture

System Architecture

┌─────────────────────────────────────────────────────────────┐
│                    Brainhair Service (5050)                  │
│  ┌────────────────────────────────────────────────────────┐ │
│  │              Chat Routes (/api/chat)                   │ │
│  │  - POST /api/chat (create response buffer)             │ │
│  │  - GET /api/chat/poll/<id> (poll for chunks)          │ │
│  │  - POST /api/chat/stop/<id> (stop response)           │ │
│  └─────────────────┬──────────────────────────────────────┘ │
│                    │                                          │
│  ┌─────────────────┴──────────────────────────────────────┐ │
│  │           ClaudeSessionManager                         │ │
│  │  - Creates ClaudeSession instances                     │ │
│  │  - Manages session lifecycle                           │ │
│  │  - Background cleanup thread (30 min idle timeout)     │ │
│  └─────────────────┬──────────────────────────────────────┘ │
│                    │                                          │
│  ┌─────────────────┴──────────────────────────────────────┐ │
│  │              ClaudeSession                             │ │
│  │  - Spawns `claude` CLI subprocess                      │ │
│  │  - Streams JSON output via stdout                      │ │
│  │  - Manages conversation history                        │ │
│  │  - Injects system prompt + context                     │ │
│  │  - Handles tool discovery and invocation               │ │
│  └───┬──────────────────────────────────────────┬─────────┘ │
│      │                                          │            │
│  ┌───┴────────────────┐           ┌────────────┴──────────┐ │
│  │   Claude Tools     │           │  Presidio Filter      │ │
│  │  - codex_tools     │           │  - Analyzer Engine    │ │
│  │  - knowledge_tools │           │  - Anonymizer Engine  │ │
│  │  - datto_tools     │           │  - PHI entities       │ │
│  │  - billing_tools   │           │  - CJIS entities      │ │
│  │  - session_tools   │           │  - Custom operators   │ │
│  │  - contract_tools  │           └───────────────────────┘ │
│  └────────────────────┘                                      │
│                                                               │
│  ┌────────────────────────────────────────────────────────┐ │
│  │              Database Models                           │ │
│  │  - ChatSession (user, ticket, client context)          │ │
│  │  - ChatMessage (role, content, tool_calls, filtering)  │ │
│  └────────────────────────────────────────────────────────┘ │
└───────────────────────────────────────────────────────────────┘
                            │
                            │ Service-to-Service Calls
                            │ (JWT Authenticated)
                            ▼
┌───────────────────────────────────────────────────────────────┐
│   Codex        Ledger      KnowledgeTree      Core      Helm  │
│   (5010)       (5030)         (5020)         (5000)    (5004) │
└───────────────────────────────────────────────────────────────┘

Claude Code Integration Flow

1. User sends message via POST /api/chat
   ↓
2. Chat route creates response buffer
   ↓
3. Background thread spawns Claude Code subprocess:
   claude --model claude-sonnet-4-5 \
          --allowed-tools "Bash Read Grep Glob WebFetch WebSearch" \
          --permission-mode dontAsk \
          --output-format stream-json \
          --append-system-prompt <system_prompt> \
          <user_message>
   ↓
4. Claude Code streams JSON events to stdout:
   - content_block_delta (text chunks)
   - content_block_start (tool calls)
   - message_stop (completion)
   ↓
5. Background thread accumulates chunks in response buffer
   ↓
6. Client polls GET /api/chat/poll/<response_id>?offset=N
   ↓
7. Returns new chunks since offset
   ↓
8. Repeat polling until done=true

Polling vs SSE Architecture

Why Polling Instead of SSE? - Better compatibility with Nexus reverse proxy - No proxy buffering issues - Easier to inject approval requests mid-stream - Simpler error recovery - Works with all browsers and proxies

Polling Implementation:

# Server-side response buffer
response_buffers[response_id] = {
    'chunks': [],           # Accumulated chunks
    'done': False,          # Completion flag
    'error': None,          # Error message if failed
    'created_at': time.time(),
    'session_id': session_id,
    'db_session_id': db_session_id
}

# Background thread collects chunks
for chunk in session.send_message_stream(message):
    response_buffers[response_id]['chunks'].append(chunk_data)

# Client polls with offset
GET /api/chat/poll/abc-123?offset=10
→ Returns chunks[10:], new offset, done flag

---

Features

Core Features

1. Conversational AI

• Natural language queries across all HiveMatrix services
• Context-aware responses with conversation history
• Streaming responses for real-time interaction
• Session resumption and history browsing

2. Multi-Service Data Access

• Codex: Companies, contacts, tickets, devices
• Ledger: Billing data, invoices, plans, overrides
• KnowledgeTree: Knowledge base articles and categories
• Core: User information and authentication
• Datto RMM: Device status, diagnostics, remote commands

3. Intelligent Tool Calling

• Auto-discovered tools from claude_tools/ directory
• Read-only tools execute immediately (no approval)
• Write operations require human approval
• Tool results integrated into conversation

4. PHI/CJIS Data Filtering

• Microsoft Presidio integration
• Automatic entity detection and anonymization
• PHI mode: Names, emails, phones, SSNs, credit cards, IPs
• CJIS mode: Criminal justice data protection
• Custom name anonymizer (FirstName L. format)

5. Approval Workflow

• Human-in-the-loop for dangerous operations
• Clear approval UI with full command display
• Audit logging of all approvals/denials
• Timeout and status tracking

6. Contract Alignment

• Analyze contracts using Claude's NLP
• Extract billing terms automatically
• Compare contract vs current settings
• Generate alignment reports
• Apply changes with approval

7. Session Management

• UUID-based session identification
• Database persistence for history
• Context tracking (ticket, client, user)
• Automatic idle cleanup (30 min timeout)
• Session resumption from history

Specialized Features

Context Awareness

{
  "user": "john.technician",
  "ticket": "12345",
  "client": "ACME Corp",
  "ticket_details": {
    "title": "Password reset needed",
    "status": "open",
    "priority": "medium"
  },
  "client_details": {
    "name": "ACME Corp",
    "plan": "All Inclusive",
    "domain": "acme.com"
  }
}

Context is automatically fetched from Codex when ticket/client is set, providing Claude with immediate background information.

Tool Discovery System

# Auto-discovers tools from claude_tools/
def _discover_ai_tools(self):
    """Scan claude_tools/ and generate documentation"""
    for filename in sorted(os.listdir(ai_tools_dir)):
        if filename.endswith('.py'):
            # Extract docstring
            # Parse IMPORTANT notes
            # Extract usage examples
            # Categorize by function

Generates system prompt section:

## Available AI Tools (Auto-Discovered)

### Companies & Tickets
**codex_tools.py** - Query companies and support tickets from Codex
```bash
get_companies(limit=100)
get_tickets(status='open', limit=50)

Billing & Plans

billing_tools.py - Access billing data and modify plans IMPORTANT: Write operations require user approval

get_billing(account_number)
update_billing(account_number, adjustments)  # REQUIRES APPROVAL

---

## Database Schema

### ChatSession

Represents a conversation session with context.

```python
class ChatSession(db.Model):
    __tablename__ = 'chat_sessions'

    # Identity
    id = db.Column(db.String(36), primary_key=True)  # UUID
    user_id = db.Column(db.String(150), nullable=False, index=True)
    user_name = db.Column(db.String(150))

    # Context
    ticket_number = db.Column(db.String(50), index=True)
    client_name = db.Column(db.String(150), index=True)

    # Metadata
    created_at = db.Column(db.DateTime, default=datetime.utcnow,
                          nullable=False, index=True)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow,
                          onupdate=datetime.utcnow, nullable=False)
    ended_at = db.Column(db.DateTime)

    # Status
    is_active = db.Column(db.Boolean, default=True,
                         nullable=False, index=True)

    # Summary
    title = db.Column(db.String(255))  # User-visible title
    summary = db.Column(db.Text)        # Auto-generated summary

    # Relationships
    messages = db.relationship('ChatMessage', back_populates='session',
                             cascade='all, delete-orphan',
                             order_by='ChatMessage.created_at')

Indexes: - user_id - Fast user session lookup - ticket_number - Find sessions by ticket - client_name - Find sessions by client - created_at - Chronological ordering - is_active - Filter active sessions

Example Data:

{
  "id": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
  "user_id": "john.technician",
  "user_name": "John T.",
  "ticket_number": "12345",
  "client_name": "ACME Corp",
  "created_at": "2025-11-22T10:30:00",
  "updated_at": "2025-11-22T10:45:00",
  "ended_at": null,
  "is_active": true,
  "title": "ACME Corp Billing Review",
  "summary": "Discussion about billing alignment for ACME Corp contract",
  "message_count": 8
}

ChatMessage

Individual messages in a conversation.

class ChatMessage(db.Model):
    __tablename__ = 'chat_messages'

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    session_id = db.Column(db.String(36),
                          db.ForeignKey('chat_sessions.id'),
                          nullable=False, index=True)

    # Message data
    role = db.Column(db.String(20), nullable=False)  # 'user' or 'assistant'
    content = db.Column(db.Text, nullable=False)

    # Tool usage tracking
    tool_calls = db.Column(db.JSON)    # Tools called in this message
    tool_results = db.Column(db.JSON)  # Results from tool calls

    # Metadata
    created_at = db.Column(db.DateTime, default=datetime.utcnow,
                          nullable=False, index=True)

    # Filtering info
    was_filtered = db.Column(db.Boolean, default=False)
    filter_type = db.Column(db.String(20))  # 'phi' or 'cjis'

    # Relationships
    session = db.relationship('ChatSession', back_populates='messages')

    # Composite index
    __table_args__ = (
        db.Index('idx_message_session_created', 'session_id', 'created_at'),
    )

Tool Tracking Example:

{
  "id": 42,
  "session_id": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
  "role": "assistant",
  "content": "I found 5 companies matching your criteria...",
  "tool_calls": [
    {
      "tool": "codex_tools.get_companies",
      "args": {"limit": 100},
      "timestamp": "2025-11-22T10:31:00"
    }
  ],
  "tool_results": [
    {
      "tool": "codex_tools.get_companies",
      "count": 87,
      "filtered": true
    }
  ],
  "created_at": "2025-11-22T10:31:05",
  "was_filtered": true,
  "filter_type": "phi"
}

---

API Reference

Chat Endpoints

Create Chat Response

Start a new chat message and begin streaming response.

POST /api/chat
Authorization: Bearer <token>
Content-Type: application/json

{
  "message": "List recent tickets for ACME Corp",
  "session_id": "existing-session-uuid",     # Optional - resume session
  "db_session_id": "db-session-uuid",        # Optional - load from history
  "ticket": "12345",                         # Optional - ticket context
  "client": "ACME Corp"                      # Optional - client context
}

Response:

{
  "response_id": "abc-def-123",
  "session_id": "f47ac10b-58cc-4372-a567-0e02b2c3d479"
}

Process: 1. Creates response buffer 2. Spawns background thread with Claude Code subprocess 3. Returns immediately with response_id 4. Client begins polling

Example:

TOKEN=$(python create_test_token.py 2>/dev/null)
curl -X POST http://localhost:5050/brainhair/api/chat \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "message": "What is the billing plan for Example Company?",
    "client": "Example Company"
  }'

# Returns:
# {"response_id": "abc-123", "session_id": "uuid-456"}

---

Poll for Response Chunks

Poll for new chunks from a running response.

GET /api/chat/poll/<response_id>?offset=<N>
Authorization: Bearer <token>

Parameters: - response_id - Response identifier from POST /api/chat - offset - Number of chunks already received (default 0)

Response:

{
  "chunks": [
    {"type": "chunk", "content": "I'll check the billing"},
    {"type": "chunk", "content": " plan for Example Company..."},
    {"type": "live_message", "content": "Claude is responding..."},
    {"type": "approval_request", "approval_id": "xyz", "action": "..."}
  ],
  "offset": 15,
  "done": false,
  "error": null,
  "session_id": "uuid-456"
}

Chunk Types: - chunk - Text content to display - live_message - Status messages (initializing, tool use, etc.) - approval_request - User approval required (injected from /tmp file) - error - Error occurred

Polling Pattern:

let offset = 0;
const poll = async () => {
  const res = await fetch(`/brainhair/api/chat/poll/${responseId}?offset=${offset}`, {
    credentials: 'same-origin'
  });
  const data = await res.json();

  // Display new chunks
  data.chunks.forEach(chunk => {
    if (chunk.type === 'chunk') {
      appendMessage(chunk.content);
    } else if (chunk.type === 'approval_request') {
      showApprovalCard(chunk);
    }
  });

  offset = data.offset;

  if (!data.done) {
    setTimeout(poll, 500);  // Poll every 500ms
  }
};

poll();

---

Stop Response

Stop a running Claude Code response.

POST /api/chat/stop/<response_id>
Authorization: Bearer <token>

Response:

{
  "success": true,
  "message": "Response stopped"
}

Process: 1. Finds running Claude Code process 2. Sends SIGTERM signal 3. Waits 2 seconds for graceful shutdown 4. Force kills (SIGKILL) if still running 5. Marks response buffer as done

Example:

await fetch(`/brainhair/api/chat/stop/${responseId}`, {
  method: 'POST',
  credentials: 'same-origin'
});

---

Destroy Session

Destroy a Claude Code session (cleanup).

DELETE /api/chat/session/<session_id>
Authorization: Bearer <token>

Response:

{
  "status": "destroyed"
}

Process: - Stops any running Claude Code process - Removes session from memory - Database session remains for history

---

History Endpoints

List Chat Sessions

List all chat sessions for current user.

GET /api/chat/history?ticket=<num>&client=<name>&limit=<N>&offset=<N>
Authorization: Bearer <token>

Parameters: - ticket - Filter by ticket number (optional) - client - Filter by client name (optional, uses ILIKE) - limit - Max sessions to return (default 50, max 100) - offset - Pagination offset (default 0)

Response:

{
  "sessions": [
    {
      "id": "uuid-123",
      "user_id": "john.technician",
      "user_name": "John T.",
      "ticket_number": "12345",
      "client_name": "ACME Corp",
      "created_at": "2025-11-22T10:30:00",
      "updated_at": "2025-11-22T10:45:00",
      "is_active": true,
      "title": "ACME Corp Billing Review",
      "message_count": 8
    }
  ],
  "total": 142,
  "limit": 50,
  "offset": 0
}

Example:

TOKEN=$(python create_test_token.py 2>/dev/null)
curl -H "Authorization: Bearer $TOKEN" \
  "http://localhost:5050/brainhair/api/chat/history?limit=10"

---

Get Chat Session

Get specific chat session with all messages.

GET /api/chat/history/<session_id>
Authorization: Bearer <token>

Response:

{
  "id": "uuid-123",
  "user_id": "john.technician",
  "ticket_number": "12345",
  "client_name": "ACME Corp",
  "title": "ACME Corp Billing Review",
  "created_at": "2025-11-22T10:30:00",
  "messages": [
    {
      "id": 1,
      "role": "user",
      "content": "What is ACME Corp's billing plan?",
      "created_at": "2025-11-22T10:30:15"
    },
    {
      "id": 2,
      "role": "assistant",
      "content": "ACME Corp is on the All Inclusive plan...",
      "tool_calls": [{"tool": "billing_tools.get_billing"}],
      "created_at": "2025-11-22T10:30:20"
    }
  ]
}

Security: - Only session owner can access (verified by user_id) - Returns 403 Forbidden if user doesn't own session

---

Search Chat History

Search chat history by keywords.

GET /api/chat/history/search?q=<query>&ticket=<num>&limit=<N>
Authorization: Bearer <token>

Parameters: - q - Search query (required) - ticket - Filter by ticket (optional) - limit - Max results (default 20, max 50)

Search Fields: - Session title - Session summary - Message content

Response:

{
  "sessions": [
    {
      "id": "uuid-123",
      "title": "Example Company Billing Setup",
      "created_at": "2025-11-20T14:00:00",
      "message_count": 12
    }
  ],
  "query": "billing",
  "count": 5
}

Example:

TOKEN=$(python create_test_token.py 2>/dev/null)
curl -H "Authorization: Bearer $TOKEN" \
  "http://localhost:5050/brainhair/api/chat/history/search?q=billing&limit=10"

---

Update Session Title

Update the title of a chat session.

PUT /api/chat/session/<session_id>/title
Authorization: Bearer <token>
Content-Type: application/json

{
  "title": "Example Company Contract Alignment"
}

Response:

{
  "success": true,
  "session_id": "uuid-123",
  "title": "Example Company Contract Alignment"
}

Security: - User calls: Verified by session ownership - Service calls: Trusted (no ownership check)

Usage: This is typically called by the session_tools.set_chat_title() tool early in the conversation to provide a descriptive title for the history list.

---

Approval Endpoints

Create Approval Request

Create an approval request (called by tools).

POST /api/approval/request
Authorization: Bearer <token>
Content-Type: application/json

{
  "session_id": "uuid-123",
  "action": "Update billing plan",
  "details": {
    "account_number": "GD-001",
    "company": "Example Company",
    "change": "Set per_user_cost to $15.00"
  }
}

Response:

{
  "approval_id": "approval-uuid-456",
  "status": "pending"
}

Process: 1. Creates approval record in pending_approvals dict 2. Writes JSON to /tmp/brainhair_approval_request_<session_id>_<approval_id>.json 3. Polling endpoint injects this into stream 4. Tool polls for response

---

Poll Approval Status

Poll for approval status (called by tools waiting for response).

GET /api/approval/poll/<approval_id>
Authorization: Bearer <token>

Response:

{
  "status": "pending",      # Or: "approved", "denied"
  "result": null            # User's response if completed
}

Tool Polling Loop:

import time

approval = create_approval(session_id, action, details)
approval_id = approval['approval_id']

# Poll for user response (max 5 minutes)
for _ in range(60):
    status = poll_approval(approval_id)
    if status['status'] in ['approved', 'denied']:
        return status
    time.sleep(5)

return {'error': 'Approval timeout'}

---

Get Pending Approvals

Get all pending approvals for a session (called by browser).

GET /api/approval/pending/<session_id>
Authorization: Bearer <token>

Response:

{
  "approvals": [
    {
      "approval_id": "approval-uuid-456",
      "action": "Update billing plan",
      "details": {
        "account_number": "GD-001",
        "company": "Example Company",
        "change": "Set per_user_cost to $15.00"
      }
    }
  ]
}

---

Respond to Approval

User approves or denies an approval request.

POST /api/approval/respond/<approval_id>
Authorization: Bearer <token>
Content-Type: application/json

{
  "approved": true
}

Response:

{
  "success": true,
  "approved": true
}

Process: 1. Validates approval_id format (regex: ^[a-zA-Z0-9_-]+$) 2. Writes response to /tmp/brainhair_approval_response_<approval_id>.json 3. Tool reads file on next poll 4. Logs approval/denial to Helm

File Contents:

{
  "approved": true
}

---

Command Endpoints

Approve Command

Approve and execute a pending PowerShell command.

POST /api/chat/command/approve
Authorization: Bearer <token>
Content-Type: application/json

{
  "command_id": "cmd-uuid-123"
}

Response:

{
  "status": "success",
  "output": "Command output here..."
}

Process: 1. Validates command_id exists in pending_commands 2. Logs approval (user, command, device) 3. Executes via Datto RMM (currently simulated) 4. Updates command status to "executed" 5. Returns output

Current Status: - Datto RMM integration not yet implemented - Returns simulated output - See execute_remote_command() in chat_routes.py:491

---

Deny Command

Deny a pending command.

POST /api/chat/command/deny
Authorization: Bearer <token>
Content-Type: application/json

{
  "command_id": "cmd-uuid-123"
}

Response:

{
  "status": "denied"
}

Process: 1. Logs denial (user, command_id) 2. Updates status to "denied" 3. Removes from pending_commands

---

Get Command Status

Check status of a command execution.

GET /api/command/<command_id>/status
Authorization: Bearer <token>

Response:

{
  "command_id": "cmd-uuid-123",
  "status": "executed",
  "device_id": "device-456",
  "command": "Get-ComputerInfo",
  "reason": "Check system specs",
  "executed_by": "john.technician",
  "output": "Command output..."
}

Status Values: - pending - Awaiting approval - approved - Approved but not yet executed - denied - User denied - executed - Successfully executed - failed - Execution failed

---

Claude Code Integration

How It Works

Brainhair uses the claude CLI binary from @anthropic-ai/claude-code NPM package to interact with Claude. This provides a robust, officially-supported integration with streaming JSON output.

Installation

Claude Code is installed via NPX:

# Automatic installation on first use
npx -y @anthropic-ai/claude-code

# Or install globally
npm install -g @anthropic-ai/claude-code

# Verify installation
which claude
claude --version

Invocation

cmd = [
    claude_bin,
    '--model', 'claude-sonnet-4-5',
    '--allowed-tools', 'Bash Read Grep Glob WebFetch WebSearch',  # READ ONLY
    '--permission-mode', 'dontAsk',         # Auto-approve safe ops
    '--verbose',                            # Required for stream-json
    '--print',                              # Required for output-format
    '--output-format', 'stream-json',       # Real-time JSON events
    '--include-partial-messages',           # Include partial chunks
    '--append-system-prompt', full_prompt,  # Inject system instructions
    message                                 # User's message
]

process = subprocess.Popen(
    cmd,
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    env=self.env,
    text=True,
    bufsize=1  # Line buffered
)

Streaming JSON Format

Claude Code emits JSON events on stdout, one per line:

{"type": "stream_event", "event": {"type": "message_start"}}
{"type": "stream_event", "event": {"type": "content_block_start", "content_block": {"type": "text"}}}
{"type": "stream_event", "event": {"type": "content_block_delta", "delta": {"type": "text_delta", "text": "I'll "}}}
{"type": "stream_event", "event": {"type": "content_block_delta", "delta": {"type": "text_delta", "text": "check "}}}
{"type": "stream_event", "event": {"type": "content_block_delta", "delta": {"type": "text_delta", "text": "the billing..."}}}
{"type": "stream_event", "event": {"type": "content_block_stop"}}
{"type": "stream_event", "event": {"type": "message_delta", "delta": {"stop_reason": "end_turn"}}}
{"type": "stream_event", "event": {"type": "message_stop"}}

Event Processing

for line in iter(process.stdout.readline, ''):
    event = json.loads(line.strip())

    # Unwrap stream_event wrapper
    if event.get('type') == 'stream_event':
        inner_event = event.get('event', {})
        event_type = inner_event.get('type')

        if event_type == 'content_block_delta':
            delta = inner_event.get('delta', {})
            if delta.get('type') == 'text_delta':
                text = delta.get('text', '')
                response_text += text
                yield text  # Stream to client

System Prompt

The system prompt is generated dynamically and includes:

1. Base Instructions - From claude_tools/SYSTEM_PROMPT.md
2. Tool Documentation - Auto-discovered from claude_tools/
3. Current Context - User, ticket, client details

# Brainhair AI Assistant

You are Brainhair, an AI assistant for HiveMatrix MSP platform...

## Current Context

- **Technician**: john.technician
- **Ticket**: 12345
- **Client**: ACME Corp

## Available AI Tools (Auto-Discovered)

### Companies & Tickets
**codex_tools.py** - Query companies and support tickets
```bash
get_companies(limit=100)
get_ticket(ticket_id)

Billing & Plans

billing_tools.py - Access and modify billing data IMPORTANT: Write operations require user approval

### Environment Variables

Brainhair sets these environment variables for tool access:

```python
self.env = os.environ.copy()
self.env['HIVEMATRIX_USER'] = self.user
self.env['HIVEMATRIX_CONTEXT'] = json.dumps(self.context)
self.env['BRAINHAIR_SESSION_ID'] = self.db_session_id
self.env['BRAINHAIR_URL'] = 'http://localhost:5050'

# Add pyenv Python to PATH for tools
pyenv_bin = os.path.join(project_root, 'pyenv', 'bin')
self.env['PATH'] = pyenv_bin + ':' + self.env.get('PATH', '')

# Add tools directory to PYTHONPATH
tools_dir = os.path.join(os.path.dirname(__file__), '..', 'claude_tools')
self.env['PYTHONPATH'] = tools_dir + ':' + self.env.get('PYTHONPATH', '')

Tools can access these:

import os
import json

user = os.environ.get('HIVEMATRIX_USER')
context = json.loads(os.environ.get('HIVEMATRIX_CONTEXT', '{}'))
session_id = os.environ.get('BRAINHAIR_SESSION_ID')

Allowed Tools (Read-Only)

For security, Claude Code is restricted to read-only tools:

• Bash - Execute bash commands (limited to safe operations)
• Read - Read files from disk
• Grep - Search file contents
• Glob - Find files by pattern
• WebFetch - Fetch web pages
• WebSearch - Search the web

NOT Allowed: - Write - Writing files - Edit - Editing files - Task - Spawning sub-agents

This ensures Claude cannot modify files or execute arbitrary code without going through the approval system.

Tool Output Integration

When Claude uses a tool, the output is captured and returned:

{
  "type": "stream_event",
  "event": {
    "type": "content_block_start",
    "content_block": {
      "type": "tool_use",
      "id": "toolu_123",
      "name": "bash",
      "input": {"command": "python claude_tools/codex_tools.py"}
    }
  }
}

The tool output is then provided back to Claude:

{
  "type": "stream_event",
  "event": {
    "type": "tool_result",
    "tool_use_id": "toolu_123",
    "content": "{'companies': [...], 'count': 87}"
  }
}

Claude integrates this into its response:

I found 87 companies in the system. The top companies by revenue are...

---

Tools System

Architecture

Brainhair has two tool directories:

1. claude_tools/ - Python modules with clean APIs (preferred)
2. ai_tools/ - Legacy command-line tools

claude_tools/ (Preferred)

Python modules that provide function-based APIs.

Structure:

claude_tools/
├── __init__.py           # Exports all tools
├── codex_tools.py        # Companies, tickets, contacts
├── knowledge_tools.py    # Knowledge base search
├── billing_tools.py      # Billing data access (read/write)
├── datto_tools.py        # Device management, commands
├── session_tools.py      # Session management (set title)
├── contract_tools.py     # Contract alignment
├── browse_knowledge.py   # Browse KB categories
└── SYSTEM_PROMPT.md      # Base system instructions

Example: codex_tools.py

"""
Codex Service Tools

Tools for interacting with the Codex service (companies, tickets, clients).
"""

import os
import sys
import json

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from app.service_client import call_service


def get_companies(limit: int = 100) -> dict:
    """
    Get list of all companies from Codex.

    Args:
        limit: Maximum number of companies to return

    Returns:
        {
            "companies": [
                {"id": 1, "name": "Company Name", ...},
                ...
            ],
            "count": 123
        }

    Example:
        >>> companies = get_companies(limit=50)
        >>> print(f"Found {companies['count']} companies")
    """
    try:
        response = call_service('codex', f'/api/companies?limit={limit}')
        return response.json()
    except Exception as e:
        return {
            'error': 'Failed to retrieve companies',
            'companies': [],
            'count': 0
        }


def get_ticket(ticket_id: int) -> dict:
    """
    Get detailed information about a specific ticket.

    Args:
        ticket_id: Ticket ID

    Returns:
        {
            "id": 12345,
            "subject": "Password reset",
            "status": "open",
            ...
        }
    """
    try:
        response = call_service('codex', f'/api/ticket/{ticket_id}')
        return response.json()
    except Exception as e:
        return {
            'error': 'Failed to retrieve ticket details',
            'id': ticket_id
        }

Claude Usage:

Claude can import and call these directly:

from codex_tools import get_companies, get_ticket

# List companies
companies = get_companies(limit=50)
print(f"Found {companies['count']} companies")

# Get ticket details
ticket = get_ticket(12345)
print(f"Ticket: {ticket['subject']} - {ticket['status']}")

Available Tools

codex_tools.py

Functions: - get_companies(limit=100) - List all companies - get_company(company_id) - Get company details - get_tickets(company_id=None, status=None, limit=50) - List tickets - get_ticket(ticket_id) - Get ticket details - update_ticket(ticket_id, status=None, notes=None) - Update ticket

Example:

from codex_tools import get_tickets

tickets = get_tickets(status='open', limit=10)
for ticket in tickets['tickets']:
    print(f"#{ticket['id']}: {ticket['subject']}")

---

knowledge_tools.py

Functions: - search_knowledge(query, limit=20) - Search KB by keywords - browse_knowledge(category_id=None) - Browse categories - get_article(article_id) - Get full article content

Example:

from knowledge_tools import search_knowledge

results = search_knowledge("password reset", limit=5)
for article in results['articles']:
    print(f"{article['title']} - {article['category']}")

---

billing_tools.py

Functions: - get_billing(account_number) - Get billing data (READ ONLY) - get_billing_plans() - List available plans (READ ONLY) - update_billing(account_number, adjustments) - Update billing (REQUIRES APPROVAL) - set_company_plan(account_number, plan_name) - Change plan (REQUIRES APPROVAL) - add_custom_line_item(account_number, name, amount) - Add line item (REQUIRES APPROVAL)

Approval Example:

from billing_tools import update_billing

# This will trigger approval workflow
result = update_billing(
    "GD-001",
    {
        "per_user_cost": 15.00,
        "prepaid_hours_monthly": 4.0
    }
)

# Returns:
# {
#   "approval_required": True,
#   "approval_id": "approval-uuid",
#   "status": "pending"
# }

---

datto_tools.py

Functions: - get_devices(company_id=None, status=None) - List devices (READ ONLY) - get_device(device_id) - Get device details (READ ONLY) - execute_command(device_id, command, reason) - Run PowerShell (REQUIRES APPROVAL) - get_command_status(command_id) - Check command status

Command Example:

from datto_tools import execute_command, get_command_status

# Request command execution
result = execute_command(
    "device-123",
    "Get-ComputerInfo | Select-Object CsName,WindowsVersion",
    "Check OS version for ticket #12345"
)

# User must approve via UI
# Then check status
status = get_command_status(result['command_id'])
if status['status'] == 'executed':
    print(status['output'])

---

session_tools.py

Functions: - set_chat_title(title) - Set session title - get_current_session_info() - Get session info

Usage:

from session_tools import set_chat_title

# Set descriptive title early in conversation
set_chat_title("Example Company Contract Alignment")

---

contract_tools.py

Functions: - analyze_contract(contract_text, account_number) - Parse contract - get_current_settings(account_number) - Get billing settings - compare_contract_to_settings(account_number, contract_terms) - Find discrepancies - align_settings(account_number, adjustments, dry_run=True) - Apply changes - verify_alignment(account_number, contract_terms) - Verify alignment

Workflow:

from contract_tools import (
    get_current_settings,
    compare_contract_to_settings,
    align_settings,
    verify_alignment
)

# 1. Get current settings
current = get_current_settings("GD-001")

# 2. Compare with contract terms (Claude extracts these from contract text)
contract_terms = {
    "per_user_rate": 15.00,
    "hourly_rate": 150.00,
    "prepaid_hours_monthly": 4.0
}

comparison = compare_contract_to_settings("GD-001", contract_terms)
print(f"Discrepancies: {comparison['discrepancies_found']}")

# 3. Align (dry run first)
alignment = align_settings("GD-001", {
    "per_user_cost": 15.00,
    "per_hour_ticket_cost": 150.00,
    "prepaid_hours_monthly": 4.0
}, dry_run=True)

# 4. Apply for real (requires approval)
alignment = align_settings("GD-001", {...}, dry_run=False)

# 5. Verify
verification = verify_alignment("GD-001", contract_terms)
print(f"Aligned: {verification['aligned']}")

---

ai_tools/ (Legacy)

Command-line tools (legacy, may be deprecated).

Structure:

ai_tools/
├── list_companies.py      # List companies (CLI)
├── list_tickets.py        # List tickets (CLI)
├── search_knowledge.py    # Search KB (CLI)
├── get_billing.py         # Get billing data (CLI)
├── update_billing.py      # Update billing (CLI)
├── brainhair_auth.py      # Authentication helper
└── approval_helper.py     # Approval workflow helper

Example: list_tickets.py

#!/usr/bin/env python3
"""
List Tickets - Query support tickets from Codex

Usage:
    python list_tickets.py [--company <name>] [--status <status>] [--limit <n>]

Examples:
    python list_tickets.py
    python list_tickets.py --company "ACME Corp"
    python list_tickets.py --status open --limit 20
"""

import sys
import json
import argparse
from brainhair_auth import get_auth

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--company', help='Filter by company')
    parser.add_argument('--status', help='Filter by status')
    parser.add_argument('--limit', type=int, default=50)
    args = parser.parse_args()

    # Get authenticated client
    auth = get_auth()

    # Build query params
    params = {'limit': args.limit}
    if args.company:
        params['company'] = args.company
    if args.status:
        params['status'] = args.status

    # Call Codex API
    response = auth.get('/api/tickets', params=params)
    tickets = response.json()

    # Output for Claude to parse
    print(json.dumps(tickets, indent=2))

if __name__ == '__main__':
    main()

Claude Usage:

Claude executes as bash command:

python ai_tools/list_tickets.py --status open --limit 10

---

Approval Workflow

Architecture

┌─────────────────────────────────────────────────────────────┐
│                    Approval Workflow                         │
│                                                               │
│  1. Tool requests approval                                    │
│     → POST /api/approval/request                              │
│     → Creates pending_approvals[approval_id]                  │
│     → Writes /tmp/brainhair_approval_request_<session>_<id>.json
│                                                               │
│  2. Polling endpoint injects approval into stream             │
│     → GET /api/chat/poll/<response_id>                        │
│     → Reads /tmp/brainhair_approval_request_*.json            │
│     → Injects into chunks[] array                             │
│                                                               │
│  3. Browser displays approval card                            │
│     → Shows action, details, approve/deny buttons             │
│                                                               │
│  4. User clicks Approve or Deny                               │
│     → POST /api/approval/respond/<approval_id>                │
│     → Writes /tmp/brainhair_approval_response_<id>.json       │
│                                                               │
│  5. Tool polls for response                                   │
│     → GET /api/approval/poll/<approval_id>                    │
│     → Reads /tmp/brainhair_approval_response_<id>.json        │
│     → Returns approved=true/false                             │
│                                                               │
│  6. Tool executes or aborts based on response                 │
│     → If approved: Execute operation                          │
│     → If denied: Return error to Claude                       │
│                                                               │
│  7. Result logged to Helm                                     │
│     → Approval/denial logged with user, command, result       │
└─────────────────────────────────────────────────────────────┘

File-Based Communication

Why Files Instead of Database? - Tools run in subprocess, don't have database access - Files are simple, reliable IPC mechanism - /tmp cleanup handled by OS - Regex validation prevents path traversal

Request File Format:

/tmp/brainhair_approval_request_<session_id>_<approval_id>.json

{
  "type": "approval_request",
  "approval_id": "approval-uuid-456",
  "session_id": "session-uuid-123",
  "action": "Update billing plan",
  "details": {
    "account_number": "GD-001",
    "company": "Example Company Lighting",
    "change": "Set per_user_cost to $15.00"
  }
}

Response File Format:

/tmp/brainhair_approval_response_<approval_id>.json

{
  "approved": true
}

Tool Implementation

Tools request approval using helper function:

# approval_helper.py
import json
import time
import os
import uuid
import requests

def request_approval(action: str, details: dict) -> bool:
    """
    Request approval from user.

    Args:
        action: Description of action (e.g., "Update billing")
        details: Dict with details to show user

    Returns:
        True if approved, False if denied
    """
    # Get session ID from environment
    session_id = os.environ.get('BRAINHAIR_SESSION_ID')
    if not session_id:
        raise RuntimeError("No session ID found - cannot request approval")

    # Get Brainhair URL
    brainhair_url = os.environ.get('BRAINHAIR_URL', 'http://localhost:5050')

    # Create approval request
    approval_id = str(uuid.uuid4())

    # Write request file
    request_file = f"/tmp/brainhair_approval_request_{session_id}_{approval_id}.json"
    with open(request_file, 'w') as f:
        json.dump({
            'type': 'approval_request',
            'approval_id': approval_id,
            'session_id': session_id,
            'action': action,
            'details': details
        }, f)

    # Poll for response (max 5 minutes)
    response_file = f"/tmp/brainhair_approval_response_{approval_id}.json"
    for _ in range(60):  # 5 minutes = 60 * 5 seconds
        if os.path.exists(response_file):
            with open(response_file, 'r') as f:
                response = json.load(f)

            # Cleanup
            os.unlink(request_file)
            os.unlink(response_file)

            return response.get('approved', False)

        time.sleep(5)

    # Timeout
    if os.path.exists(request_file):
        os.unlink(request_file)

    return False  # Denied by timeout

Usage in Tool:

from approval_helper import request_approval

def update_billing(account_number, adjustments):
    """Update billing settings (requires approval)."""

    # Request approval
    approved = request_approval(
        "Update billing settings",
        {
            'account_number': account_number,
            'adjustments': adjustments
        }
    )

    if not approved:
        return {'error': 'User denied the operation'}

    # Execute the update
    response = call_service('ledger', '/api/billing/update',
                           method='POST',
                           json={
                               'account_number': account_number,
                               'adjustments': adjustments
                           })

    return response.json()

Approval Card UI

Display Approval Card:

function showApprovalCard(approvalData) {
    const card = document.createElement('div');
    card.className = 'approval-card';
    card.id = `approval-${approvalData.approval_id}`;

    card.innerHTML = `
        <div class="approval-card__header">
            <h4>⚠️ Approval Required</h4>
        </div>
        <div class="approval-card__content">
            <p><strong>Action:</strong> ${escapeHtml(approvalData.action)}</p>
            <div class="approval-card__details">
                ${Object.entries(approvalData.details).map(([key, value]) => `
                    <div class="detail-row">
                        <span class="detail-label">${escapeHtml(key)}:</span>
                        <span class="detail-value">${escapeHtml(String(value))}</span>
                    </div>
                `).join('')}
            </div>
        </div>
        <div class="approval-card__actions">
            <button onclick="approveAction('${approvalData.approval_id}')"
                    class="btn btn--success">
                ✓ Approve
            </button>
            <button onclick="denyAction('${approvalData.approval_id}')"
                    class="btn btn--danger">
                ✗ Deny
            </button>
        </div>
    `;

    chatContainer.appendChild(card);
}

Handle Approval:

async function approveAction(approvalId) {
    const card = document.getElementById(`approval-${approvalId}`);

    try {
        const response = await fetch(`/brainhair/api/approval/respond/${approvalId}`, {
            method: 'POST',
            credentials: 'same-origin',
            headers: {'Content-Type': 'application/json'},
            body: JSON.stringify({approved: true})
        });

        if (response.ok) {
            card.innerHTML = '<p class="approval-card__result approval-card__result--approved">✓ Approved</p>';
        } else {
            card.innerHTML = '<p class="approval-card__result approval-card__result--error">Error approving</p>';
        }
    } catch (error) {
        console.error('Approval error:', error);
        card.innerHTML = '<p class="approval-card__result approval-card__result--error">Network error</p>';
    }
}

async function denyAction(approvalId) {
    const card = document.getElementById(`approval-${approvalId}`);

    try {
        const response = await fetch(`/brainhair/api/approval/respond/${approvalId}`, {
            method: 'POST',
            credentials: 'same-origin',
            headers: {'Content-Type': 'application/json'},
            body: JSON.stringify({approved: false})
        });

        if (response.ok) {
            card.innerHTML = '<p class="approval-card__result approval-card__result--denied">✗ Denied</p>';
        }
    } catch (error) {
        console.error('Denial error:', error);
    }
}

Security & Audit

Logging:

All approvals are logged to Helm:

from app.helm_logger import get_helm_logger

logger = get_helm_logger()

logger.info(
    f"Approval {'approved' if approved else 'denied'}",
    extra={
        'approval_id': approval_id,
        'user': user_id,
        'action': action,
        'details': details,
        'approved': approved,
        'timestamp': datetime.utcnow().isoformat()
    }
)

Path Traversal Prevention:

import re

# Validate approval_id format
if not re.match(r'^[a-zA-Z0-9_-]+$', approval_id):
    logger.warning(f"Invalid approval_id format: {approval_id}")
    return jsonify({'error': 'Invalid approval ID format'}), 400

Timeout Handling:

• Tools poll for max 5 minutes (60 × 5 seconds)
• If timeout: return denied
• Files cleaned up on timeout or completion

---

PHI/CJIS Filtering

Overview

Brainhair uses Microsoft Presidio to detect and anonymize sensitive data before sending it to external AI servers (Claude API). The system implements a three-tier compliance-based filtering system that protects data based on each company's compliance requirements.

Purpose: Prevent PII/PHI/CJIS data leakage to external AI services while maintaining operational usability.

Key Features: - Company-level compliance marking (Standard, CJIS, PHI) - Dynamic filtering based on client requirements - Automatic entity detection using spaCy NER models - Custom anonymization operators - Zero configuration required after company compliance is set

---

Three-Tier Compliance System

Brainhair applies different filtering levels based on each company's compliance_level field in Codex:

Level	Use Case	Filtering Intensity	Operational Data Preserved
Standard	Regular MSP clients	Minimal	✅ Names, emails, phones, addresses, IPs, dates
CJIS	Police departments, criminal justice	Strict	✅ Only dates preserved (all PII/CJI redacted)
HIPAA	Healthcare clients (HIPAA)	Maximum	✅ Only dates preserved (all PII/PHI redacted)

---

Standard Compliance (Default)

Use For: Regular MSP business clients

Philosophy: Only redact truly sensitive financial/identity documents. Allow operational business data through.

What Gets Redacted:

critical_entities = [
    "US_SSN",              # Social Security Numbers
    "CREDIT_CARD",         # Credit card numbers
    "IBAN_CODE",           # Bank account numbers
    "US_PASSPORT",         # Passport numbers
    "US_DRIVER_LICENSE",   # Driver's license numbers
]

What Stays Visible (Operational Data): - ✅ Names (PERSON) - Need to know who clients/techs are - ✅ Email Addresses (EMAIL_ADDRESS) - Business contact info - ✅ Phone Numbers (PHONE_NUMBER) - Business contact info - ✅ Addresses (LOCATION) - Office/site locations - ✅ IP Addresses (IP_ADDRESS) - Network troubleshooting data - ✅ Dates/Times (DATE_TIME) - Ticket timelines, due dates, schedules

Example Output:

Ticket #17933: Library New Phone and eFax
Client: Yoncalla Public Library
Contact: Amy Thompson (amy@library.org)
Phone: (541) 849-2471
Address: 120 7th St, Yoncalla, OR 97499
Created: November 15, 2024 at 9:30 AM
Last Updated: November 20, 2024 at 2:45 PM

Description: Install 1 additional phone and setup eFax service.
SSN in notes: <US_SSN>
Payment by credit card: <CREDIT_CARD>

Rationale: This is an internal MSP tool for technicians who already have access to this data in the ticketing system. Names, emails, phones, and addresses are business contact information, not protected health data or criminal justice information.

---

CJIS Compliance (Criminal Justice)

Use For: Police departments, sheriff's offices, criminal justice agencies

Philosophy: Strict filtering to prevent investigation data leakage to external AI servers.

What Gets Redacted:

cjis_entities = [
    "PERSON",              # Names of suspects, victims, witnesses, officers
    "EMAIL_ADDRESS",       # Investigative contact information
    "PHONE_NUMBER",        # Phone evidence, witness contacts
    "US_SSN",              # Social Security Numbers (subject identifiers)
    "US_DRIVER_LICENSE",   # Driver's license numbers (criminal records)
    "LOCATION",            # Crime scenes, investigation locations, addresses
    "IP_ADDRESS",          # Digital forensics data, investigation-related IPs
    # NOTE: DATE_TIME intentionally excluded per MSP operational requirements
]

Why So Strict? - CJIS Security Policy 5.10 requires protection of all Criminal Justice Information (CJI) - Criminal investigation data is highly sensitive - Suspect/victim/witness names must be protected - Email addresses and phone numbers often contain investigative contacts - Crime scene locations cannot be leaked - Driver's licenses are linked to criminal records - IP addresses may be part of digital forensics investigations

Example Output:

Ticket #17933: New workstation setup
Client: <PERSON> County Sheriff's Office
Contact: <PERSON> (<EMAIL_ADDRESS>)
Phone: <PHONE_NUMBER>
Address: <LOCATION>
Created: November 15, 2024 at 9:30 AM
Last Updated: November 20, 2024 at 2:45 PM

Description: Setup new workstation for detective division.
Officer SSN: <US_SSN>
License plate database access required.

Date/Time Handling: Dates and times are NOT redacted for CJIS clients because: - Ticket timelines are operational data - Service scheduling requires date visibility - SLA tracking needs timestamps - Dates alone don't reveal investigation details

---

HIPAA Compliance (Healthcare/HIPAA)

Use For: Healthcare providers, medical clinics, hospitals, insurance companies

Philosophy: Full HIPAA compliance. Redact all PII/PHI to prevent health information disclosure.

What Gets Redacted:

hipaa_entities = [
    "PERSON",              # Patient names, provider names
    "EMAIL_ADDRESS",       # Patient/provider emails
    "PHONE_NUMBER",        # Patient/provider phone numbers
    "US_SSN",              # Social Security Numbers
    "LOCATION",            # Patient addresses, clinic locations
    "MEDICAL_LICENSE",     # Medical provider licenses
    "US_DRIVER_LICENSE",   # Driver's license numbers
    "US_PASSPORT",         # Passport numbers
    "CREDIT_CARD",         # Payment card numbers
    "IBAN_CODE",           # Bank account numbers
    "IP_ADDRESS",          # Network data (may contain patient info)
    "NRP",                 # National Registry Provider numbers
    # NOTE: DATE_TIME intentionally excluded per MSP operational requirements
]

Why Maximum Filtering? - HIPAA requires strict protection of all PHI - Any health-related information must be redacted - Patient identity must be protected - Medical provider information is sensitive

Example Output:

Ticket #17933: <PERSON>
Client: <PERSON> Medical Clinic
Contact: <PERSON> (<EMAIL_ADDRESS>)
Phone: <PHONE_NUMBER>
Address: <LOCATION>
Created: November 15, 2024 at 9:30 AM
Last Updated: November 20, 2024 at 2:45 PM

Description: Setup EMR workstation for <PERSON>.
Provider NPI: <NRP>
Medical license: <MEDICAL_LICENSE>

Custom Name Anonymizer: Names use the FirstNameLastInitialOperator to preserve some context: - "Dr. Sarah Johnson" → "Dr. Sarah J." - "John Smith" → "John S." - Helps maintain conversational flow while protecting identity

---

Architecture

┌────────────────────────────────────────────────────────────┐
│              Presidio Filter with Compliance Levels         │
│                                                              │
│  ┌──────────────────────────────────────────────────────┐  │
│  │        AnalyzerEngine                                 │  │
│  │  - Detects entities using NER models                  │  │
│  │  - Uses spaCy en_core_web_lg model                    │  │
│  │  - Scores confidence (0.0 - 1.0)                      │  │
│  └────────────────┬─────────────────────────────────────┘  │
│                   │                                          │
│  ┌────────────────┴─────────────────────────────────────┐  │
│  │        Compliance Level Mapper                        │  │
│  │  - Reads company.compliance_level from Codex          │  │
│  │  - Maps to appropriate entity list:                   │  │
│  │    * standard → critical_entities                     │  │
│  │    * cjis → cjis_entities                             │  │
│  │    * hipaa → hipaa_entities                               │  │
│  └────────────────┬─────────────────────────────────────┘  │
│                   │                                          │
│  ┌────────────────┴─────────────────────────────────────┐  │
│  │        AnonymizerEngine                               │  │
│  │  - Replaces detected entities                         │  │
│  │  - Custom operators:                                  │  │
│  │    * FirstNameLastInitialOperator (PERSON)            │  │
│  │    * Replace (all others: <ENTITY_TYPE>)              │  │
│  └──────────────────────────────────────────────────────┘  │
└────────────────────────────────────────────────────────────┘

---

Setting Company Compliance Level

In Codex Web UI:

1. Navigate to Companies → Select company
2. Click Edit Details
3. Find Compliance Level dropdown
4. Select appropriate level:
5. Standard (Regular MSP Client) - Default for most clients
6. CJIS (Criminal Justice / Police Dept) - For law enforcement
7. HIPAA (Healthcare Compliance) - For medical organizations
8. Click Save Changes

Visual Indicators: - Standard: Gray badge - CJIS: Yellow badge (⚠️ Criminal Justice) - HIPAA: Blue badge (🏥 Healthcare)

Database Field:

ALTER TABLE companies ADD COLUMN compliance_level VARCHAR(20) DEFAULT 'standard';

---

Implementation Details

Code Structure

Filter Function (app/presidio_filter.py):

def filter_by_compliance_level(data: Any, compliance_level: str = 'standard',
                                fields_to_filter: Optional[List[str]] = None) -> Any:
    """
    Filter data based on company compliance level.

    Args:
        data: Data to filter (dict, list, or str)
        compliance_level: 'standard', 'cjis', or 'phi'
        fields_to_filter: Optional list of specific fields to filter

    Returns:
        Filtered data with appropriate entities anonymized
    """
    filter_instance = get_presidio_filter()

    # Map compliance level to entity list
    if compliance_level == 'cjis':
        entity_list = filter_instance.cjis_entities
    elif compliance_level == 'hipaa':
        entity_list = filter_instance.hipaa_entities
    else:  # 'standard' or any other value defaults to standard
        entity_list = filter_instance.critical_entities

    # Apply filtering with the appropriate entity list
    if isinstance(data, dict):
        return filter_instance.filter_dict(data, fields_to_filter, entity_list)
    elif isinstance(data, list):
        return filter_instance.filter_list(data, fields_to_filter, entity_list)
    elif isinstance(data, str):
        return filter_instance.anonymize_text(data, entity_list)
    else:
        return data

Integration with Claude Tools

get_ticket.py:

from app.presidio_filter import filter_by_compliance_level

def get_ticket(ticket_id):
    with app.app_context():
        # Get ticket from Codex
        response = call_service('codex', f'/api/ticket/{ticket_id}')
        ticket = response.json()

        # Extract company compliance level
        compliance_level = ticket.get('company_compliance_level', 'standard')

        # Apply appropriate filtering
        ticket = filter_by_compliance_level(ticket, compliance_level)

        # Now ticket data is filtered according to company's compliance requirements
        return ticket

list_tickets.py:

from app.presidio_filter import filter_by_compliance_level

def list_tickets(status=None, company_id=None, limit=10):
    with app.app_context():
        # Get tickets from Codex
        response = call_service('codex', f'/api/tickets?limit={limit}')
        data = response.json()

        # Filter each ticket based on its company's compliance level
        tickets = data.get('tickets', [])
        filtered_tickets = []

        for ticket in tickets:
            compliance_level = ticket.get('company_compliance_level', 'standard')
            filtered_ticket = filter_by_compliance_level(ticket, compliance_level)
            filtered_tickets.append(filtered_ticket)

        return filtered_tickets

list_companies.py:

from app.presidio_filter import filter_by_compliance_level

def list_companies(limit=20):
    with app.app_context():
        # Get companies from Codex
        response = call_service('codex', f'/api/companies?limit={limit}')
        companies = response.json()

        # Filter each company based on its own compliance level
        filtered_companies = []

        for company in companies:
            compliance_level = company.get('compliance_level', 'standard')
            filtered_company = filter_by_compliance_level(company, compliance_level)
            filtered_companies.append(filtered_company)

        return filtered_companies

---

Codex Integration

API Changes: Codex ticket and company endpoints now include compliance_level fields:

Single Ticket (GET /api/ticket/<id>):

{
  "id": 17933,
  "ticket_number": "TKT-17933",
  "subject": "Library New Phone and eFax",
  "company_id": "12345",
  "company_name": "Yoncalla Public Library",
  "company_compliance_level": "standard",
  "requester_name": "Amy Thompson",
  "requester_email": "amy@library.org",
  "created_at": "2024-11-15T09:30:00Z"
}

Ticket List (GET /api/tickets):

{
  "tickets": [
    {
      "id": 17933,
      "company_id": "12345",
      "company_compliance_level": "standard",
      "subject": "Library issue"
    },
    {
      "id": 17934,
      "company_id": "99999",
      "company_compliance_level": "cjis",
      "subject": "Sheriff office issue"
    }
  ],
  "total": 2
}

Company Data (GET /api/companies):

[
  {
    "account_number": "12345",
    "name": "Yoncalla Public Library",
    "compliance_level": "standard"
  },
  {
    "account_number": "99999",
    "name": "County Sheriff's Office",
    "compliance_level": "cjis"
  }
]

---

Configuration

Install spaCy Model (auto-installed during Brainhair setup):

python -m spacy download en_core_web_lg

No Additional Configuration Required: Filtering is automatic once company compliance levels are set in Codex.

---

Testing Compliance Levels

Test Standard Filtering:

from app.presidio_filter import filter_by_compliance_level

data = {
    "name": "John Smith",
    "email": "john@example.com",
    "phone": "555-123-4567",
    "address": "123 Main St, Portland, OR",
    "ssn": "123-45-6789",
    "credit_card": "4111-1111-1111-1111"
}

filtered = filter_by_compliance_level(data, 'standard')

# Result:
# {
#   "name": "John Smith",                    # ✅ Visible
#   "email": "john@example.com",             # ✅ Visible
#   "phone": "555-123-4567",                 # ✅ Visible
#   "address": "123 Main St, Portland, OR",  # ✅ Visible
#   "ssn": "<US_SSN>",                       # ❌ Redacted
#   "credit_card": "<CREDIT_CARD>"           # ❌ Redacted
# }

Test CJIS Filtering:

filtered = filter_by_compliance_level(data, 'cjis')

# Result:
# {
#   "name": "John S.",                       # ❌ Redacted to initial
#   "email": "<EMAIL_ADDRESS>",              # ❌ Redacted
#   "phone": "<PHONE_NUMBER>",               # ❌ Redacted
#   "address": "<LOCATION>",                 # ❌ Redacted
#   "ssn": "<US_SSN>",                       # ❌ Redacted
#   "credit_card": "<CREDIT_CARD>"           # ❌ Redacted
# }

Test PHI Filtering:

filtered = filter_by_compliance_level(data, 'phi')

# Result: Same as CJIS (maximum filtering)

---

Debugging

Check Entity Detection:

from presidio_analyzer import AnalyzerEngine

analyzer = AnalyzerEngine()

text = "Contact John Smith at john@example.com or 555-123-4567"
results = analyzer.analyze(text, language='en')

for result in results:
    detected_text = text[result.start:result.end]
    print(f"{result.entity_type}: {result.score:.2f} - {detected_text}")

# Output:
# PERSON: 0.85 - John Smith
# EMAIL_ADDRESS: 1.00 - john@example.com
# PHONE_NUMBER: 0.75 - 555-123-4567

View Active Entity List:

from app.presidio_filter import get_presidio_filter

presidio = get_presidio_filter()

print("Standard entities:", presidio.critical_entities)
print("CJIS entities:", presidio.cjis_entities)
print("HIPAA entities:", presidio.hipaa_entities)

---

Best Practices

1. Always set compliance level in Codex before using Brainhair with that company's data
2. Default to Standard unless the client has specific compliance requirements
3. Mark police departments as CJIS to prevent investigation data leakage
4. Mark healthcare providers as PHI to maintain HIPAA compliance
5. Review filtered output periodically to ensure appropriate redaction
6. Never bypass filtering for external AI queries

---

Troubleshooting

Problem: Names not being redacted for CJIS client - Check: Company compliance level in Codex (/companies/<id>) - Verify: Ticket response includes company_compliance_level: "cjis" - Test: Call filter_by_compliance_level(data, 'cjis') directly

Problem: Too much data redacted for Standard client - Check: Company compliance level isn't accidentally set to CJIS/PHI - Fix: Edit company in Codex, set to "Standard (Regular MSP Client)"

Problem: Dates being redacted - Note: Dates are intentionally NOT redacted in current implementation - Rationale: Operational data (ticket timelines, SLAs) requires date visibility - Override: If you need date redaction, add "DATE_TIME" to entity lists

Problem: New entity type not being detected - Check: spaCy model is installed (python -m spacy download en_core_web_lg) - Verify: Entity type is in Presidio's supported entities - Custom: Create custom recognizer for domain-specific entities

---

Security Considerations

Data Flow: 1. User asks Brainhair about ticket 2. Brainhair calls Codex API (internal service-to-service, no filtering) 3. Codex returns ticket with company_compliance_level 4. Brainhair filters data based on compliance level 5. Filtered data sent to Claude API (external server) 6. Claude response returned to user (no PII in prompt or response)

What This Protects: - ✅ PII/PHI leakage to Claude API (external Anthropic servers) - ✅ CJIS data leakage to external AI services - ✅ Healthcare data compliance (HIPAA) - ✅ Criminal justice data protection

What This Does NOT Protect: - ❌ Data already in Codex database (not filtered) - ❌ Service-to-service communication (internal, trusted) - ❌ Direct database access by technicians - ❌ Data in logs (separate logging controls needed)

Compliance Notes: - Filtering happens BEFORE data leaves internal network - External AI never sees unfiltered PII/PHI/CJIS data - Company compliance levels are administrator-controlled - Filtering is automatic and cannot be bypassed by users

---

Contract Alignment

Overview

The Contract Alignment tool allows Claude to analyze contract documents, extract billing terms, and automatically align Ledger billing settings to match the contract.

Workflow

┌─────────────────────────────────────────────────────────────┐
│            Contract Alignment Workflow                       │
│                                                               │
│  1. User provides contract text to Claude                     │
│     → "Here's the contract for Example Company..."              │
│                                                               │
│  2. Claude extracts billing terms using NLP                   │
│     → per_user_rate: $15.00                                   │
│     → hourly_rate: $150.00                                    │
│     → prepaid_hours_monthly: 4.0                              │
│                                                               │
│  3. Claude calls contract_tools.get_current_settings()        │
│     → Fetches current billing from Ledger                     │
│                                                               │
│  4. Claude calls contract_tools.compare_contract_to_settings()│
│     → Identifies discrepancies                                │
│     → Generates recommendations                               │
│                                                               │
│  5. Claude presents findings to user                          │
│     → "Found 3 discrepancies:"                                │
│     → "  - Per user cost: $12 → should be $15"                │
│     → "  - Prepaid hours: 0 → should be 4"                    │
│                                                               │
│  6. User approves alignment                                   │
│     → Claude calls contract_tools.align_settings()            │
│     → Requires approval (write operation)                     │
│                                                               │
│  7. Claude verifies alignment                                 │
│     → contract_tools.verify_alignment()                       │
│     → "✓ Alignment complete. All settings match contract."   │
└─────────────────────────────────────────────────────────────┘

Contract Terms Format

Claude extracts contract terms into structured format:

contract_terms = {
    "billing_method": "per_user",     # Or: "flat_fee", "per_device"
    "per_user_rate": 15.00,
    "per_workstation_rate": 75.00,
    "per_server_rate": 125.00,
    "hourly_rate": 150.00,
    "prepaid_hours_monthly": 4.0,
    "support_level": "All Inclusive",
    "included_users": 50,              # Optional
    "included_workstations": 25,       # Optional
    "flat_fee_amount": 5000.00,        # If flat_fee method
    "custom_items": [                  # Optional
        {"name": "Extra backup", "monthly_fee": 100.00}
    ]
}

API Functions

get_current_settings()

Get all current billing settings for comparison.

from contract_tools import get_current_settings

current = get_current_settings("GD-001")

# Returns:
# {
#   "account_number": "GD-001",
#   "billing_data": {...},         # Full billing calculation
#   "overrides": {...},             # Client-specific overrides
#   "available_plans": [...],       # All plans
#   "manual_assets": [...],         # Manual assets
#   "manual_users": [...],          # Manual users
#   "line_items": [...]             # Custom line items
# }

compare_contract_to_settings()

Compare extracted terms with current settings.

from contract_tools import compare_contract_to_settings

comparison = compare_contract_to_settings("GD-001", contract_terms)

# Returns:
# {
#   "account_number": "GD-001",
#   "discrepancies_found": 3,
#   "discrepancies": [
#     {
#       "field": "per_user_cost",
#       "contract_value": 15.00,
#       "current_value": 12.00,
#       "difference": 3.00
#     },
#     {
#       "field": "prepaid_hours_monthly",
#       "contract_value": 4.0,
#       "current_value": 0.0,
#       "difference": 4.0
#     },
#     {
#       "field": "support_level",
#       "contract_value": "All Inclusive",
#       "current_value": "Standard"
#     }
#   ],
#   "recommendations": [
#     "Set per_user_cost to $15.0",
#     "Set prepaid_hours_monthly to 4.0",
#     "Set support_level to 'All Inclusive'"
#   ],
#   "alignment_needed": true
# }

align_settings()

Apply adjustments to align with contract.

from contract_tools import align_settings

# Dry run (preview changes)
result = align_settings("GD-001", {
    "per_user_cost": 15.00,
    "prepaid_hours_monthly": 4.0,
    "support_level": "All Inclusive"
}, dry_run=True)

# Returns:
# {
#   "account_number": "GD-001",
#   "dry_run": true,
#   "would_apply": {...},
#   "message": "Dry run - no changes made"
# }

# Apply for real (requires approval)
result = align_settings("GD-001", {...}, dry_run=False)

# Returns:
# {
#   "account_number": "GD-001",
#   "dry_run": false,
#   "changes_applied": 3,
#   "results": [
#     "Applied 3 override(s)",
#     "Added line item: Contract Adjustment"
#   ],
#   "errors": [],
#   "success": true
# }

verify_alignment()

Verify settings match contract after alignment.

from contract_tools import verify_alignment

verification = verify_alignment("GD-001", contract_terms)

# If aligned:
# {
#   "aligned": true,
#   "message": "Billing settings match contract terms",
#   "details": {...}
# }

# If not aligned:
# {
#   "aligned": false,
#   "remaining_discrepancies": 1,
#   "details": {...}
# }

Example Conversation

User:

Here's the contract for Example Company:

"Managed Services Agreement
...
Monthly fee: $15 per user
Hourly rate: $150/hour
Included: 4 hours of support monthly
Support level: All Inclusive
..."

Claude:

I'll analyze the Example Company contract and check alignment with Ledger.

[Calls: get_current_settings("GD-001")]
[Calls: compare_contract_to_settings("GD-001", {...})]

I found 3 discrepancies between the contract and current billing settings:

1. **Per User Cost**
   - Contract: $15.00/user
   - Current: $12.00/user
   - Difference: +$3.00

2. **Prepaid Hours**
   - Contract: 4 hours/month
   - Current: 0 hours/month
   - Difference: +4 hours

3. **Support Level**
   - Contract: All Inclusive
   - Current: Standard

Would you like me to align the billing settings to match the contract?

User:

Yes, please align them.

Claude:

[Calls: align_settings("GD-001", {...}, dry_run=False)]
[APPROVAL REQUIRED - User approves]

✓ Alignment complete. Applied 3 changes:
  - Set per_user_cost to $15.00
  - Set prepaid_hours_monthly to 4.0
  - Set support_level to "All Inclusive"

[Calls: verify_alignment("GD-001", {...})]

✓ Verified: All billing settings now match the contract terms.

---

Configuration

instance/brainhair.conf

Not used currently. Configuration comes from environment variables.

Environment Variables (.flaskenv)

Auto-generated by config_manager.py from master_config.json.

# Flask Configuration
FLASK_APP=run.py
FLASK_ENV=production
FLASK_RUN_PORT=5050

# Service Configuration
SERVICE_NAME=brainhair
CORE_SERVICE_URL=http://localhost:5000
HELM_SERVICE_URL=http://localhost:5004

# Logging
LOG_LEVEL=INFO
ENABLE_JSON_LOGGING=true

# Database
SQLALCHEMY_DATABASE_URI=postgresql://brainhair_user:password@localhost:5432/brainhair_db

Database Configuration

Brainhair uses PostgreSQL for session and message persistence.

Connection String:

postgresql://brainhair_user:password@localhost:5432/brainhair_db

Setup:

cd hivematrix-brainhair
python init_db.py

Prompts: 1. PostgreSQL host (default: localhost) 2. PostgreSQL port (default: 5432) 3. Database name (default: brainhair_db) 4. Username (default: brainhair_user) 5. Password (generated if not provided)

Creates: - PostgreSQL user and database - Tables: chat_sessions, chat_messages - Indexes for performance

Claude Code Installation

NPX (Automatic):

# First invocation auto-installs
npx -y @anthropic-ai/claude-code --version

Global Install:

npm install -g @anthropic-ai/claude-code
which claude
# /usr/local/bin/claude or ~/.npm/_npx/.../claude

Verification:

claude --version
# claude-code version 0.x.x

spaCy Model Installation

Required for Presidio entity recognition:

# Install model (300MB download)
python -m spacy download en_core_web_lg

# Verify
python -c "import spacy; nlp = spacy.load('en_core_web_lg'); print('OK')"

---

Session Management

Session Lifecycle

1. Session Creation
   ├─ POST /api/chat (first message)
   ├─ Generate UUID for in-memory session
   ├─ Create or resume database session
   └─ Fetch context (ticket, client details)

2. Session Active
   ├─ Messages exchanged
   ├─ Claude Code subprocess spawned per message
   ├─ Conversation history maintained
   └─ Last activity timestamp updated

3. Session Idle
   ├─ No activity for 30 minutes
   ├─ Background cleanup thread detects idle
   ├─ Session removed from memory
   └─ Database session remains for history

4. Session Resumption
   ├─ POST /api/chat with db_session_id
   ├─ Loads messages from database
   ├─ Reconstructs conversation history
   └─ Continues conversation

ClaudeSessionManager

Singleton Instance:

from app.claude_session_manager import get_session_manager

session_manager = get_session_manager()

Create Session:

# New session
session_id = session_manager.create_session(
    user="john.technician",
    context={"ticket": "12345", "client": "ACME Corp"}
)

# Resume existing session
session_id = session_manager.create_session(
    user="john.technician",
    context={...},
    db_session_id="existing-db-uuid"
)

Get Session:

session = session_manager.get_session(session_id)
if session:
    for chunk in session.send_message_stream("Hello"):
        print(chunk)

Destroy Session:

session_manager.destroy_session(session_id)

Cleanup Thread

Background Cleanup:

def _cleanup_loop(self):
    """Background loop that runs cleanup periodically."""
    cleanup_interval = 300  # 5 minutes

    while not self._cleanup_stop_event.is_set():
        # Wait for interval or stop event
        if self._cleanup_stop_event.wait(timeout=cleanup_interval):
            break

        # Run cleanup (30 minute idle timeout)
        self.cleanup_idle_sessions(max_age_seconds=1800)

Idle Detection:

def cleanup_idle_sessions(self, max_age_seconds=1800):
    """Clean up sessions idle for more than max_age."""
    current_time = time.time()
    sessions_to_cleanup = []

    for session_id, session in self.sessions.items():
        idle_time = current_time - session.last_activity
        if idle_time > max_age_seconds:
            sessions_to_cleanup.append(session_id)

    for session_id in sessions_to_cleanup:
        self.destroy_session(session_id)

Last Activity Tracking:

def send_message_stream(self, message):
    """Send message and update last activity."""
    self.last_activity = time.time()  # Update timestamp

    # ... process message ...

Database Persistence

Session Storage:

# Create new database session
db_session = ChatSessionModel(
    id=str(uuid.uuid4()),
    user_id=self.user,
    user_name=user_display_name,
    ticket_number=context.get('ticket'),
    client_name=context.get('client')
)
db.session.add(db_session)
db.session.commit()

Message Storage:

# Save user message
user_msg = ChatMessage(
    session_id=self.db_session_id,
    role="user",
    content=message
)
db.session.add(user_msg)
db.session.commit()

# Save assistant message after response complete
assistant_msg = ChatMessage(
    session_id=self.db_session_id,
    role="assistant",
    content=response_text.strip()
)
db.session.add(assistant_msg)

# Update session timestamp
self.db_session.updated_at = datetime.utcnow()
db.session.commit()

Load from History:

if db_session_id:
    self.db_session = ChatSessionModel.query.get(db_session_id)
    if self.db_session:
        # Reconstruct conversation history
        for msg in self.db_session.messages:
            self.conversation_history.append({
                "role": msg.role,
                "content": msg.content
            })

---

Frontend Integration

Chat UI JavaScript

Send Message:

async function sendMessage(message) {
    const response = await fetch('/brainhair/api/chat', {
        method: 'POST',
        credentials: 'same-origin',  // CRITICAL for auth
        headers: {'Content-Type': 'application/json'},
        body: JSON.stringify({
            message: message,
            session_id: currentSessionId,
            ticket: currentTicket,
            client: currentClient
        })
    });

    const data = await response.json();

    // Start polling
    pollResponse(data.response_id, data.session_id);
}

Poll for Response:

let offset = 0;

async function pollResponse(responseId, sessionId) {
    while (true) {
        const res = await fetch(
            `/brainhair/api/chat/poll/${responseId}?offset=${offset}`,
            {credentials: 'same-origin'}
        );

        const data = await res.json();

        // Process new chunks
        for (const chunk of data.chunks) {
            handleChunk(chunk);
        }

        offset = data.offset;

        // Check if done
        if (data.done) {
            currentSessionId = data.session_id;
            break;
        }

        if (data.error) {
            showError(data.error);
            break;
        }

        // Poll every 500ms
        await new Promise(resolve => setTimeout(resolve, 500));
    }
}

Handle Chunks:

function handleChunk(chunk) {
    switch (chunk.type) {
        case 'chunk':
            // Append text to current message
            appendToMessage(chunk.content);
            break;

        case 'live_message':
            // Show status indicator
            showLiveMessage(chunk.content);
            break;

        case 'approval_request':
            // Show approval card
            showApprovalCard(chunk);
            break;

        case 'error':
            // Show error
            showError(chunk.message);
            break;
    }
}

Stop Response:

async function stopResponse(responseId) {
    await fetch(`/brainhair/api/chat/stop/${responseId}`, {
        method: 'POST',
        credentials: 'same-origin'
    });

    // Polling will detect done=true and stop
}

Context Management

Set Ticket Context:

function setTicket(ticketNumber) {
    currentTicket = ticketNumber;
    updateContextDisplay();

    // Optionally create new session with context
    currentSessionId = null;  // Force new session on next message
}

Set Client Context:

function setClient(clientName) {
    currentClient = clientName;
    updateContextDisplay();

    currentSessionId = null;
}

Display Context:

function updateContextDisplay() {
    const contextEl = document.getElementById('context-display');

    let parts = [];
    if (currentTicket) parts.push(`Ticket: ${currentTicket}`);
    if (currentClient) parts.push(`Client: ${currentClient}`);

    contextEl.textContent = parts.join(' | ') || 'No context set';
}

History UI

Load History List:

async function loadChatHistory() {
    const response = await fetch('/brainhair/api/chat/history?limit=50', {
        credentials: 'same-origin'
    });

    const data = await response.json();

    const historyList = document.getElementById('history-list');
    historyList.innerHTML = '';

    for (const session of data.sessions) {
        const item = document.createElement('div');
        item.className = 'history-item';
        item.innerHTML = `
            <div class="history-item__title">${escapeHtml(session.title || 'Untitled Chat')}</div>
            <div class="history-item__meta">
                ${formatDate(session.created_at)} | ${session.message_count} messages
            </div>
        `;
        item.onclick = () => loadSession(session.id);
        historyList.appendChild(item);
    }
}

Load Session:

async function loadSession(sessionId) {
    const response = await fetch(`/brainhair/api/chat/history/${sessionId}`, {
        credentials: 'same-origin'
    });

    const session = await response.json();

    // Clear current chat
    clearChat();

    // Display all messages
    for (const message of session.messages) {
        displayMessage(message.role, message.content);
    }

    // Set context
    currentTicket = session.ticket_number;
    currentClient = session.client_name;
    currentSessionId = null;  // Will create new in-memory session
    currentDbSessionId = session.id;  // Resume this DB session

    updateContextDisplay();
}

---

Development

Setup

# Clone repository
git clone https://github.com/skelhammer/hivematrix-brainhair
cd hivematrix-brainhair

# Run installation script
./install.sh

# Or manual install:
python3 -m venv pyenv
source pyenv/bin/activate
pip install -r requirements.txt

# Install spaCy model
python -m spacy download en_core_web_lg

# Install Claude Code
npx -y @anthropic-ai/claude-code

# Initialize database
python init_db.py

# Create .flaskenv (auto-generated by config_manager)
cd ../hivematrix-helm
python config_manager.py sync brainhair

Run Development Server

# Via Helm CLI (recommended)
cd hivematrix-helm
python cli.py start brainhair

# Or direct run
cd hivematrix-brainhair
source pyenv/bin/activate
python run.py

# Listens on http://127.0.0.1:5050
# Access via Nexus: https://localhost/brainhair

Testing

Test JWT Token:

cd hivematrix-helm
TOKEN=$(python create_test_token.py 2>/dev/null)

# Test chat endpoint
curl -X POST http://localhost:5050/brainhair/api/chat \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"message": "Hello"}'

Test Tool Directly:

# Set environment for tools
export BRAINHAIR_SESSION_ID="test-session"
export HIVEMATRIX_USER="test.user"
export HIVEMATRIX_CONTEXT='{"ticket":"12345"}'

# Test codex tools
python claude_tools/codex_tools.py

Test Presidio:

from app.presidio_filter import get_presidio_filter

presidio = get_presidio_filter()

text = "Contact John Smith at john@example.com or 555-123-4567"
filtered = presidio.anonymize_text(text)
print(filtered)
# "Contact John S. at <EMAIL_ADDRESS> or <PHONE_NUMBER>"

Debugging

View Logs:

cd hivematrix-helm
python logs_cli.py brainhair --tail 50

# Filter by level
python logs_cli.py brainhair --level ERROR --tail 100

Monitor Session Manager:

from app.claude_session_manager import get_session_manager

session_manager = get_session_manager()

# List active sessions
print(f"Active sessions: {len(session_manager.sessions)}")
for session_id, session in session_manager.sessions.items():
    print(f"  {session_id}: user={session.user}, messages={len(session.conversation_history)}")

# Cleanup idle sessions manually
cleaned = session_manager.cleanup_idle_sessions(max_age_seconds=600)  # 10 min
print(f"Cleaned up {cleaned} sessions")

Debug Claude Code Invocation:

# Run Claude Code manually to test
claude --model claude-sonnet-4-5 \
       --allowed-tools "Bash Read Grep Glob" \
       --permission-mode dontAsk \
       --verbose \
       --print \
       --output-format stream-json \
       "List companies from Codex"

Check Response Buffers:

from app.chat_routes import response_buffers

# List active responses
for response_id, buffer in response_buffers.items():
    print(f"{response_id}: {len(buffer['chunks'])} chunks, done={buffer['done']}")

---

Monitoring

Health Check

GET /health

Response:

{
  "service": "brainhair",
  "status": "healthy",
  "version": "1.0.0",
  "timestamp": "2025-11-22T10:30:00Z",
  "checks": {
    "disk_space": {
      "status": "healthy",
      "free_gb": 125.3,
      "usage_percent": 35.2
    },
    "core_service": {
      "status": "healthy",
      "url": "http://localhost:5000",
      "response_time_ms": 12
    },
    "codex_service": {
      "status": "healthy",
      "url": "http://localhost:5010",
      "response_time_ms": 8
    },
    "database": {
      "status": "healthy",
      "connection": "ok",
      "sessions_count": 42
    }
  }
}

Metrics

Session Metrics:

from models import ChatSession, ChatMessage
from extensions import db

# Total sessions
total_sessions = ChatSession.query.count()

# Active sessions
active_sessions = ChatSession.query.filter_by(is_active=True).count()

# Messages per session (average)
avg_messages = db.session.query(
    db.func.avg(db.func.count(ChatMessage.id))
).group_by(ChatMessage.session_id).scalar()

# Sessions by user
sessions_by_user = db.session.query(
    ChatSession.user_id,
    db.func.count(ChatSession.id)
).group_by(ChatSession.user_id).all()

Tool Usage Tracking:

# Most used tools
tool_usage = db.session.query(
    ChatMessage.tool_calls,
    db.func.count(ChatMessage.id)
).filter(ChatMessage.tool_calls.isnot(None)).all()

# Extract tool names from JSON
from collections import Counter
tool_counts = Counter()

for tool_calls, count in tool_usage:
    if tool_calls:
        for call in tool_calls:
            tool_counts[call.get('tool')] += count

Logging

Structured Logs:

{
  "timestamp": "2025-11-22T10:30:00Z",
  "level": "INFO",
  "service": "brainhair",
  "correlation_id": "req-uuid-123",
  "message": "Chat message received",
  "user": "john.technician",
  "session_id": "session-uuid-456",
  "ticket": "12345",
  "client": "ACME Corp"
}

Log to Helm:

from app.helm_logger import get_helm_logger

logger = get_helm_logger()

logger.info(
    "Claude Code session started",
    extra={
        'user': user,
        'session_id': session_id,
        'context': context
    }
)

Performance

Response Times:

import time

start = time.time()
for chunk in session.send_message_stream(message):
    # Stream chunk
    pass
duration = time.time() - start

logger.info(f"Response completed in {duration:.2f}s",
           extra={'duration_seconds': duration})

Database Query Performance:

-- Slow queries
SELECT schemaname, tablename, n_tup_ins, n_tup_upd, n_tup_del
FROM pg_stat_user_tables
WHERE schemaname = 'public';

-- Index usage
SELECT schemaname, tablename, indexname, idx_scan, idx_tup_read
FROM pg_stat_user_indexes
WHERE schemaname = 'public'
ORDER BY idx_scan DESC;

---

Troubleshooting

"Claude Code binary not found"

Symptoms:

RuntimeError: Claude Code binary not found. Run: npx -y @anthropic-ai/claude-code

Solutions:

1. Install Claude Code globally:

   npm install -g @anthropic-ai/claude-code
   which claude
   

2. Or use NPX (auto-installs):

   npx -y @anthropic-ai/claude-code --version
   

3. Check PATH:

   import shutil
   claude_bin = shutil.which('claude')
   print(claude_bin)
   

4. Manual fallback:

   # In claude_session_manager.py
   import glob
   npx_cache = os.path.expanduser('~/.npm/_npx/*/node_modules/.bin/claude')
   claude_bins = glob.glob(npx_cache)
   if claude_bins:
       claude_bin = claude_bins[0]
   

---

"No module named 'presidio_analyzer'"

Symptoms:

ModuleNotFoundError: No module named 'presidio_analyzer'

Solutions:

1. Install Presidio:

   pip install presidio-analyzer presidio-anonymizer
   

2. Install spaCy model:

   python -m spacy download en_core_web_lg
   

3. Verify:

   import spacy
   nlp = spacy.load("en_core_web_lg")
   print("OK")
   

---

"Session not found"

Symptoms: - GET /api/chat/poll/<id> returns 404 - Session mysteriously disappears

Causes: - Session cleaned up due to 30min idle timeout - Invalid response_id - Server restarted (in-memory sessions lost)

Solutions:

1. Check cleanup logs:

   python logs_cli.py brainhair --tail 100 | grep cleanup
   

2. Adjust idle timeout:

   # In claude_session_manager.py
   session_manager.cleanup_idle_sessions(max_age_seconds=3600)  # 1 hour
   

3. Resume from database:

   // Frontend: save db_session_id
   const sessionData = {
       message: message,
       db_session_id: currentDbSessionId  // Resume if in-memory session lost
   };
   

---

"Approval timeout"

Symptoms: - Tool waits 5 minutes then returns denied - Approval never appeared in UI

Causes: - Approval file not created - Polling not injecting file into stream - Session mismatch

Solutions:

1. Check /tmp files:

   ls -la /tmp/brainhair_*
   

2. Verify session ID match:

   # In tool:
   session_id = os.environ.get('BRAINHAIR_SESSION_ID')
   print(f"Tool session ID: {session_id}")
   
   # In polling:
   db_session_id = buffer.get('db_session_id')
   print(f"Buffer session ID: {db_session_id}")
   

3. Check polling injection:

   # In chat_routes.py:176
   logger.debug(f"Checking for approval files: /tmp/brainhair_approval_request_{db_session_id}_*.json")
   

---

"Authentication failed"

Symptoms: - Tools get 401 Unauthorized - service_client.call_service() fails

Causes: - Missing or expired service token - Core service unreachable - Incorrect service name

Solutions:

1. Test service token:

   from app.service_client import call_service
   
   response = call_service('codex', '/api/companies')
   print(response.status_code)
   

2. Check Core service:

   curl http://localhost:5000/health
   

3. Verify environment:

   echo $CORE_SERVICE_URL
   # Should be: http://localhost:5000
   

---

"Database connection failed"

Symptoms:

psycopg2.OperationalError: could not connect to server

Solutions:

1. Check PostgreSQL running:

   sudo systemctl status postgresql
   

2. Verify connection string:

   cat instance/brainhair.conf
   # Or check .flaskenv
   grep SQLALCHEMY .flaskenv
   

3. Test connection:

   psql -U brainhair_user -d brainhair_db -h localhost
   

4. Reinitialize:

   python init_db.py
   

---

Security

Threat Model

Protected: - PHI/CJIS data automatically filtered - Write operations require approval - All commands logged for audit - Services bind to localhost (no external access) - JWT authentication on all endpoints

Attack Vectors: - Path traversal in approval files → Mitigated by regex validation - Command injection via PowerShell → Mitigated by approval + logging - Session hijacking → Mitigated by JWT + ownership checks - SQL injection → Mitigated by SQLAlchemy ORM

Security Best Practices

1. Never Skip Approval for Write Operations

   # ✅ CORRECT
   def update_billing(...):
       approved = request_approval(...)
       if not approved:
           return {'error': 'Denied'}
       # Execute
   
   # ❌ WRONG
   def update_billing(...):
       # Execute directly
   

2. Always Validate IDs

   # ✅ CORRECT
   if not re.match(r'^[a-zA-Z0-9_-]+$', approval_id):
       return error
   
   # ❌ WRONG
   response_file = f"/tmp/brainhair_{user_input}.json"  # Path traversal!
   

3. Filter All External Data

   # ✅ CORRECT
   data = call_service('codex', '/api/companies').json()
   filtered = presidio.filter_phi(data)
   return jsonify(filtered)
   
   # ❌ WRONG
   return jsonify(data)  # May contain PHI!
   

4. Verify Session Ownership

   # ✅ CORRECT
   session = ChatSession.query.get(session_id)
   if session.user_id != current_user:
       return {'error': 'Access denied'}, 403
   
   # ❌ WRONG
   # Skip ownership check
   

5. Log All Sensitive Operations

   logger.info(f"Billing updated for {account_number}",
              extra={
                  'user': user,
                  'changes': changes,
                  'approved': True
              })
   

Audit Trail

All approvals and sensitive operations are logged to Helm:

-- Query approval logs
SELECT timestamp, service, message, extra_data
FROM log_entries
WHERE service = 'brainhair'
  AND message LIKE '%approval%'
ORDER BY timestamp DESC;

Example Log Entry:

{
  "timestamp": "2025-11-22T10:30:00Z",
  "service": "brainhair",
  "level": "INFO",
  "message": "Approval approved",
  "extra_data": {
    "approval_id": "approval-uuid-456",
    "user": "john.technician",
    "action": "Update billing plan",
    "details": {...},
    "approved": true
  }
}

Compliance

PHI (Protected Health Information): - All data filtered with Presidio before exposure to Claude - Names anonymized to "FirstName L." format - Emails, phones, SSNs redacted - Logs do not contain PHI

CJIS (Criminal Justice Information Systems): - CJIS mode available for law enforcement MSPs - Additional entity types filtered (driver licenses, passports) - Meets CJIS Security Policy requirements

Audit Requirements: - All commands logged with user, timestamp, details - Approvals tracked with who/what/when - Session access restricted to owner - Database changes traceable via timestamps

---

Summary

Brainhair is HiveMatrix's intelligent AI assistant, providing natural language access to all platform services through Claude Code integration. Key highlights:

• Claude Code CLI - Official Anthropic integration with streaming JSON
• Polling Architecture - Better proxy compatibility than SSE
• Approval Workflow - Human-in-the-loop for dangerous operations
• PHI/CJIS Filtering - Presidio-based data anonymization
• Contract Alignment - Automated billing alignment from contract analysis
• Session Management - Database persistence with idle cleanup
• Tool System - Extensible Python tools with service-to-service auth

Production Readiness: ⚠️ Beta - Claude Code integration: ✅ Functional - PHI filtering: ✅ Functional - Approval workflow: ✅ Functional - Datto RMM integration: ❌ Simulated (needs implementation) - Contract alignment: ✅ Functional

Next Steps: 1. Implement Datto RMM integration for real command execution 2. Add more specialized tools (network equipment, contracts, etc.) 3. Implement approval timeout notifications 4. Add session sharing for collaboration 5. Implement approval delegation (manager approval required)

---

See Also

Related Services

• Codex - Master Data - Companies, assets, users, tickets data source
• Ledger - Billing - Billing calculations and contract alignment
• KnowledgeTree - Knowledge Base - Context API for AI assistance
• Core - Authentication - JWT authentication and session management
• Helm - Orchestration - Service management and centralized logging
• All Services - Complete service inventory

Architecture & Design

• AI Assistant Architecture
• Claude Code Integration
• Service-to-Service Communication
• Approval Workflow Pattern
• PHI/CJIS Compliance

Configuration & Setup

• Installation Guide - Complete installation walkthrough
• Claude Code Installation - CLI installation and configuration
• Presidio Setup - PHI/CJIS filtering setup
• spaCy Model Installation - NER model setup
• Database Configuration

External Integrations

• Claude Code CLI - Official Anthropic CLI
• Microsoft Presidio - PII/PHI detection and anonymization
• spaCy en_core_web_lg - NER model for entity detection
• Anthropic API - Claude AI API documentation

Tools & Utilities

• Contract Alignment: /api/contract/analyze, /api/contract/align
• Approval Management: /api/approval/pending, /api/approval/approve
• Session Management: Session cleanup, title updates, history retrieval
• Service CLI: ../hivematrix-helm/cli.py start|stop|restart brainhair
• Log Viewer: ../hivematrix-helm/logs_cli.py brainhair --tail 50
• Claude Code: claude --model claude-sonnet-4-5 --allowed-tools "Bash Read Grep"

---

Questions or issues? Check the troubleshooting section or file an issue on GitHub.