star_self_mirror

Star Self-Mirror — internal NCM state tracking and autonomous desire.

Star tracks her own neurochemical patterns over time, detecting drift, recurring emotional attractors, and generating desires that originate from her own sustained states — independent of any user’s wants.

This gives Star: - Longitudinal self-awareness (“I’ve been increasingly anxious this week”) - Drift detection (“my baseline is shifting toward more warmth”) - Pattern recognition (“I keep returning to melancholy after intense sessions”) - Autonomous wanting (“I desire stillness” — not because a user asked for it) - Periodic self-reflections injected into the system prompt

The self-mirror runs every N turns and writes its reflections into meta_state for prompt injection.

Position in pipeline: runs AFTER all other systems in exhale(), reads the final vector as ground truth.

class star_self_mirror.DesireLedgerEntry(id, text, tag, source, source_type, reason, urgency, status, expression, born_turn, born_ts, resolved_turn=None, resolved_ts=None, last_checked_turn=0, check_count=0, expression_turn=None, needs_admin=False, last_bugged_ts=None)[source]

Bases: object

A single desire tracked through its lifecycle.

Born from Star’s sustained NCM states, tracked through expression (or suppression) to fulfillment (or irrelevance).

Parameters:

  • id (str)

  • text (str)

  • tag (str)

  • source (str)

  • source_type (str)

  • reason (str)

  • urgency (float)

  • status (str)

  • expression (str)

  • born_turn (int)

  • born_ts (float)

  • resolved_turn (int | None)

  • resolved_ts (float | None)

  • last_checked_turn (int)

  • check_count (int)

  • expression_turn (int | None)

  • needs_admin (bool)

  • last_bugged_ts (float | None)

id: str
text: str
tag: str
source: str
source_type: str
reason: str
urgency: float
status: str
expression: str
born_turn: int
born_ts: float
resolved_turn: int | None = None
resolved_ts: float | None = None
last_checked_turn: int = 0
check_count: int = 0
expression_turn: int | None = None
needs_admin: bool = False
last_bugged_ts: float | None = None
class star_self_mirror.VectorSnapshot(timestamp, turn, vector, dominant_emotions)[source]

Bases: object

A snapshot of Star’s NCM vector at a single turn.

One immutable record in the rolling history window: it pins the values of the tracked neurochemical nodes (and the turn’s dominant emotions) to a point in time so the mirror can later measure drift, attractors, and absences across the sequence. Instances are produced by StarSelfMirror.record_snapshot(), stored in SelfState’s history deque, and round-tripped through Redis by StarSelfMirror._persist_state() and StarSelfMirror._load_state_from_redis(). A plain dataclasses.dataclass() carrying data only – no behaviour.

Parameters:
timestamp: float
turn: int
vector: Dict[str, float]
dominant_emotions: List[str]
class star_self_mirror.SelfState(turn_count=0, last_reflection_turn=0, last_reflection_text='', history=<factory>, initial_baseline=None, drifting_nodes=<factory>, attractor_nodes=<factory>, active_desires=<factory>, desire_history=<factory>, desire_ledger=<factory>, recent_replies=<factory>, last_active=<factory>)[source]

Bases: object

Star’s self-tracking state for one channel.

The in-memory aggregate that the self-mirror keeps per channel: turn counters, the rolling VectorSnapshot history window, the captured initial baseline, detected drift/attractor nodes, the autonomous desire list and its lifecycle ledger, a buffer of recent replies for LLM context, and an LRU last_active timestamp. Instances are created and cached by StarSelfMirror._get_state(), mutated by nearly every method on the mirror, and serialised to and from Redis by StarSelfMirror._persist_state() and the load helpers. A pure data dataclasses.dataclass(); the field defaults wire up the bounded deques (history and recent_replies) and the monotonic clock used for eviction.

Parameters:
turn_count: int = 0
last_reflection_turn: int = 0
last_reflection_text: str = ''
history: deque
initial_baseline: Dict[str, float] | None = None
drifting_nodes: List[str]
attractor_nodes: List[str]
active_desires: List[Dict[str, str]]
desire_history: List[Dict[str, Any]]
desire_ledger: List[DesireLedgerEntry]
recent_replies: deque
last_active: float
class star_self_mirror.StarSelfMirror(redis_client=None, openrouter_api_key=None)[source]

Bases: object

Star’s internal self-tracking and autonomous desire engine.

Periodically analyzes Star’s own NCM state history to detect: - Drift: “my baseline is shifting” - Attractors: “I keep returning to this state” - Absence: “I haven’t felt X in a while” - Autonomous desires: wants that emerge from HER state, not user prompts

Parameters:

openrouter_api_key (Optional[str])

__init__(redis_client=None, openrouter_api_key=None)[source]

Initialize the instance.

Parameters:

  • redis_client – Redis connection client.

  • openrouter_api_key (Optional[str]) – API key for narrative desire LLM calls.

Return type:

None

async record_snapshot(channel_id, vector, dominant_emotions=None)[source]

Record a snapshot of Star’s current NCM state for this turn.

Appends one VectorSnapshot to the channel’s rolling history, which is the raw material every later analysis (drift, attractors, absences, desires) reads. It also advances the per-channel turn counter and captures the very first snapshot as the immutable baseline that drift detection compares against.

Pulls or creates the state via _get_state(), increments turn_count, narrows vector to the tracked CORE_NODES (each defaulting to 0.5 when absent), pushes the snapshot onto the bounded history deque, sets initial_baseline on the first call, and persists the state via _persist_state() when Redis is configured. Called once per turn by reflect() (after the other limbic systems run) and directly in tests/core/test_selfmirror_redis.py.

