# Psyche Messaging

Event-driven messaging for Psyche AI virtual assistants.

## Overview

`psyche-messaging` provides a Redis-based event bus aligned with the
`@oshun/event-bus` TypeScript patterns for cross-language compatibility and
consistent event-driven architecture across the Oshun platform.

## Features

- **Publish/Subscribe**: Pattern-based event subscription with wildcards
- **Event Persistence**: Events stored with configurable TTL for replay
- **Retry Handling**: Exponential, linear, or fixed backoff strategies
- **Dead Letter Queue**: Failed events stored for investigation and replay
- **Correlation Tracking**: Trace events across services with correlation IDs
- **Cross-Domain**: Route events between Psyche and other Oshun domains

## Installation

```bash
# Basic installation
poetry add psyche-messaging

# With Redis support
poetry add psyche-messaging[redis]
```

## Quick Start

### Publishing Events

```python
from psyche_messaging import EventBus, EventBusConfig, events

# Create and connect
config = EventBusConfig(host="localhost", port=6379)
bus = await create_event_bus(config)

# Publish using event factory
event = events.session_started(
    session_id="sess-1",
    user_id="user-1",
    persona_id="persona-1",
    channel="web",
)
await bus.publish(
    event.type,
    event.payload,
    correlation_id=event.correlation_id,
)

# Publish directly
await bus.publish(
    "psyche.session.started",
    {"session_id": "sess-1", "user_id": "user-1", "persona_id": "persona-1"},
    correlation_id="sess-1",
)
```

### Subscribing to Events

```python
from psyche_messaging import EventBus, EventContext, PsycheEventType

# Event handler with context
async def on_session_started(ctx: EventContext):
    payload = ctx.event.payload
    print(f"Session {payload['session_id']} started")

    # Acknowledge processing
    await ctx.ack()

# Subscribe to specific event
sub = await bus.subscribe(
    PsycheEventType.SESSION_STARTED,
    on_session_started,
)

# Pattern subscription (wildcards)
sub = await bus.psubscribe("psyche.session.*", on_session_started)

# Unsubscribe
await sub.unsubscribe()
```

### Event Context

The event context provides methods for handling events:

```python
async def handler(ctx: EventContext):
    try:
        # Process event
        result = await process(ctx.event.payload)

        # Acknowledge success
        await ctx.ack()

    except TemporaryError:
        # Retry with delay
        await ctx.nack(delay_ms=5000)

    except PermanentError as e:
        # Send to dead letter queue
        await ctx.dead_letter(str(e))

    # Publish follow-up event
    await ctx.publish(
        "psyche.process.completed",
        {"result": result},
    )

    # Reply to source domain
    await ctx.reply({"status": "ok"})
```

## Event Types

### Session Events

```python
events.session_started(session_id, user_id, persona_id, channel, settings)
events.session_ended(session_id, user_id, persona_id, reason, duration_ms, turn_count)
events.session_paused(session_id)
events.session_resumed(session_id)
```

### Avatar Events

```python
events.avatar_state_changed(session_id, avatar_id, changes, version)
events.avatar_expression_changed(session_id, avatar_id, expression)
events.avatar_animation_started(session_id, avatar_id, animation_name, speed)
events.avatar_animation_ended(session_id, avatar_id, animation_name)
events.avatar_created(avatar_id, persona_id, model_path, metadata)
events.avatar_trained(avatar_id, training_result)
```

### Voice Events

```python
events.voice_speaking_started(session_id, persona_id, text, duration_ms)
events.voice_speaking_ended(session_id, persona_id)
events.voice_listening_started(session_id)
events.voice_transcript_interim(session_id, transcript, confidence, language)
events.voice_transcript_final(session_id, transcript, confidence, language)
events.voice_response_generated(session_id, persona_id, text, audio_url, visemes)
```

### Memory Events

```python
events.memory_created(session_id, persona_id, memory_id, memory_type, content, importance)
events.memory_updated(session_id, persona_id, memory_id, memory_type, content, importance)
events.memory_retrieved(session_id, persona_id, memory_ids, query)
events.memory_consolidated(persona_id, source_memory_ids, consolidated_memory_id, summary)
```

### Persona Events

```python
events.persona_created(persona_id, name, version, config)
events.persona_updated(persona_id, name, version, changes)
events.persona_activated(persona_id, name)
events.persona_deactivated(persona_id, name)
```

### Tool Execution Events

```python
events.tool_execution_started(session_id, tool_call_id, tool_name, arguments)
events.tool_execution_completed(session_id, tool_call_id, tool_name, result, duration_ms)
events.tool_execution_failed(session_id, tool_call_id, tool_name, error, duration_ms)
```

### Conferencing Events

