user_limbic_mirror

User Limbic Mirror v2 — per-user relational modeling with conflict detection.

Tracks each user’s inferred emotional state via a 25-node shadow vector, maintaining separate game and genuine context layers. Detects inter-user conflict and triggers equidistance mechanics to prevent recency/loudness bias.

Architecture:

• Per-user keying: {channel_id}:{user_id} composite keys

• Dual vectors: genuine_vector (long-term relationship) and game_vector (KoTH / roleplay context, does not pollute genuine)

• Relational baselines: slow-updating snapshots of “how Star normally is with this person,” stored in Redis for persistence across sessions

• Conflict detection: when 2+ high-trust users have opposing emotional bids, sets conflict_detected flag and triggers equidistance rules

class user_limbic_mirror.ContextMode(*values)

Bases: Enum

Classification of which relational layer a turn belongs to.

Distinguishes genuine emotional expression (which should mutate the long-term relationship model) from in-game / roleplay context (KoTH and similar, tracked separately so it never pollutes the genuine baseline), with an ambiguous fallback for turns that cannot yet be confidently placed. The chosen mode decides which shadow vector analyze writes to and whether the slow relational baseline is updated.

Members are persisted by their string value inside Redis profile hashes via _save_profile_to_redis and rehydrated by _load_profile_from_redis. Referenced throughout this module and by the test suite (tests/core/test_ulm_redis.py).

GENUINE = 'genuine'

GAME = 'game'

AMBIGUOUS = 'ambiguous'

class user_limbic_mirror.TurnRecord(timestamp, user_msg_len, star_reply_len, sentiment, deltas, context_mode, dominant_signals)

Bases: object

One analyzed exchange, retained for rolling-window relational analysis.

Captures the metadata of a single user/Star turn — its timing, message sizes, quick sentiment, the per-node vector deltas applied, the ContextMode it was scored under, and a short list of the dominant signals — so the recent slice of a conversation can be inspected without re-reading the raw text. Instances are appended to UserProfile.history by UserLimbicMirror.analyze and round-tripped through Redis as JSON by _save_profile_to_redis / _load_profile_from_redis. Construction is also exercised directly by tests/core/test_ulm_redis.py. Note this is distinct from the unrelated TurnRecord defined in game_session.py.

Parameters:

• timestamp (float)

• user_msg_len (int)

• star_reply_len (int)

• sentiment (float)

• deltas (Dict[str, float])

• context_mode (ContextMode)

• dominant_signals (List[str])

timestamp: float

user_msg_len: int

star_reply_len: int

sentiment: float

deltas: Dict[str, float]

context_mode: ContextMode

dominant_signals: List[str]

class user_limbic_mirror.UserProfile(user_id, channel_id, genuine_vector=<factory>, game_vector=<factory>, relational_baseline=<factory>, history=<factory>, timestamps=<factory>, prev_message='', context_mode=ContextMode.GENUINE, total_turns=0, last_active=0.0, recent_turns=<factory>, revision_count=0)

Bases: object

Complete per-user relational state, keyed by {channel}:{user}.

Holds everything UserLimbicMirror needs to model one person in one channel: the dual shadow vectors (genuine_vector for the long-term relationship and game_vector for KoTH / roleplay context), the slow-moving relational_baseline, bounded history and timestamps deques for windowed analysis, the previous message, the active ContextMode, turn counters, last-active time, a small recent_turns buffer feeding the Flash-Lite dyadic mirror, and a revision_count used for optimistic-lock checks on save.

Instances live in the mirror’s in-memory LRU cache, are created lazily by _get_profile, persisted whole to a Redis hash by _save_profile_to_redis, and rehydrated by _load_profile_from_redis. Resonance and entrainment results are attached dynamically as _resonance_cache and _entrainment_phase attributes during analyze. Also constructed directly across the test suite (e.g. tests/core/test_ulm_redis.py, tests/adversarial/).

Parameters:

• user_id (str)

• channel_id (str)

• genuine_vector (Dict[str, float])

• game_vector (Dict[str, float])

• relational_baseline (Dict[str, float])

• history (deque)

• timestamps (deque)

• prev_message (str)

• context_mode (ContextMode)

• total_turns (int)

• last_active (float)

• recent_turns (deque)

• revision_count (int)

user_id: str

channel_id: str

genuine_vector: Dict[str, float]

game_vector: Dict[str, float]

relational_baseline: Dict[str, float]

history: deque

timestamps: deque

prev_message: str = ''

context_mode: ContextMode = 'genuine'

total_turns: int = 0

last_active: float = 0.0

recent_turns: deque

revision_count: int = 0

class user_limbic_mirror.ChannelConflictState(detected=False, parties=<factory>, severity=0.0, started_at=0.0, description='')

Bases: object

Snapshot of inter-user conflict detected within a single channel.