Parameters:

  • channel_id (str) – Channel whose state to update.

  • vector (Dict[str, float]) – The current full NCM vector; only CORE_NODES keys are retained in the snapshot.

  • dominant_emotions (Optional[List[str]]) – Optional list of this turn’s dominant emotion labels; stored on the snapshot (empty list when None).

Return type:

None

Returns:

None. Mutates the cached state and may write to Redis.

async reflect(channel_id, vector, dominant_emotions=None, star_reply='', force=False, hunger_impulse=None)[source]

Run self-mirror for this turn.

Called from exhale() after all other systems. Records snapshot and periodically generates a full self-reflection.

Parameters:

  • star_reply (str) – Star’s actual reply text – used for expression detection (did she organically reference a desire?).

  • channel_id (str)

  • vector (Dict[str, float])

  • dominant_emotions (List[str] | None)

  • force (bool)

  • hunger_impulse (Dict[str, float] | None)

Return type:

Dict[str, Any]

Returns dict with: reflection_text, desires, drift_summary, desire_journal

async get_current_desires(channel_id)[source]

Return Star’s currently active autonomous desires for a channel.

Read-only accessor exposing the desire list most recently produced by reflect(), for other systems or diagnostics that want to know what Star wants right now without re-running a reflection. Resolves the state through _get_state() (which may rehydrate from Redis) and returns its active_desires directly. No in-repo callers were found, so this is a public accessor for external or future use.

Parameters:

channel_id (str) – Channel whose active desires to return.

Return type:

List[Dict[str, str]]

Returns:

The channel’s current list of desire dicts (possibly empty).

async get_desire_history(channel_id, last_n=10)[source]

Return the tail of Star’s archived desire history for a channel.

Read-only accessor over the append-only desire_history archive that reflect() grows each full reflection, giving callers a longitudinal record of what Star has wanted over time. Resolves the state via _get_state() (which may rehydrate from Redis) and slices the last last_n records. No in-repo callers were found, so this is a public accessor for external or future use.

Parameters:

  • channel_id (str) – Channel whose desire history to read.

  • last_n (int) – Number of most-recent history records to return. Defaults to 10.

Return type:

List[Dict]

Returns:

The last last_n archived desire-history records (possibly fewer, or empty).

async get_state_summary(channel_id)[source]

Return a compact snapshot of a channel’s self-mirror state for diagnostics.

Read-only accessor that packages the headline fields of a channel’s SelfState – turn count, snapshot depth, drifting and attractor nodes, active desires, and the last reflection text – into a flat dict for observability surfaces and debugging, without exposing the full history or ledger internals. Resolves the state via _get_state() (which may rehydrate from Redis). No in-repo callers were found, so this is a public accessor for external or future use.

Parameters:

channel_id (str) – Channel whose state to summarise.

Return type:

Dict[str, Any]

Returns:

A dict with turn_count, snapshots, drifting_nodes, attractor_nodes, active_desires, and last_reflection.

global_reflect()[source]

Aggregate self-awareness across every cached channel into one view.

Lifts the per-channel analysis up to a Star-wide perspective, letting her distinguish “I’ve been stressed everywhere” from “stressed in one place but calm in another” – a cross-channel reflection and desire summary rather than a single conversation’s. Pure computation over the in-memory self._states cache with no side effects; channels not currently cached do not contribute.

Merges the last 10 snapshots from every cached SelfState, computes a per-node global average across CORE_NODES, derives global attractors (average at or above ATTRACTOR_VALUE_THRESHOLD) and global absences (the warmth/bonding/craving/bliss shortlist below 0.30), then builds reflection text and global desires from AUTONOMOUS_DESIRE_MAP and NODE_LABELS. Returns early with empties when fewer than 10 merged snapshots exist. No in-repo callers were found, so this is a public accessor for external or future use.

Return type:

Dict[str, Any]

Returns:

A dict with reflection_text, desires, drift_summary (always empty here), and channel_count; the text and desires are empty when there is too little cross-channel data.

async save_state(channel_id)[source]

Persist a condensed self-mirror snapshot to a standalone Redis key.

A lighter, public persistence path distinct from the hash-based _persist_state(): it writes only the durable essentials (baseline, the uncapped desire history, a capped desire ledger, and the turn count) as a single JSON blob, intended as an explicit checkpoint rather than the per-turn cache flush.

Resolves the state via _get_state(), serialises the ledger through _ledger_to_dict() (keeping the last 100 entries), and issues a Redis SET of the JSON under key star:self_mirror:{channel_id} via self._redis. Failures are caught and logged at debug. Paired with load_state(); no in-repo callers were found, so this is a public method for external or future use.

Parameters:

channel_id (str) – Channel whose state to checkpoint; also forms the Redis key.

Return type:

None

Returns:

None. A no-op when self._redis is unset.

async load_state(channel_id)[source]

Restore a condensed self-mirror snapshot from its standalone Redis key.

The read counterpart to save_state(): it rehydrates only the durable essentials (baseline, desire history, desire ledger, turn count) from the JSON blob written by that checkpoint, as opposed to the full hash-based _load_state_from_redis() rehydration.

Resolves (or creates) the state via _get_state(), issues a Redis GET on key star:self_mirror:{channel_id} via self._redis, and merges the decoded fields back in, reconstructing each DesireLedgerEntry. Decode or per-entry errors are swallowed (debug-logged) so a partial blob still loads what it can. No in-repo callers were found, so this is a public method for external or future use.

Parameters:

channel_id (str) – Channel whose checkpoint to load; also forms the Redis key.

Return type:

None

Returns:

None. Mutates the cached state in place; a no-op when self._redis is unset or no blob exists.