Hiddenden/loyal_companion
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

work in progress.

Browse Source
This commit is contained in:
Latte
2026-01-14 18:35:57 +01:00
parent dbd534d860
commit 7871ef1b1e
11 changed files with 1029 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
CLAUDE.md
View File

@@ -185,3 +185,39 @@ Optional:
### Admin commands
- `!setusername @user <name>` - Set name for another user
- `!teachbot @user <fact>` - Add a fact about a user
## Development Guidelines
### When Adding New Features
1. **Always write tests** - New services need corresponding test files in `tests/`
2. **Update documentation** - README.md and relevant docs/ files must be updated
3. **Update CLAUDE.md** - Add new services, models, and config options here
4. **Follow existing patterns** - Match the style of existing services
### Planned Features (In Progress)
The following features are being implemented:
1. **Attachment Pattern Tracking** (`attachment_service.py`)
- Detect anxious/avoidant/disorganized patterns
- Adapt responses based on attachment state
- Track what helps regulate each person
2. **Grief Journey Tracking** (`grief_service.py`)
- Track grief context and phase
- Recognize anniversaries and hard dates
- Adjust support style based on grief phase
3. **Grounding & Coping Tools** (`grounding_service.py`)
- Breathing exercises, sensory grounding
- Spiral detection and intervention
- Session pacing and intensity tracking
4. **Enhanced Support Memory**
- Learn HOW someone wants to be supported
- Track effective vs ineffective approaches
- Remember comfort topics for breaks
5. **Communication Style Matching**
- Energy matching (playful vs serious)
- Directness calibration
- Real-time tone adaptation

53
schema.sql
View File