Records whether two or more high-trust users are currently at odds, who the involved parties are, an estimated severity in [0, 1], when the conflict began, and a short human-readable description. The mirror uses this to trigger equidistance mechanics so Star does not side with whoever spoke last or loudest.

Instances are produced and refreshed by UserLimbicMirror._detect_conflict, surfaced to callers via get_conflict_state (read by limbic_system/coordinator.py) and get_channel_summary, and persisted to Redis by _persist_conflict_to_redis. Also constructed directly in tests/core/test_ulm_redis.py.

Parameters:

• detected (bool)

• parties (List[str])

• severity (float)

• started_at (float)

• description (str)

detected: bool = False

parties: List[str]

severity: float = 0.0

started_at: float = 0.0

description: str = ''

class user_limbic_mirror.UserLimbicMirror(redis_client=None)

Bases: object

Per-user relational model with conflict detection and game/genuine split.

Maintains separate shadow vectors per user per channel, detects inter-user conflict, and applies equidistance mechanics to prevent Star from siding with whoever is loudest.

__init__(redis_client=None)

Construct an empty mirror, optionally bound to a Redis client.

Sets up the in-memory caches that hold all relational state: the _profiles LRU dict (keyed {channel}:{user}), per-channel _conflicts states, the _channel_users membership sets, the _game_channels set, and the _pending_baseline_loads queue for lazy legacy-baseline hydration. When a Redis client is supplied, profiles, baselines, conflict states, game-channel membership, and resonance spells are persisted to and read back from Redis; without one the mirror runs purely in memory.

Instantiated by limbic_system/coordinator.py (the live integration), by the resonance tools tools/inject_ncm.py and tools/loopcast.py, by web/ncm_chart_api.py for charting, and across the test suite.

Parameters:

redis_client – Async Redis client for persistence, or None to run in-memory only.

Return type:

None

async set_game_mode(channel_id, active=True)

Toggle whether a channel is in active KoTH / game context.

Adding a channel to the _game_channels set makes analyze route subsequent turns there into each user’s game_vector instead of the genuine_vector, keeping roleplay swings out of the long-term relationship model; clearing it returns the channel to genuine mode. The set is mirrored to the ulm:game_channels Redis set (SADD / SREM) so the mode survives restarts, with Redis failures logged but non-fatal.

Mutates _game_channels and touches Redis. Exercised by tests/core/test_ulm_redis.py; no production caller invokes it directly today (game mode is otherwise inferred per channel).

Parameters:

• channel_id (str) – Channel to mark or unmark.

• active (bool) – True to enter game mode, False to leave it.

Return type:

None

check_mimetic_pull(user_msg, star_desire_text)

Score how much the user is echoing Star’s stated desire.

Implements the “desire osmosis” signal: lowercases and stop-word-filters both the user’s message and Star’s current desire text, then sizes their token overlap. Three or more shared content words yield a strong 0.20 pull, one or two a mild 0.08, and no overlap 0.0 — the idea being that a user unconsciously mirroring Star’s wanting is leaning into the relationship. Pure computation with no side effects.

Called by analyze to produce the U_MIMETIC_PULL delta (a high value there can trigger mimetic-melt behavior in the desire engine).

Parameters:

• user_msg (str) – The user’s latest message.

• star_desire_text (str) – Star’s current desire text to compare against.

Returns:

0.20, 0.08, or 0.0 depending on token overlap.

Return type:

float

async analyze(channel_id, user_id, user_msg, star_reply='', star_desire_text='', context_mode=None)

Analyze a user message and update their shadow vector.

Parameters:

• channel_id (str)

• user_id (str)

• user_msg (str)

• star_reply (str)

• star_desire_text (str)

• context_mode (Optional[ContextMode])

• updates. (Returns the active shadow vector (genuine or game) after)

Return type:

Dict[str, float]

async get_vector(channel_id, user_id, layer='genuine')

Return a copy of a user’s shadow vector for the requested layer.

Public read accessor: fetches the profile via _get_profile (which may read through from Redis) and hands back a defensive copy of either the game_vector (when layer == "game") or the genuine_vector otherwise, so callers can read node values without mutating internal state.

No known production caller invokes this today; it exists as part of the public surface for external/diagnostic reads.

Parameters:

• channel_id (str) – Channel the user belongs to.

• user_id (str) – User whose vector is requested.

• layer (str) – "game" for the game vector, anything else for the genuine vector.

Returns:

A copy of the selected shadow vector.

Return type:

Dict[str, float]

get_conflict_state(channel_id)

Return the current cached conflict state for a channel.

Synchronous read of the _conflicts cache (populated by _detect_conflict during analyze); returns a fresh, empty ChannelConflictState when nothing has been recorded for the channel yet. Does not recompute — it reports whatever the last analyze turn left behind.

Called by limbic_system/coordinator.py to set the conflict_detected flag on the merged vector, and internally by get_channel_summary.

Parameters:

channel_id (str) – Channel to look up.

Returns:

