# BitByLaw Motia III - Developer Guide

> **For AI Assistants**: This document contains all critical patterns, conventions, and best practices. Read this first to understand the codebase structure and ensure consistency.

**Quick Navigation:**
- [Core Concepts](#core-concepts) - System architecture and patterns
- [Step Development](#step-development-best-practices) - How to create new steps
- [Services](#service-layer-patterns) - Reusable business logic
- [Integrations](#external-integrations) - xAI, EspoCRM, Advoware
- [Testing & Debugging](#testing-and-debugging)

---

## Project Status

**Migration:** ✅ 100% Complete (21/21 Steps migrated from Motia v0.17 → Motia III v1.0-RC)

**New to the project?** Start here:
1. [README.md](../README.md) - Project Overview & Quick Start
2. [MIGRATION_GUIDE.md](../MIGRATION_GUIDE.md) - Complete migration patterns
3. [ARCHITECTURE.md](ARCHITECTURE.md) - System design and architecture

---

## Core Concepts

### System Overview

**Architecture:**
- **Pure Python** (Motia III)
- **Event-Driven** with queue-based async processing
- **Redis-backed** distributed locking and caching
- **REST APIs** for webhooks and proxies

**Key Components:**
1. **Steps** (`steps/`) - Business logic handlers (HTTP, Queue, Cron)
2. **Services** (`services/`) - Shared API clients and utilities
3. **Configuration** (`iii-config.yaml`) - System setup

**Data Flow:**
```
Webhook (HTTP) → Queue Event → Event Handler → External APIs
                    ↓                ↓
              Redis Lock      Service Layer
                              (EspoCRM, Advoware, xAI)
```

---

## Step Development Best Practices

### File Naming Convention

**CRITICAL: Always use `_step.py` suffix!**

```
✅ CORRECT:
steps/vmh/webhook/document_create_api_step.py
steps/vmh/document_sync_event_step.py
steps/vmh/beteiligte_sync_cron_step.py

❌ WRONG:
steps/vmh/document_handler.py        # Missing _step.py
steps/vmh/sync.py                    # Missing _step.py
```

**Naming Pattern:**
- **Webhooks**: `{entity}_{action}_api_step.py`
  - Examples: `beteiligte_create_api_step.py`, `document_update_api_step.py`
- **Event Handlers**: `{entity}_sync_event_step.py`
  - Examples: `document_sync_event_step.py`, `beteiligte_sync_event_step.py`
- **Cron Jobs**: `{entity}_sync_cron_step.py`
  - Examples: `beteiligte_sync_cron_step.py`, `calendar_sync_cron_step.py`

### Step Template

**Complete step template with all required patterns:**

```python
"""Module-level docstring describing the step's purpose"""
from typing import Dict, Any
from motia import FlowContext, http, queue, cron, ApiRequest, ApiResponse


config = {
    "name": "Clear Human-Readable Name",
    "description": "Brief description of what this step does",
    "flows": ["flow-name"],  # Logical grouping
    "triggers": [
        # Pick ONE trigger type:
        http("POST", "/path/to/endpoint"),           # For webhooks
        # queue("topic.name"),                       # For event handlers
        # cron("0 */15 * * * *"),                    # For scheduled jobs
    ],
    "enqueues": ["next.topic"],  # Topics this step emits (optional)
}


async def handler(request: ApiRequest, ctx: FlowContext[Any]) -> ApiResponse:
    """
    Handler docstring explaining:
    - What triggers this handler
    - What it does
    - What events it emits
    """
    try:
        # 1. Log entry
        ctx.logger.info("=")="=" * 80)
        ctx.logger.info(f"🔄 STEP STARTED: {config['name']}")
        ctx.logger.info("=" * 80)
        
        # 2. Extract and validate input
        payload = request.body
        
        # 3. Business logic
        # ...
        
        # 4. Enqueue events if needed
        await ctx.enqueue({
            'topic': 'next.step',
            'data': {
                'entity_id': entity_id,
                'action': 'create'
            }
        })
        
        # 5. Log success
        ctx.logger.info("✅ Step completed successfully")
        
        # 6. Return response
        return ApiResponse(
            status_code=200,
            body={'success': True}
        )
        
    except Exception as e:
        # Always log errors with context
        ctx.logger.error(f"❌ Error in {config['name']}: {e}")
        ctx.logger.error(f"Payload: {request.body}")
        
        return ApiResponse(
            status_code=500,
            body={'success': False, 'error': str(e)}
        )
```

### Handler Signatures by Trigger Type

**HTTP Trigger (Webhooks, APIs):**
```python
from motia import ApiRequest, ApiResponse, FlowContext

async def handler(request: ApiRequest, ctx: FlowContext) -> ApiResponse:
    # Access: request.body, request.query_params, request.path_params
    return ApiResponse(status_code=200, body={...})
```

**Queue Trigger (Event Handlers):**
```python
from motia import FlowContext

async def handler(event_data: Dict[str, Any], ctx: FlowContext) -> None:
    # Process event_data
    # No return value
    pass
```

**Cron Trigger (Scheduled Jobs):**
```python
from motia import FlowContext

async def handler(input_data: None, ctx: FlowContext) -> None:
    # Cron jobs receive no input
    # No return value
    pass
```

### Logging Best Practices

**ALWAYS use `ctx.logger`, NEVER use `print()` or module-level `logger`:**

```python
# ✅ CORRECT: Context-aware logging
ctx.logger.info("Processing started")
ctx.logger.debug(f"Data: {data}")
ctx.logger.warn("Skipping invalid entry")
ctx.logger.error(f"Failed: {e}")

# ❌ WRONG: Direct print or module logger
print("Processing started")              # Not visible in iii logs
logger.info("Processing started")        # Loses context
```

**Log Levels:**
- `info()` - Normal operations (start, success, counts)
- `debug()` - Detailed data dumps (payloads, responses)
- `warn()` - Non-critical issues (skipped items, fallbacks)
- `error()` - Failures requiring attention

**Log Structure:**
```python
# Section headers with visual separators
ctx.logger.info("=" * 80)
ctx.logger.info("🔄 SYNC HANDLER STARTED")
ctx.logger.info("=" * 80)
ctx.logger.info(f"Entity Type: {entity_type}")
ctx.logger.info(f"Action: {action.upper()}")
ctx.logger.info(f"Document ID: {entity_id}")
ctx.logger.info("=" * 80)

# Use emojis for visual scanning
ctx.logger.info("📥 Downloading file...")
ctx.logger.info("✅ Downloaded 1024 bytes")
ctx.logger.error("❌ Upload failed")
ctx.logger.warn("⚠️  No collections found")
```

### Event Topics Naming

**Pattern:** `{module}.{entity}.{action}`

```
✅ Examples:
vmh.document.create
vmh.document.update
vmh.document.delete
vmh.beteiligte.create
calendar.sync.employee
advoware.proxy.response

❌ Avoid:
document-create          # Use dots, not dashes
DocumentCreate           # Use lowercase
create_document          # Wrong order
```

---

## Service Layer Patterns

### Service Base Class Pattern

**All services should follow this pattern:**

```python
"""Service docstring"""
import logging
from typing import Optional
from services.logging_utils import get_service_logger

logger = logging.getLogger(__name__)


class MyService:
    """Service for interacting with External API"""
    
    def __init__(self, context=None):
        """
        Initialize service.
        
        Args:
            context: Optional Motia FlowContext for logging
        """
        self.context = context
        self.logger = get_service_logger('my_service', context)
        self._session = None
        
        # Load config from env
        self.api_key = os.getenv('MY_API_KEY', '')
        if not self.api_key:
            raise ValueError("MY_API_KEY not configured")
        
        self.logger.info("MyService initialized")
    
    def _log(self, message: str, level: str = 'info') -> None:
        """Internal logging helper"""
        log_func = getattr(self.logger, level, self.logger.info)
        log_func(message)
    
    async def _get_session(self):
        """Lazy session initialization"""
        if self._session is None or self._session.closed:
            self._session = aiohttp.ClientSession()
        return self._session
    
    async def close(self) -> None:
        """Cleanup resources"""
        if self._session and not self._session.closed:
            await self._session.close()
```

### Sync Utilities Pattern

**For bidirectional sync operations, inherit from `BaseSyncUtils`:**

```python
from services.sync_utils_base import BaseSyncUtils

class MyEntitySync(BaseSyncUtils):
    """Sync utilities for MyEntity"""
    
    def _get_lock_key(self, entity_id: str) -> str:
        """Required: Define lock key pattern"""
        return f"sync_lock:myentity:{entity_id}"
    
    async def should_sync(self, entity: Dict) -> Tuple[bool, str]:
        """
        Decide if sync is needed.
        
        Returns:
            (needs_sync: bool, reason: str)
        """
        # Implementation
        pass
```

**Base class provides:**
- `_log()` - Context-aware logging
- `_acquire_redis_lock()` - Distributed locking
- `_release_redis_lock()` - Lock cleanup
- `self.espocrm` - EspoCRM API client
- `self.redis` - Redis client
- `self.context` - Motia context
- `self.logger` - Integration logger

---

## External Integrations

### xAI Collections Integration

**Status:** ✅ Fully Implemented

**Service:** `services/xai_service.py`

**Environment Variables:**
```bash
XAI_API_KEY=xai-...              # For file uploads (api.x.ai)
XAI_MANAGEMENT_KEY=xai-token-... # For collections (management-api.x.ai)
```

**Usage Example:**
```python
from services.xai_service import XAIService

xai = XAIService(ctx)

# Upload file
file_id = await xai.upload_file(
    file_content=bytes_data,
    filename="document.pdf",
    mime_type="application/pdf"
)

# Add to collection
await xai.add_to_collection("collection_id", file_id)

# Add to multiple collections
added = await xai.add_to_collections(["col1", "col2"], file_id)

# Remove from collection
await xai.remove_from_collection("collection_id", file_id)
```

**Architecture:**
- Files are uploaded ONCE to Files API (`api.x.ai/v1/files`)
- Same `file_id` can be added to MULTIPLE collections
- Removing from collection does NOT delete the file (may be used elsewhere)
- Hash-based change detection prevents unnecessary reuploads

**Document Sync Flow:**
```
1. EspoCRM Webhook → vmh.document.{create|update|delete}
2. Document Sync Handler:
   a. Acquire distributed lock (prevents duplicate syncs)
   b. Load document from EspoCRM
   c. Check if sync needed:
      - dateiStatus = "Neu" or "Geändert" → SYNC
      - Hash changed → SYNC
      - Entity has xAI collections → SYNC
   d. Download file from EspoCRM
   e. Calculate MD5 hash
   f. Upload to xAI (or reuse existing file_id)
   g. Add to required collections
   h. Update EspoCRM metadata (xaiFileId, xaiCollections, xaiSyncedHash)
   i. Release lock
3. Delete: Remove from collections (keep file)
```

**EspoCRM Fields (CDokumente):**
```python
{
    'xaiId': 'file-abc123',                    # xAI file_id
    'xaiCollections': ['col1', 'col2'],        # Array of collection IDs
    'xaiSyncedHash': 'abc123def456',           # MD5 at last sync
    'fileStatus': 'synced',                    # Status: neu, geändert, synced
    'md5sum': 'abc123def456',                  # Current file hash
    'sha256': 'def456...',                     # SHA-256 (optional)
}
```

### EspoCRM Integration

**Service:** `services/espocrm.py`

**Environment Variables:**
```bash
ESPOCRM_API_BASE_URL=https://crm.bitbylaw.com/api/v1
ESPOCRM_API_KEY=your-api-key
ESPOCRM_API_TIMEOUT_SECONDS=30
```

**Usage:**
```python
from services.espocrm import EspoCRMAPI

espocrm = EspoCRMAPI(ctx)

# Get entity
entity = await espocrm.get_entity('CDokumente', entity_id)

# Update entity
await espocrm.update_entity('CDokumente', entity_id, {
    'xaiId': file_id,
    'fileStatus': 'synced'
})

# List entities
result = await espocrm.list_entities(
    'CDokumente',
    where=[{'type': 'equals', 'attribute': 'fileStatus', 'value': 'neu'}],
    select='id,name,fileStatus',
    max_size=50
)

# Download attachment
file_bytes = await espocrm.download_attachment(attachment_id)
```

### Advoware Integration

**Service:** `services/advoware_service.py`

**Authentication:** HMAC-512 with token caching

**Environment Variables:**
```bash
ADVOWARE_API_BASE_URL=https://api.advoware.de
ADVOWARE_API_KEY=your-key
ADVOWARE_API_SECRET=your-secret
REDIS_HOST=localhost
REDIS_PORT=6379
```

**Proxy Endpoints:**
- `GET /advoware/proxy?endpoint={path}` - Proxy GET requests
- `POST /advoware/proxy` - Proxy POST requests
- `PUT /advoware/proxy` - Proxy PUT requests
- `DELETE /advoware/proxy` - Proxy DELETE requests

---

## Testing and Debugging

### Start System

```bash
# Start iii Engine
cd /opt/motia-iii/bitbylaw
/opt/bin/iii -c iii-config.yaml

# Start iii Console (Web UI)
/opt/bin/iii-console --enable-flow --host 0.0.0.0 --port 3113 \
  --engine-host 192.168.67.233 --engine-port 3111 --ws-port 3114
```

### Check Registered Steps

```bash
curl http://localhost:3111/_console/functions | python3 -m json.tool
```

### Test HTTP Endpoint

```bash
# Test document webhook
curl -X POST "http://localhost:3111/vmh/webhook/document/create" \
  -H "Content-Type: application/json" \
  -d '[{"id": "test123", "entityType": "CDokumente"}]'

# Test advoware proxy
curl "http://localhost:3111/advoware/proxy?endpoint=employees"
```

### Manually Trigger Cron

```bash
curl -X POST "http://localhost:3111/_console/cron/trigger" \
  -H "Content-Type: application/json" \
  -d '{"function_id": "steps::VMH Beteiligte Sync Cron::trigger::0"}'
```

### View Logs

```bash
# Live logs via journalctl
journalctl -u motia-iii -f

# Search for specific step
journalctl --since "today" | grep -i "document sync"

# Check for errors
tail -100 /opt/motia-iii/bitbylaw/iii_new.log | grep -i error
```

### Common Issues

**Step not showing up:**
1. Check file naming: Must end with `_step.py`
2. Check for import errors: `grep -i "importerror\|traceback" iii.log`
3. Verify `config` dict is present
4. Restart iii engine

**Redis connection failed:**
- Check `REDIS_HOST` and `REDIS_PORT` environment variables
- Verify Redis is running: `redis-cli ping`
- Service will work without Redis but with warnings

**AttributeError '_log' not found:**
- Ensure service inherits from `BaseSyncUtils` OR
- Implement `_log()` method manually

---

## Key Patterns Summary

### ✅ DO

- **Always** use `_step.py` suffix for step files
- **Always** use `ctx.logger` for logging (never `print`)
- **Always** wrap handlers in try/except with error logging
- **Always** use visual separators in logs (`"=" * 80`)
- **Always** return `ApiResponse` from HTTP handlers
- **Always** document what events a step emits
- **Always** use distributed locks for sync operations
- **Always** calculate hashes for change detection

### ❌ DON'T

- **Don't** use module-level `logger` in steps
- **Don't** forget `async` on handler functions
- **Don't** use blocking I/O (use `aiohttp`, not `requests`)
- **Don't** return values from queue/cron handlers
- **Don't** hardcode credentials (use environment variables)
- **Don't** skip lock cleanup in `finally` blocks
- **Don't** use `print()` for logging

---

## Module Documentation

---

## Module Documentation

### Steps

**Advoware Proxy** ([Module README](../steps/advoware_proxy/README.md))
- Universal HTTP proxy with HMAC-512 authentication
- Endpoints: GET, POST, PUT, DELETE
- Redis token caching

**Calendar Sync** ([Module README](../steps/advoware_cal_sync/README.md))
- Bidirectional Advoware ↔ Google Calendar sync
- Cron: Every 15 minutes
- API trigger: `/advoware/calendar/sync`

**VMH Integration** ([Module README](../steps/vmh/README.md))
- EspoCRM ↔ Advoware bidirectional sync
- Webhooks: Beteiligte, Bankverbindungen, Documents
- xAI Collections integration for documents

### Services

| Service | Purpose | Config |
|---------|---------|--------|
| `xai_service.py` | xAI file uploads & collections | `XAI_API_KEY`, `XAI_MANAGEMENT_KEY` |
| `espocrm.py` | EspoCRM REST API client | `ESPOCRM_API_BASE_URL`, `ESPOCRM_API_KEY` |
| `advoware_service.py` | Advoware API with HMAC auth | `ADVOWARE_API_KEY`, `ADVOWARE_API_SECRET` |
| `document_sync_utils.py` | Document sync logic | Redis connection |
| `beteiligte_sync_utils.py` | Beteiligte sync logic | Redis connection |
| `sync_utils_base.py` | Base class for sync utils | - |

---

## Quick Reference

### Environment Variables

**Required:**
```bash
# EspoCRM
ESPOCRM_API_BASE_URL=https://crm.bitbylaw.com/api/v1
ESPOCRM_API_KEY=your-key

# Advoware
ADVOWARE_API_BASE_URL=https://api.advoware.de
ADVOWARE_API_KEY=your-key
ADVOWARE_API_SECRET=your-secret

# xAI
XAI_API_KEY=xai-...
XAI_MANAGEMENT_KEY=xai-token-...

# Redis
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB_ADVOWARE_CACHE=1
```

**Optional:**
```bash
ESPOCRM_API_TIMEOUT_SECONDS=30
ESPOCRM_METADATA_TTL_SECONDS=300
```

### File Structure

```
bitbylaw/
├── iii-config.yaml           # Motia III configuration
├── pyproject.toml            # Python dependencies (uv)
├── steps/                    # Business logic
│   ├── advoware_proxy/
│   ├── advoware_cal_sync/
│   └── vmh/
│       ├── webhook/          # HTTP webhook handlers
│       │   ├── *_create_api_step.py
│       │   ├── *_update_api_step.py
│       │   └── *_delete_api_step.py
│       ├── *_sync_event_step.py   # Queue event handlers
│       └── *_sync_cron_step.py    # Scheduled jobs
├── services/                 # Shared services
│   ├── xai_service.py
│   ├── espocrm.py
│   ├── advoware_service.py
│   ├── *_sync_utils.py
│   ├── sync_utils_base.py
│   ├── logging_utils.py
│   └── exceptions.py
├── docs/                     # Documentation
│   ├── INDEX.md             # This file
│   ├── ARCHITECTURE.md
│   └── DOCUMENT_SYNC_XAI_STATUS.md
└── tests/                    # Test scripts
    └── test_xai_collections_api.py
```

### Motia III vs Old Motia

| Old Motia v0.17 | Motia III v1.0-RC |
|-----------------|-------------------|
| `type: 'api'` | `triggers: [http()]` |
| `type: 'event'` | `triggers: [queue()]` |
| `type: 'cron'` | `triggers: [cron()]` |
| `context.emit()` | `ctx.enqueue()` |
| `emits: [...]` | `enqueues: [...]` |
| `subscribes: [...]` | `triggers: [queue('topic')]` |
| 5-field cron | 6-field cron (seconds first) |
| `context.logger` | `ctx.logger` |
| Motia Workbench | iii Console |
| Node.js + Python | Pure Python |

### Cron Syntax

**6 fields (Motia III):** `second minute hour day month weekday`

```
0 */15 * * * *     # Every 15 minutes
0 0 */6 * * *      # Every 6 hours
0 0 2 * * *        # Daily at 2 AM
0 30 9 * * 1-5     # Monday-Friday at 9:30 AM
```

---

## Additional Resources

### Documentation
- [MIGRATION_GUIDE.md](../MIGRATION_GUIDE.md) - v0.17 → v1.0 migration
- [MIGRATION_STATUS.md](../MIGRATION_STATUS.md) - Migration progress
- [ARCHITECTURE.md](ARCHITECTURE.md) - System design
- [DOCUMENT_SYNC_XAI_STATUS.md](DOCUMENT_SYNC_XAI_STATUS.md) - xAI integration details

### External Resources
- [Motia III Documentation](https://iii.dev)
- [Python SDK](https://pypi.org/project/motia/)
- [xAI API Docs](https://docs.x.ai/)
- [EspoCRM API](https://docs.espocrm.com/development/api/)
- [Redis Documentation](https://redis.io/documentation)

### Support & Troubleshooting

| Issue | Solution |
|-------|----------|
| Step not registered | Check `_step.py` suffix, restart iii engine |
| Import errors | Check logs: `grep -i importerror iii.log` |
| Redis unavailable | Service works with warnings, check `REDIS_HOST` |
| `'_log' not found` | Inherit from `BaseSyncUtils` or implement `_log()` |
| Webhook not triggering | Verify endpoint in iii Console, check EspoCRM config |
| xAI upload fails | Verify `XAI_API_KEY` and `XAI_MANAGEMENT_KEY` |

---

**Last Updated:** 2026-03-08  
**Migration Status:** ✅ Complete  
**xAI Integration:** ✅ Implemented