```python
events.participant_joined(session_id, room_id, participant_id, display_name, role)
events.participant_left(session_id, room_id, participant_id, display_name)
events.speaker_changed(session_id, room_id, speaker_id, previous_speaker_id)
events.screen_share_started(session_id, room_id, sharer_id, screen_id)
events.screen_share_ended(session_id, room_id, sharer_id)
events.room_started(session_id, room_id, name)
events.room_ended(session_id, room_id, duration_ms)
```

### Emotion Events

```python
events.emotion_detected(session_id, emotion, confidence, source, context)
events.sentiment_analyzed(session_id, sentiment, score, text)
```

### Error Events

```python
events.error_occurred(error_type, error_message, session_id, stack_trace, context)
```

## Configuration

### Event Bus Configuration

```python
from psyche_messaging import EventBusConfig

config = EventBusConfig(
    # Redis connection
    host="localhost",
    port=6379,
    password=None,
    db=0,
    tls=False,

    # Namespacing
    channel_prefix="psyche:events:",
    dead_letter_prefix="psyche:dlq:",

    # Event storage
    store_events=True,
    event_ttl_seconds=86400,  # 24 hours

    # Dead letter
    dead_letter_ttl_seconds=604800,  # 7 days
    dead_letter_max_entries=10000,
)
```

### Subscription Options

```python
from psyche_messaging import SubscriptionOptions, RetryConfig, RetryStrategy, DomainScope

options = SubscriptionOptions(
    # Consumer group (for competing consumers)
    consumer_group="my-service",
    consumer_name="instance-1",

    # Filtering
    source_filter=[DomainScope.PSYCHE, DomainScope.LILITH],

    # Concurrency
    concurrency=5,

    # Retry configuration
    retry_config=RetryConfig(
        max_attempts=3,
        strategy=RetryStrategy.EXPONENTIAL,
        initial_delay_ms=1000,
        max_delay_ms=30000,
        multiplier=2.0,
    ),

    # Dead letter
    enable_dead_letter=True,
    dead_letter_ttl_seconds=604800,
)
```

## Dead Letter Queue

### Viewing Failed Events

```python
# Get dead letter entries
entries = await bus.get_dead_letters(
    PsycheEventType.SESSION_STARTED,
    limit=100,
)

for entry in entries:
    print(f"Failed: {entry['event']['id']}")
    print(f"Reason: {entry['reason']}")
    print(f"Attempts: {entry['attempts']}")
```

### Replaying Events

```python
# Replay a specific event
success = await bus.replay_dead_letter(
    PsycheEventType.SESSION_STARTED,
    event_id="evt-123",
)
```

### Removing Events

```python
# Remove from dead letter queue
success = await bus.remove_dead_letter(
    PsycheEventType.SESSION_STARTED,
    event_id="evt-123",
)
```

## Cross-Domain Communication

Events can be routed to specific domains:

```python
from psyche_messaging import DomainScope

# Publish to specific domains
await bus.publish(
    "psyche.avatar.created",
    payload,
    source=DomainScope.PSYCHE,
    targets=[DomainScope.ISIS, DomainScope.BELLONA],
)

# Subscribe with source filter
options = SubscriptionOptions(
    source_filter=[DomainScope.ISIS],
)
sub = await bus.subscribe("isis.asset.generated", handler, options)
```

## Domain Scopes

| Scope       | Description           |
| ----------- | --------------------- |
| `YEMAYA`    | Creative Studio       |
| `LILITH`    | Consciousness         |
| `ISIS`      | Generative Factory    |
| `SOPHIA`    | Research & Knowledge  |
| `HATHOR`    | Worldbuilding         |
| `BELLONA`   | Engine Integration    |
| `APHRODITE` | Adult Live Streaming  |
| `OSHUN`     | Shared infrastructure |
| `PSYCHE`    | AI Virtual Assistant  |
| `VERITAS`   | AI News Agency        |

## Environment Variables

```bash
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_PASSWORD=
REDIS_DB=0
REDIS_TLS=false
PSYCHE_EVENT_PREFIX=psyche:events:
```

## Health Check

```python
health = await bus.health_check()
# {
#     "status": "healthy",
#     "connected": True,
#     "subscriptions": 5,
#     "pattern_subscriptions": 2,
#     "listener_running": True,
# }
```

## Integration with @oshun/event-bus

This library is designed to be compatible with the TypeScript
`@oshun/event-bus`:

- Same event envelope format
- Same domain scopes
- Same correlation/causation tracking
- Compatible retry strategies
- Interoperable dead letter queues

Events published from Python can be consumed by TypeScript services and vice
versa.

## Dependencies

- `pydantic` >= 2.5.0 - Event model validation
- `psyche-cache` - Redis client (local dependency)
- `redis` (optional) - For event bus functionality

## License

Proprietary - Oshun Platform