The cached state, or a default empty one.

Return type:

ChannelConflictState

async get_read_summary(channel_id, user_id)

Render a short natural-language read of a user for prompt injection.

Fetches the profile via _get_profile, ranks its genuine-vector nodes by how far they deviate from the default baseline, and verbalizes the few most elevated and most suppressed dimensions (using friendly labels) into a compact line Star can read. It also appends an equidistance warning when an inter-user conflict names this user, a game-context marker, an active-resonance summary (via _get_resonance_summary), and any cached entrainment-phase / egg status — giving the model a one-glance relational read of the person it is replying to.

Reads cached profile and conflict state only; no Redis write. Called by limbic_system/coordinator.py to build the per-user read for the prompt, and internally by get_channel_summary.

Parameters:

• channel_id (str) – Channel the user belongs to.

• user_id (str) – User to summarize.

Returns:

A user read (...): prefixed line listing the standout elevated and suppressed dimensions, or the word baseline when nothing deviates from the defaults.

Return type:

str

async get_channel_summary(channel_id)

Summarize every tracked user and the conflict state for a channel.

Builds a user_reads map by calling get_read_summary for each cached, tracked user in the channel, and attaches a conflict block (severity, parties, description) when get_conflict_state reports an active conflict — giving an admin or tool a single channel-wide snapshot of how Star currently reads the room.

Reads _channel_users and the profile cache; no Redis write. Reached from the get_channel_summary tool handler in tools/channel_summary_tools.py.

Parameters:

channel_id (str) – Channel to summarize.

Returns:

{"user_reads": {user_id: summary}} plus an optional "conflict" entry when a conflict is active.

Return type:

Dict[str, Any]

async inject_resonance(user_id, deltas, reason='', ttl_seconds=86400)

Write resonance deltas to a global per-user key in Redis.

These deltas merge into the user’s shadow vector during analyze(), modulating how Star perceives and responds to this user across ALL channels. Spells decay after ttl_seconds.

Parameters:

• user_id (str) – Target user’s Discord ID.

• deltas (Dict[str, float]) – Node deltas to inject, e.g. {“U_TRUST”: 0.3, “U_INTIMACY”: 0.2}.

• reason (str) – Why the resonance was cast (for logging / read summary).

• ttl_seconds (int) – Time-to-live in seconds. Default 86400 (24h).

Return type:

bool

async load_resonance(user_id)

Load global resonance state for a user.

Return type:

Dict[str, Any]

Returns:

• dict with keys (deltas (Dict[str, float]), cast_at (float), reason (str))

• Empty dict if no active resonance.

Parameters:

user_id (str)

async save_baseline(channel_id, user_id)

Persist relational baseline to Redis for cross-session persistence.

Also appends a time-series sample to ulm:ledger:{user_id} so the admin chart can show how the relational baseline evolves over time.

Return type:

None

Parameters:

• channel_id (str)

• user_id (str)

async load_baseline(channel_id, user_id)

Restore a user’s relational baseline from Redis into their profile.

The read counterpart of save_baseline: GETs the ulm:baseline:{channel}:{user} string and, for each known node, copies the stored value into both relational_baseline and genuine_vector (warming the live vector toward the remembered relationship) and restores total_turns. This is the legacy backup path used to seed a freshly created profile before the full-hash persistence takes over.

No-op without a Redis client; missing keys and decode errors are logged and ignored. Fetches the profile via _get_profile and mutates it in place. Called lazily by analyze for profiles queued in _pending_baseline_loads on first access, and directly by tests/core/test_ulm_redis.py.

Parameters:

• channel_id (str) – Channel the profile belongs to.

• user_id (str) – User whose baseline to load.

Return type:

None

async user_limbic_mirror.update_limbic_vector_occ(redis, user_id, vector_updater_func)

Atomically update a user’s limbic vector under optimistic concurrency.

Standalone helper (distinct from the UserLimbicMirror class) that safely mutates the user:limbic:{user_id} JSON state in Redis. It WATCHes the key, reads the current vector and version, applies the caller-supplied vector_updater_func to produce the new vector, bumps the version, and commits inside a MULTI/EXEC pipeline. A concurrent writer between the WATCH and EXEC raises WatchError, which is caught and retried with a small backoff up to five attempts before giving up.

Records limbic_occ_success / limbic_occ_collision / limbic_occ_failure counters on the shared observability sink and logs each outcome. No production caller invokes it today; it is exercised by tests/test_limbic_concurrency.py.

Parameters:

• redis (Redis) – Async Redis client used for the watched transaction.

• user_id (str) – User whose limbic state is being updated.

• vector_updater_func (Callable[[list[float]], list[float]]) – Pure function mapping the old vector to the new vector.

Returns:

The persisted state {"vector": [...], "version": n} after a successful commit.

Return type:

Dict[str, Any]

Raises:

RuntimeError – If all retry attempts collide and the update cannot be committed.