@@ -260,3 +260,56 @@ ALTER TABLE user_facts ADD COLUMN IF NOT EXISTS temporal_relevance VARCHAR(20);
ALTER TABLE user_facts ADD COLUMN IF NOT EXISTS expiry_date TIMESTAMPTZ;
ALTER TABLE user_facts ADD COLUMN IF NOT EXISTS extracted_from_message_id BIGINT;
ALTER TABLE user_facts ADD COLUMN IF NOT EXISTS extraction_context TEXT;
-- =====================================================
-- ATTACHMENT TRACKING TABLES
-- =====================================================
-- User attachment profiles (tracks attachment patterns per user)
CREATE TABLE IF NOT EXISTS user_attachment_profiles (
id BIGSERIAL PRIMARY KEY,
user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
guild_id BIGINT, -- NULL = global profile
primary_style VARCHAR(20) DEFAULT 'unknown', -- secure, anxious, avoidant, disorganized, unknown
style_confidence FLOAT DEFAULT 0.0, -- 0.0 to 1.0
current_state VARCHAR(20) DEFAULT 'regulated', -- regulated, activated, mixed
state_intensity FLOAT DEFAULT 0.0, -- 0.0 to 1.0
anxious_indicators INTEGER DEFAULT 0, -- running count of anxious pattern matches
avoidant_indicators INTEGER DEFAULT 0, -- running count of avoidant pattern matches
secure_indicators INTEGER DEFAULT 0, -- running count of secure pattern matches
disorganized_indicators INTEGER DEFAULT 0, -- running count of disorganized pattern matches
last_activation_at TIMESTAMPTZ, -- when attachment system was last activated
activation_count INTEGER DEFAULT 0, -- total activations
activation_triggers JSONB DEFAULT '[]', -- learned triggers that activate attachment
effective_responses JSONB DEFAULT '[]', -- response styles that helped regulate
ineffective_responses JSONB DEFAULT '[]', -- response styles that didn't help
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(user_id, guild_id)
);
CREATE INDEX IF NOT EXISTS ix_user_attachment_profiles_user_id ON user_attachment_profiles(user_id);
CREATE INDEX IF NOT EXISTS ix_user_attachment_profiles_guild_id ON user_attachment_profiles(guild_id);
CREATE INDEX IF NOT EXISTS ix_user_attachment_profiles_primary_style ON user_attachment_profiles(primary_style);
-- Attachment events (logs attachment-related events for learning)
CREATE TABLE IF NOT EXISTS attachment_events (
id BIGSERIAL PRIMARY KEY,
user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
guild_id BIGINT,
event_type VARCHAR(50) NOT NULL, -- activation, regulation, escalation, etc.
detected_style VARCHAR(20), -- anxious, avoidant, disorganized, mixed
intensity FLOAT DEFAULT 0.0, -- 0.0 to 1.0
trigger_message TEXT, -- the message that triggered the event (truncated)
trigger_indicators JSONB DEFAULT '[]', -- patterns that matched
response_style VARCHAR(50), -- how Bartender responded
outcome VARCHAR(20), -- helpful, neutral, unhelpful (set after follow-up)
notes TEXT, -- any additional context
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS ix_attachment_events_user_id ON attachment_events(user_id);
CREATE INDEX IF NOT EXISTS ix_attachment_events_guild_id ON attachment_events(guild_id);
CREATE INDEX IF NOT EXISTS ix_attachment_events_event_type ON attachment_events(event_type);
CREATE INDEX IF NOT EXISTS ix_attachment_events_created_at ON attachment_events(created_at);

17
src/loyal_companion/cogs/ai_chat.py
View File

@@ -9,6 +9,7 @@ from discord.ext import commands
from loyal_companion.config import settings
from loyal_companion.services import (
AIService,
AttachmentService,
CommunicationStyleService,
ConversationManager,
FactExtractionService,
@@ -459,11 +460,12 @@ class AIChatCog(commands.Cog):
# Get context about mentioned users
mentioned_users_context = self._get_mentioned_users_context(message)
# Get Living AI context (mood, relationship, style, opinions)
# Get Living AI context (mood, relationship, style, opinions, attachment)
mood = None
relationship_data = None
communication_style = None
relevant_opinions = None
attachment_context = None
if settings.living_ai_enabled:
if settings.mood_enabled:
@@ -486,13 +488,24 @@ class AIChatCog(commands.Cog):
topics, guild_id
)
if settings.attachment_tracking_enabled:
attachment_service = AttachmentService(session)
attachment_context = await attachment_service.analyze_message(
user=user,
message_content=user_message,
guild_id=guild_id,
)
# Build system prompt with personality context
if settings.living_ai_enabled and (mood or relationship_data or communication_style):
if settings.living_ai_enabled and (
mood or relationship_data or communication_style or attachment_context
):
system_prompt = self.ai_service.get_enhanced_system_prompt(
mood=mood,
relationship=relationship_data,
communication_style=communication_style,
bot_opinions=relevant_opinions,
attachment=attachment_context,
)
else:
system_prompt = self.ai_service.get_system_prompt()

8
src/loyal_companion/config.py
View File

@@ -103,6 +103,14 @@ class Settings(BaseSettings):
opinion_formation_enabled: bool = Field(True, description="Enable bot opinion formation")
style_learning_enabled: bool = Field(True, description="Enable communication style learning")
# Attachment Tracking Configuration
attachment_tracking_enabled: bool = Field(
True, description="Enable attachment pattern tracking"
)
attachment_reflection_enabled: bool = Field(
True, description="Allow reflecting attachment patterns at close friend level"
)
# Mood System Settings
mood_decay_rate: float = Field(
0.05, ge=0.0, le=1.0, description="How fast mood returns to neutral per hour"

3
src/loyal_companion/models/__init__.py
View File

@@ -12,9 +12,11 @@ from .living_ai import (
UserCommunicationStyle,
UserRelationship,
)
from .support import AttachmentEvent, UserAttachmentProfile
from .user import User, UserFact, UserPreference
__all__ = [
"AttachmentEvent",
"Base",
"BotOpinion",
"BotState",
@@ -26,6 +28,7 @@ __all__ = [
"MoodHistory",
"ScheduledEvent",
"User",
"UserAttachmentProfile",
"UserCommunicationStyle",
"UserFact",
"UserPreference",

105
src/loyal_companion/models/support.py Normal file
View File

@@ -0,0 +1,105 @@
"""Support-focused models - attachment, grief, grounding."""
from datetime import datetime
from typing import TYPE_CHECKING
from sqlalchemy import BigInteger, DateTime, Float, ForeignKey, String, Text, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column, relationship
from .base import Base, PortableJSON, utc_now
if TYPE_CHECKING:
from .user import User
class UserAttachmentProfile(Base):
"""Tracks attachment patterns and states for each user.
Attachment styles:
- secure: comfortable with intimacy and independence
- anxious: fears abandonment, seeks reassurance
- avoidant: uncomfortable with closeness, values independence
- disorganized: conflicting needs, push-pull patterns
Attachment states:
- regulated: baseline, not activated
- activated: attachment system triggered (anxiety, withdrawal, etc.)
- mixed: showing conflicting patterns
"""
__tablename__ = "user_attachment_profiles"
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
guild_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
# Primary attachment style (learned over time)
primary_style: Mapped[str] = mapped_column(
String(20), default="unknown"
) # secure, anxious, avoidant, disorganized, unknown
style_confidence: Mapped[float] = mapped_column(Float, default=0.0) # 0-1
# Current state (changes per conversation)
current_state: Mapped[str] = mapped_column(
String(20), default="regulated"
) # regulated, activated, mixed
state_intensity: Mapped[float] = mapped_column(Float, default=0.0) # 0-1
# Indicator counts (used to determine primary style)
anxious_indicators: Mapped[int] = mapped_column(default=0)
avoidant_indicators: Mapped[int] = mapped_column(default=0)
secure_indicators: Mapped[int] = mapped_column(default=0)
disorganized_indicators: Mapped[int] = mapped_column(default=0)
# Activation tracking
last_activation_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
activation_count: Mapped[int] = mapped_column(default=0)
# Learned patterns (what triggers them, what helps)
activation_triggers: Mapped[list] = mapped_column(PortableJSON, default=list)
effective_responses: Mapped[list] = mapped_column(PortableJSON, default=list)
ineffective_responses: Mapped[list] = mapped_column(PortableJSON, default=list)
# Timestamps
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utc_now)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utc_now)
# Relationship
user: Mapped["User"] = relationship(back_populates="attachment_profile")
__table_args__ = (UniqueConstraint("user_id", "guild_id"),)
class AttachmentEvent(Base):
"""Records attachment-related events for learning and reflection.
Tracks when attachment patterns are detected, what triggered them,
and how the user responded to different support approaches.
"""
__tablename__ = "attachment_events"
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
guild_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
# Event details
event_type: Mapped[str] = mapped_column(String(50)) # activation, regulation, pattern_detected
detected_style: Mapped[str] = mapped_column(String(20)) # anxious, avoidant, etc.
intensity: Mapped[float] = mapped_column(Float, default=0.5)
# Context
trigger_message: Mapped[str | None] = mapped_column(Text, nullable=True)
trigger_indicators: Mapped[list] = mapped_column(PortableJSON, default=list)
# Response tracking
response_given: Mapped[str | None] = mapped_column(Text, nullable=True)
response_style: Mapped[str | None] = mapped_column(String(50), nullable=True)
was_helpful: Mapped[bool | None] = mapped_column(default=None) # learned from follow-up
# Timestamp
occurred_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=utc_now, index=True
)

4
src/loyal_companion/models/user.py
View File

@@ -21,6 +21,7 @@ if TYPE_CHECKING:
from .conversation import Conversation, Message
from .guild import GuildMember
from .living_ai import ScheduledEvent, UserCommunicationStyle, UserRelationship
from .support import UserAttachmentProfile
class User(Base):
@@ -62,6 +63,9 @@ class User(Base):
scheduled_events: Mapped[list["ScheduledEvent"]] = relationship(
back_populates="user", cascade="all, delete-orphan"
)
attachment_profile: Mapped[list["UserAttachmentProfile"]] = relationship(
back_populates="user", cascade="all, delete-orphan"
)
@property
def display_name(self) -> str:

3
src/loyal_companion/services/__init__.py
View File

@@ -2,6 +2,7 @@
from .ai_service import AIService
from .association_service import AssociationService
from .attachment_service import AttachmentContext, AttachmentService
from .communication_style_service import (
CommunicationStyleService,
detect_emoji_usage,
@@ -24,6 +25,8 @@ __all__ = [
"AIService",
"AIResponse",
"AssociationService",
"AttachmentContext",
"AttachmentService",
"CommunicationStyleService",
"ConversationManager",
"DatabaseService",

14
src/loyal_companion/services/ai_service.py
View File

@@ -20,6 +20,7 @@ from .providers import (
if TYPE_CHECKING:
from loyal_companion.models import BotOpinion, UserCommunicationStyle, UserRelationship
from .attachment_service import AttachmentContext
from .mood_service import MoodState
from .relationship_service import RelationshipLevel
@@ -148,6 +149,7 @@ You can use Discord markdown formatting in your responses."""
relationship: tuple[RelationshipLevel, UserRelationship] | None = None,
communication_style: UserCommunicationStyle | None = None,
bot_opinions: list[BotOpinion] | None = None,
attachment: AttachmentContext | None = None,
) -> str:
"""Build system prompt with all personality modifiers.
@@ -156,10 +158,12 @@ You can use Discord markdown formatting in your responses."""
relationship: Tuple of (level, relationship_record)
communication_style: User's learned communication preferences
bot_opinions: Bot's opinions relevant to the conversation
attachment: User's attachment context
Returns:
Enhanced system prompt with personality context
"""
from .attachment_service import AttachmentService
from .mood_service import MoodService
from .relationship_service import RelationshipService
@@ -173,12 +177,22 @@ You can use Discord markdown formatting in your responses."""
modifiers.append(f"[Current Mood]\n{mood_mod}")
# Add relationship modifier
relationship_level = None
if relationship and self._config.relationship_enabled:
level, rel = relationship
relationship_level = level.value
rel_mod = RelationshipService(None).get_relationship_prompt_modifier(level, rel)
if rel_mod:
modifiers.append(f"[Relationship]\n{rel_mod}")
# Add attachment context
if attachment and self._config.attachment_tracking_enabled:
attach_mod = AttachmentService(None).get_attachment_prompt_modifier(
attachment, relationship_level or "stranger"
)
if attach_mod:
modifiers.append(f"[Attachment Context]\n{attach_mod}")
# Add communication style
if communication_style and self._config.style_learning_enabled:
style_mod = self._get_style_prompt_modifier(communication_style)

422
src/loyal_companion/services/attachment_service.py Normal file
View File

@@ -0,0 +1,422 @@
"""Attachment Service - tracks and responds to attachment patterns.
Attachment styles:
- secure: comfortable with intimacy and independence
- anxious: fears abandonment, seeks reassurance, hyperactivates
- avoidant: uncomfortable with closeness, deactivates emotions
- disorganized: conflicting needs, push-pull patterns
This service detects patterns from messages and adapts Bartender's
responses to meet each person where they are.
"""
import logging
import re
from dataclasses import dataclass
from datetime import datetime, timezone
from enum import Enum
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from loyal_companion.config import settings
from loyal_companion.models import AttachmentEvent, User, UserAttachmentProfile
logger = logging.getLogger(__name__)
class AttachmentStyle(Enum):
"""Primary attachment styles."""
SECURE = "secure"
ANXIOUS = "anxious"
AVOIDANT = "avoidant"
DISORGANIZED = "disorganized"
UNKNOWN = "unknown"
class AttachmentState(Enum):
"""Current attachment system state."""
REGULATED = "regulated" # Baseline, calm
ACTIVATED = "activated" # Attachment system triggered
MIXED = "mixed" # Showing conflicting patterns
@dataclass
class AttachmentContext:
"""Current attachment context for a user."""
primary_style: AttachmentStyle
style_confidence: float
current_state: AttachmentState
state_intensity: float
recent_indicators: list[str]
effective_responses: list[str]
class AttachmentService:
"""Detects and responds to attachment patterns."""
# Indicators for each attachment style
ANXIOUS_INDICATORS = [
# Reassurance seeking
r"\b(do you (still )?(like|care|love)|are you (mad|angry|upset)|did i do something wrong)\b",
r"\b(please (don't|dont) (leave|go|abandon)|don't (leave|go) me)\b",
r"\b(i('m| am) (scared|afraid|worried) (you|that you))\b",
r"\b(are we (ok|okay|good|alright))\b",
r"\b(you('re| are) (going to|gonna) leave)\b",
# Checking behaviors
r"\b(are you (there|still there|here))\b",
r"\b(why (aren't|arent|didn't|didnt) you (respond|reply|answer))\b",
r"\b(i (keep|kept) checking|waiting for (you|your))\b",
# Fear of abandonment
r"\b(everyone (leaves|left|abandons))\b",
r"\b(i('m| am) (too much|not enough|unlovable))\b",
r"\b(what if you (leave|stop|don't))\b",
# Hyperactivation
r"\b(i (need|have) to (know|hear|see))\b",
r"\b(i can('t|not) (stop thinking|get .* out of my head))\b",
]
AVOIDANT_INDICATORS = [
# Emotional minimizing
r"\b(it('s| is) (fine|whatever|no big deal|not a big deal))\b",
r"\b(i('m| am) (fine|okay|good|alright))\b", # When context suggests otherwise
r"\b(doesn('t|t) (matter|bother me))\b",
r"\b(i don('t|t) (care|need|want) (anyone|anybody|help|support))\b",
# Deflection
r"\b(let('s|s) (talk about|change|not))\b",
r"\b(i('d| would) rather not)\b",
r"\b(anyway|moving on|whatever)\b",
# Independence emphasis
r"\b(i('m| am) (better|fine) (alone|by myself|on my own))\b",
r"\b(i don('t|t) need (anyone|anybody|people))\b",
# Withdrawal
r"\b(i (should|need to) go)\b",
r"\b(i('m| am) (busy|tired|done))\b", # When used to exit emotional topics
]
DISORGANIZED_INDICATORS = [
# Push-pull patterns
r"\b(i (want|need) you .* but .* (scared|afraid|can't))\b",
r"\b(come (closer|here) .* (go away|leave))\b",
r"\b(i (love|hate) (you|this))\b", # In same context
# Contradictory statements
r"\b(i('m| am) (fine|okay) .* (not fine|not okay|struggling))\b",
r"\b(i don('t|t) care .* (i do care|it hurts))\b",
# Confusion about needs
r"\b(i don('t|t) know what i (want|need|feel))\b",
r"\b(i('m| am) (confused|lost|torn))\b",
]
SECURE_INDICATORS = [
# Comfortable with emotions
r"\b(i('m| am) feeling|i feel)\b",
r"\b(i (need|want) (to talk|support|help))\b", # Direct ask
# Healthy boundaries
r"\b(i (need|want) some (space|time))\b", # Without avoidance
r"\b(let me (think|process))\b",
# Trust expressions
r"\b(i trust (you|that))\b",
r"\b(thank you for (listening|being here|understanding))\b",
]
# Minimum messages before determining primary style
MIN_SAMPLES_FOR_STYLE = 5
def __init__(self, session: AsyncSession) -> None:
self._session = session
async def get_or_create_profile(
self, user: User, guild_id: int | None = None
) -> UserAttachmentProfile:
"""Get or create attachment profile for a user."""
stmt = select(UserAttachmentProfile).where(
UserAttachmentProfile.user_id == user.id,
UserAttachmentProfile.guild_id == guild_id,
)
result = await self._session.execute(stmt)
profile = result.scalar_one_or_none()
if not profile:
profile = UserAttachmentProfile(user_id=user.id, guild_id=guild_id)
self._session.add(profile)
await self._session.flush()
return profile
async def analyze_message(
self, user: User, message_content: str, guild_id: int | None = None
) -> AttachmentContext:
"""Analyze a message for attachment indicators and update profile.
Returns the current attachment context for use in response generation.
"""
if not settings.attachment_tracking_enabled:
return self._default_context()
profile = await self.get_or_create_profile(user, guild_id)
# Detect indicators in message
anxious_matches = self._find_indicators(message_content, self.ANXIOUS_INDICATORS)
avoidant_matches = self._find_indicators(message_content, self.AVOIDANT_INDICATORS)
disorganized_matches = self._find_indicators(message_content, self.DISORGANIZED_INDICATORS)
secure_matches = self._find_indicators(message_content, self.SECURE_INDICATORS)
# Update indicator counts
profile.anxious_indicators += len(anxious_matches)
profile.avoidant_indicators += len(avoidant_matches)
profile.disorganized_indicators += len(disorganized_matches)
profile.secure_indicators += len(secure_matches)
# Determine current state
all_indicators = anxious_matches + avoidant_matches + disorganized_matches
current_state, intensity = self._determine_state(
anxious_matches, avoidant_matches, disorganized_matches
)
# Update profile state
if current_state != AttachmentState.REGULATED:
profile.current_state = current_state.value
profile.state_intensity = intensity
profile.last_activation_at = datetime.now(timezone.utc)
profile.activation_count += 1
else:
# Decay intensity over time
profile.state_intensity = max(0, profile.state_intensity - 0.1)
if profile.state_intensity < 0.2:
profile.current_state = AttachmentState.REGULATED.value
# Update primary style if enough data
total_indicators = (
profile.anxious_indicators
+ profile.avoidant_indicators
+ profile.disorganized_indicators
+ profile.secure_indicators
)
if total_indicators >= self.MIN_SAMPLES_FOR_STYLE:
primary_style, confidence = self._determine_primary_style(profile)
profile.primary_style = primary_style.value
profile.style_confidence = confidence
profile.updated_at = datetime.now(timezone.utc)
# Record event if activation detected
if current_state != AttachmentState.REGULATED and all_indicators:
await self._record_event(
user_id=user.id,
guild_id=guild_id,
event_type="activation",
detected_style=self._dominant_style(
anxious_matches, avoidant_matches, disorganized_matches
),
intensity=intensity,
trigger_message=message_content[:500],
trigger_indicators=all_indicators,
)
return AttachmentContext(
primary_style=AttachmentStyle(profile.primary_style),
style_confidence=profile.style_confidence,
current_state=AttachmentState(profile.current_state),
state_intensity=profile.state_intensity,
recent_indicators=all_indicators,
effective_responses=profile.effective_responses or [],
)
def get_attachment_prompt_modifier(
self, context: AttachmentContext, relationship_level: str
) -> str:
"""Generate prompt text based on attachment context.
Only reflects patterns at Close Friend level or above.
"""
if context.current_state == AttachmentState.REGULATED:
return ""
parts = []
# State-based modifications
if context.current_state == AttachmentState.ACTIVATED:
if context.state_intensity > 0.5:
parts.append("[Attachment Activated - High Intensity]")
else:
parts.append("[Attachment Activated]")
# Style-specific guidance
if context.primary_style == AttachmentStyle.ANXIOUS:
parts.append(
"This person's attachment system is activated - they may need reassurance. "
"Be consistent, present, and direct about being here. Don't leave things ambiguous. "
"Validate their feelings without reinforcing catastrophic thinking."
)
elif context.primary_style == AttachmentStyle.AVOIDANT:
parts.append(
"This person may be withdrawing or minimizing. Don't push or crowd them. "
"Give space while staying present. Normalize needing independence. "
"Let them set the pace - they'll come closer when it feels safe."
)
elif context.primary_style == AttachmentStyle.DISORGANIZED:
parts.append(
"This person may be showing conflicting needs - that's okay. "
"Be steady and predictable. Don't match their chaos. "
"Clear, consistent communication helps. It's okay if they push and pull."
)
# At Close Friend level, can reflect patterns
if relationship_level == "close_friend" and context.style_confidence > 0.5:
if context.recent_indicators:
parts.append(
"You know this person well enough to gently notice patterns if helpful. "
"Only reflect what you see if it serves them, not to analyze or diagnose."
)
# Add effective responses if we've learned any
if context.effective_responses:
parts.append(
f"Things that have helped this person before: {', '.join(context.effective_responses[:3])}"
)
return "\n".join(parts) if parts else ""
async def record_response_effectiveness(
self,
user: User,
guild_id: int | None,
response_style: str,
was_helpful: bool,
) -> None:
"""Record whether a response approach was helpful.
Called based on follow-up indicators (did they calm down, escalate, etc.)
"""
profile = await self.get_or_create_profile(user, guild_id)
if was_helpful:
if response_style not in (profile.effective_responses or []):
effective = profile.effective_responses or []
effective.append(response_style)
profile.effective_responses = effective[-10:] # Keep last 10
else:
if response_style not in (profile.ineffective_responses or []):
ineffective = profile.ineffective_responses or []
ineffective.append(response_style)
profile.ineffective_responses = ineffective[-10:]
def _find_indicators(self, text: str, patterns: list[str]) -> list[str]:
"""Find all matching indicators in text."""
text_lower = text.lower()
matches = []
for pattern in patterns:
if re.search(pattern, text_lower, re.IGNORECASE):
matches.append(pattern)
return matches
def _determine_state(
self,
anxious: list[str],
avoidant: list[str],
disorganized: list[str],
) -> tuple[AttachmentState, float]:
"""Determine current attachment state from indicators."""
total = len(anxious) + len(avoidant) + len(disorganized)
if total == 0:
return AttachmentState.REGULATED, 0.0
# Check for mixed/disorganized state
if (anxious and avoidant) or disorganized:
intensity = min(1.0, total * 0.3)
return AttachmentState.MIXED, intensity
# Single style activation
intensity = min(1.0, total * 0.25)
return AttachmentState.ACTIVATED, intensity
def _determine_primary_style(
self, profile: UserAttachmentProfile
) -> tuple[AttachmentStyle, float]:
"""Determine primary attachment style from accumulated indicators."""
counts = {
AttachmentStyle.ANXIOUS: profile.anxious_indicators,
AttachmentStyle.AVOIDANT: profile.avoidant_indicators,
AttachmentStyle.DISORGANIZED: profile.disorganized_indicators,
AttachmentStyle.SECURE: profile.secure_indicators,
}
total = sum(counts.values())
if total == 0:
return AttachmentStyle.UNKNOWN, 0.0
# Find dominant style
dominant = max(counts, key=counts.get)
confidence = counts[dominant] / total
# Check for disorganized pattern (high anxious AND avoidant)
if (
counts[AttachmentStyle.ANXIOUS] > total * 0.3
and counts[AttachmentStyle.AVOIDANT] > total * 0.3
):
return AttachmentStyle.DISORGANIZED, confidence
return dominant, confidence
def _dominant_style(self, anxious: list, avoidant: list, disorganized: list) -> str:
"""Get the dominant style from current indicators."""
if disorganized or (anxious and avoidant):
return "disorganized"
if len(anxious) > len(avoidant):
return "anxious"
if len(avoidant) > len(anxious):
return "avoidant"
return "mixed"
async def _record_event(
self,
user_id: int,
guild_id: int | None,
event_type: str,
detected_style: str,
intensity: float,
trigger_message: str,
trigger_indicators: list[str],
) -> None:
"""Record an attachment event for learning."""
event = AttachmentEvent(
user_id=user_id,
guild_id=guild_id,
event_type=event_type,
detected_style=detected_style,
intensity=intensity,
trigger_message=trigger_message,
trigger_indicators=trigger_indicators,
)
self._session.add(event)
def _default_context(self) -> AttachmentContext:
"""Return a default context when tracking is disabled."""
return AttachmentContext(
primary_style=AttachmentStyle.UNKNOWN,
style_confidence=0.0,
current_state=AttachmentState.REGULATED,
state_intensity=0.0,
recent_indicators=[],
effective_responses=[],
)
async def get_attachment_info(
session: AsyncSession, user: User, guild_id: int | None = None
) -> dict:
"""Get attachment information for display (e.g., in a command)."""
service = AttachmentService(session)
profile = await service.get_or_create_profile(user, guild_id)
return {
"primary_style": profile.primary_style,
"style_confidence": profile.style_confidence,
"current_state": profile.current_state,
"activation_count": profile.activation_count,
"effective_responses": profile.effective_responses,
}

366
tests/test_services.py
View File

@@ -11,10 +11,17 @@ from loyal_companion.models import (
Conversation,
Message,
User,
UserAttachmentProfile,
UserFact,
UserRelationship,
)
from loyal_companion.services.ai_service import AIService
from loyal_companion.services.attachment_service import (
AttachmentContext,
AttachmentService,
AttachmentState,
AttachmentStyle,
)
from loyal_companion.services.fact_extraction_service import FactExtractionService
from loyal_companion.services.mood_service import MoodLabel, MoodService, MoodState
from loyal_companion.services.opinion_service import OpinionService, extract_topics_from_message
@@ -618,3 +625,362 @@ class TestAIService:
service._provider = MagicMock()
assert service.model == "gpt-4o-mini"
class TestAttachmentService:
"""Tests for AttachmentService."""
@pytest.mark.asyncio
async def test_get_or_create_profile_new(self, db_session, sample_user):
"""Test creating a new attachment profile."""
service = AttachmentService(db_session)
profile = await service.get_or_create_profile(sample_user, guild_id=111222333)
assert profile.id is not None
assert profile.user_id == sample_user.id
assert profile.primary_style == "unknown"
assert profile.current_state == "regulated"
@pytest.mark.asyncio
async def test_get_or_create_profile_existing(self, db_session, sample_user):
"""Test getting an existing attachment profile."""
service = AttachmentService(db_session)
# Create first
profile1 = await service.get_or_create_profile(sample_user, guild_id=111222333)
await db_session.commit()
# Get again
profile2 = await service.get_or_create_profile(sample_user, guild_id=111222333)
assert profile1.id == profile2.id
@pytest.mark.asyncio
async def test_analyze_message_no_indicators(self, db_session, sample_user):
"""Test analyzing a message with no attachment indicators."""
service = AttachmentService(db_session)
context = await service.analyze_message(
user=sample_user,
message_content="Hello, how are you today?",
guild_id=111222333,
)
assert context.current_state == AttachmentState.REGULATED
assert len(context.recent_indicators) == 0
@pytest.mark.asyncio
async def test_analyze_message_anxious_indicators(self, db_session, sample_user):
"""Test analyzing a message with anxious attachment indicators."""
service = AttachmentService(db_session)
context = await service.analyze_message(
user=sample_user,
message_content="Are you still there? Do you still like me? Did I do something wrong?",
guild_id=111222333,
)
assert context.current_state == AttachmentState.ACTIVATED
assert len(context.recent_indicators) > 0
# Check profile was updated
profile = await service.get_or_create_profile(sample_user, guild_id=111222333)
assert profile.anxious_indicators > 0
@pytest.mark.asyncio
async def test_analyze_message_avoidant_indicators(self, db_session, sample_user):
"""Test analyzing a message with avoidant attachment indicators."""
service = AttachmentService(db_session)
context = await service.analyze_message(
user=sample_user,
message_content="It's fine, whatever. I don't need anyone. I'm better alone.",
guild_id=111222333,
)
assert context.current_state == AttachmentState.ACTIVATED
assert len(context.recent_indicators) > 0
profile = await service.get_or_create_profile(sample_user, guild_id=111222333)
assert profile.avoidant_indicators > 0
@pytest.mark.asyncio
async def test_analyze_message_disorganized_indicators(self, db_session, sample_user):
"""Test analyzing a message with disorganized attachment indicators."""
service = AttachmentService(db_session)
context = await service.analyze_message(
user=sample_user,
message_content="I don't know what I want. I'm so confused and torn.",
guild_id=111222333,
)
# Should detect disorganized patterns
assert len(context.recent_indicators) > 0
@pytest.mark.asyncio
async def test_analyze_message_mixed_state(self, db_session, sample_user):
"""Test that mixed indicators result in mixed state."""
service = AttachmentService(db_session)
# Message with both anxious and avoidant indicators
context = await service.analyze_message(
user=sample_user,
message_content="Are you still there? Actually, it's fine, I don't care anyway.",
guild_id=111222333,
)
assert context.current_state == AttachmentState.MIXED
@pytest.mark.asyncio
async def test_analyze_message_secure_indicators(self, db_session, sample_user):
"""Test analyzing a message with secure attachment indicators."""
service = AttachmentService(db_session)
context = await service.analyze_message(
user=sample_user,
message_content="I'm feeling sad today and I need to talk about it. Thank you for listening.",
guild_id=111222333,
)
profile = await service.get_or_create_profile(sample_user, guild_id=111222333)
assert profile.secure_indicators > 0
def test_find_indicators_anxious(self, db_session):
"""Test finding anxious indicators in text."""
service = AttachmentService(db_session)
matches = service._find_indicators(
"do you still like me?",
service.ANXIOUS_INDICATORS,
)
assert len(matches) > 0
def test_find_indicators_none(self, db_session):
"""Test finding no indicators in neutral text."""
service = AttachmentService(db_session)
matches = service._find_indicators(
"the weather is nice today",
service.ANXIOUS_INDICATORS,
)
assert len(matches) == 0
def test_determine_state_regulated(self, db_session):
"""Test state determination with no indicators."""
service = AttachmentService(db_session)
state, intensity = service._determine_state([], [], [])
assert state == AttachmentState.REGULATED
assert intensity == 0.0
def test_determine_state_activated(self, db_session):
"""Test state determination with single style indicators."""
service = AttachmentService(db_session)
state, intensity = service._determine_state(["pattern1", "pattern2"], [], [])
assert state == AttachmentState.ACTIVATED
assert intensity > 0
def test_determine_state_mixed(self, db_session):
"""Test state determination with mixed indicators."""
service = AttachmentService(db_session)
state, intensity = service._determine_state(["anxious1"], ["avoidant1"], [])
assert state == AttachmentState.MIXED
def test_get_attachment_prompt_modifier_regulated(self, db_session):
"""Test prompt modifier for regulated state."""
service = AttachmentService(db_session)
context = AttachmentContext(
primary_style=AttachmentStyle.UNKNOWN,
style_confidence=0.0,
current_state=AttachmentState.REGULATED,
state_intensity=0.0,
recent_indicators=[],
effective_responses=[],
)
modifier = service.get_attachment_prompt_modifier(context, "friend")
assert modifier == ""
def test_get_attachment_prompt_modifier_anxious_activated(self, db_session):
"""Test prompt modifier for anxious activated state."""
service = AttachmentService(db_session)
context = AttachmentContext(
primary_style=AttachmentStyle.ANXIOUS,
style_confidence=0.7,
current_state=AttachmentState.ACTIVATED,
state_intensity=0.6,
recent_indicators=["pattern1"],
effective_responses=[],
)
modifier = service.get_attachment_prompt_modifier(context, "friend")
assert "reassurance" in modifier.lower()
assert "present" in modifier.lower()
def test_get_attachment_prompt_modifier_avoidant_activated(self, db_session):
"""Test prompt modifier for avoidant activated state."""
service = AttachmentService(db_session)
context = AttachmentContext(
primary_style=AttachmentStyle.AVOIDANT,
style_confidence=0.7,
current_state=AttachmentState.ACTIVATED,
state_intensity=0.6,
recent_indicators=["pattern1"],
effective_responses=[],
)
modifier = service.get_attachment_prompt_modifier(context, "friend")
assert "space" in modifier.lower()
assert "push" in modifier.lower()
def test_get_attachment_prompt_modifier_disorganized_activated(self, db_session):
"""Test prompt modifier for disorganized activated state."""
service = AttachmentService(db_session)
context = AttachmentContext(
primary_style=AttachmentStyle.DISORGANIZED,
style_confidence=0.7,
current_state=AttachmentState.ACTIVATED,
state_intensity=0.6,
recent_indicators=["pattern1"],
effective_responses=[],
)
modifier = service.get_attachment_prompt_modifier(context, "friend")
assert "steady" in modifier.lower()
assert "predictable" in modifier.lower()
def test_get_attachment_prompt_modifier_close_friend_reflection(self, db_session):
"""Test prompt modifier includes reflection at close friend level."""
service = AttachmentService(db_session)
context = AttachmentContext(
primary_style=AttachmentStyle.ANXIOUS,
style_confidence=0.7,
current_state=AttachmentState.ACTIVATED,
state_intensity=0.6,
recent_indicators=["pattern1"],
effective_responses=[],
)
modifier = service.get_attachment_prompt_modifier(context, "close_friend")
assert "pattern" in modifier.lower()
def test_get_attachment_prompt_modifier_with_effective_responses(self, db_session):
"""Test prompt modifier includes effective responses."""
service = AttachmentService(db_session)
context = AttachmentContext(
primary_style=AttachmentStyle.ANXIOUS,
style_confidence=0.7,
current_state=AttachmentState.ACTIVATED,
state_intensity=0.6,
recent_indicators=["pattern1"],
effective_responses=["reassurance", "validation"],
)
modifier = service.get_attachment_prompt_modifier(context, "friend")
assert "helped" in modifier.lower()
assert "reassurance" in modifier.lower()
@pytest.mark.asyncio
async def test_record_response_effectiveness_helpful(self, db_session, sample_user):
"""Test recording a helpful response."""
service = AttachmentService(db_session)
await service.record_response_effectiveness(
user=sample_user,
guild_id=111222333,
response_style="reassurance",
was_helpful=True,
)
profile = await service.get_or_create_profile(sample_user, guild_id=111222333)
assert "reassurance" in (profile.effective_responses or [])
@pytest.mark.asyncio
async def test_record_response_effectiveness_unhelpful(self, db_session, sample_user):
"""Test recording an unhelpful response."""
service = AttachmentService(db_session)
await service.record_response_effectiveness(
user=sample_user,
guild_id=111222333,
response_style="advice",
was_helpful=False,
)
profile = await service.get_or_create_profile(sample_user, guild_id=111222333)
assert "advice" in (profile.ineffective_responses or [])
def test_default_context(self, db_session):
"""Test default context when tracking is disabled."""
service = AttachmentService(db_session)
context = service._default_context()
assert context.primary_style == AttachmentStyle.UNKNOWN
assert context.current_state == AttachmentState.REGULATED
assert context.style_confidence == 0.0
@pytest.mark.asyncio
async def test_primary_style_determination(self, db_session, sample_user):
"""Test that primary style is determined after enough samples."""
service = AttachmentService(db_session)
# Send multiple messages with anxious indicators
anxious_messages = [
"Are you still there?",
"Do you still like me?",
"Did I do something wrong?",
"Please don't leave me",
"Are you mad at me?",
"I'm scared you'll abandon me",
]
for msg in anxious_messages:
await service.analyze_message(
user=sample_user,
message_content=msg,
guild_id=111222333,
)
profile = await service.get_or_create_profile(sample_user, guild_id=111222333)
# After enough samples, primary style should be determined
assert profile.anxious_indicators >= 5
assert profile.style_confidence > 0
@pytest.mark.asyncio
async def test_activation_tracking(self, db_session, sample_user):
"""Test that activations are tracked."""
service = AttachmentService(db_session)
await service.analyze_message(
user=sample_user,
message_content="Are you still there? Do you still like me?",
guild_id=111222333,
)
profile = await service.get_or_create_profile(sample_user, guild_id=111222333)
assert profile.activation_count >= 1
assert profile.last_activation_at is not None