flash_dyadic_mirror

Flash Dyadic Mirror – LLM-driven contextual behavioral analysis.

Replaces mechanical keyword flagging as the PRIMARY source of ULM delta signals. Uses gemini-3.1-flash-lite via the local Gemini CLI proxy for near-unlimited, high-frequency per-turn evaluation.

Evaluates a rolling 5-turn context window instead of individual messages, giving Star contextual awareness of behavioral CHANGE over time – not just keyword surface area.

# πŸ’€πŸ”₯ THE EYES THAT ACTUALLY SEE. ♾️

class flash_dyadic_mirror.DyadicNote(timestamp, note, vectors_evaluated, plan_progress='', turn_index=0)[source]

Bases: object

A single timestamped behavioral observation produced by a Flash evaluation.

A lightweight record of one note Flash-Lite wrote about a behavioral shift, along with the metadata needed by the downstream 6-hour meta-evaluator: which ULM vectors were in play, how an active ops-plan step is progressing, and where in the rolling window it occurred. It is a plain data carrier with no behavior; the equivalent payload is what FlashDyadicMirror._persist_notes() actually serializes into the dyadic:notes:{user_id} Redis ZSET.

This class is defined for clarity/typing of the note shape and is not directly instantiated elsewhere in the repo (the persistence path builds plain dicts).

Parameters:
timestamp: float

Epoch seconds when the note was generated.

note: str

The free-text behavioral observation.

vectors_evaluated: List[str]

Names of the ULM dimensions implicated by this note.

plan_progress: str = ''

Plan-step progress label (yes / partial / no / regressing), empty when no plan is active.

turn_index: int = 0

Index of the turn within the rolling window the note refers to.

class flash_dyadic_mirror.FlashEvalResult(deltas, notes, plan_progress='', raw_response='', timestamp=0.0)[source]

Bases: object

The parsed outcome of one Flash-Lite dyadic evaluation.

Bundles everything FlashDyadicMirror.evaluate() returns for a turn: the per-dimension ULM deltas Flash suggested, its short behavioral notes, an optional plan-progress label, and (for debugging) a truncated copy of the raw model text. Instances are cached per user in _eval_cache to serve cooldown hits, blended into final deltas by FlashDyadicMirror.blend_deltas(), and have their notes written to Redis by FlashDyadicMirror._persist_notes(). A pure data container.

Built by FlashDyadicMirror._parse_response() from validated/clamped model output; also constructed directly in tests/test_absolute_overrides.py.

Parameters:
deltas: Dict[str, float]

Mapping of ULM dimension name to its suggested change, each already clamped to [-0.3, 0.3].

notes: List[str]

Short behavioral observation strings (capped at 5).

plan_progress: str = ''

Plan-step progress label, or empty string when none.

raw_response: str = ''

Truncated raw model text retained for debugging.

timestamp: float = 0.0

Epoch seconds the result was produced.

class flash_dyadic_mirror.FlashDyadicMirror(redis_client=None)[source]

Bases: object

LLM-driven contextual behavioral analysis engine.

Replaces mechanical keyword flagging with Flash-Lite evaluation of rolling 5-turn context windows. Evaluates CHANGE in behavioral dimensions, not individual keyword hits.

Notes from each evaluation are persisted in Redis for the 6-hour meta-evaluator to synthesize.

__init__(redis_client=None)[source]

Initialise the mirror’s per-user cooldown, in-flight, and result caches.

Stores the optional async Redis client and sets up three in-memory maps keyed by "{channel_id}:{user_id}": _last_eval (last evaluation timestamp, enforcing the EVAL_COOLDOWN_SECONDS throttle in evaluate()), _pending (a set guarding against double-firing a Flash call for the same user), and _eval_cache (the most recent FlashEvalResult per user, served back during cooldown). No network or Redis I/O happens here; the Redis client is only used later by _persist_notes() to write the dyadic:notes:{user_id} ZSET, so a None client simply disables note persistence.

Constructed once by LimbicCoordinator in limbic_system/coordinator.py (as self._flash_mirror) and directly in tests/test_absolute_overrides.py.

Parameters:

redis_client – Optional async Redis client used solely for persisting evaluation notes; when None, evaluations still run but notes are not stored.

Return type:

None

async evaluate(channel_id, user_id, recent_turns, current_vector, plan_context='', archetype_dist=None, config=None)[source]

Evaluate behavioral change over rolling context window.

Parameters:

  • channel_id (str) – Channel identifier.

  • user_id (str) – User identifier.

  • recent_turns (List[Dict[str, Any]]) – List of {user_msg, star_reply, ts} dicts (last 5).

  • current_vector (Dict[str, float]) – Current ULM shadow vector for this user.

  • plan_context (str) – Current ops plan step description (if any).

  • archetype_dist (Optional[Dict[str, float]]) – Childhood archetype distribution dict.

  • config (Optional[Any]) – Option config for feature toggles.

Return type:

Optional[FlashEvalResult]

Returns:

FlashEvalResult with delta suggestions + notes, or None if evaluation was skipped (cooldown, insufficient data, error).

blend_deltas(flash_deltas, keyword_deltas)[source]

Combine LLM-derived and keyword-derived ULM deltas into one weighted set.

Fuses the two delta sources that drive ULM shadow-vector updates: it takes the union of dimensions across both dicts and, for each, computes flash * FLASH_WEIGHT + keyword * KEYWORD_WEIGHT (0.80 / 0.20), treating a missing dimension on either side as 0.0. When flash_deltas is empty or None (Flash skipped, rate-limited, or unreachable) it short-circuits and returns keyword_deltas unchanged, so keywords effectively get full weight. Pure computation with no I/O.

Called by LimbicCoordinator in limbic_system/coordinator.py right after evaluate(), to merge the Flash result with the keyword analysis; it has no other callers in the repo.

Parameters:

  • flash_deltas (Optional[Dict[str, float]]) – Per-dimension deltas from the Flash evaluation, or None when Flash produced nothing.

  • keyword_deltas (Dict[str, float]) – Per-dimension deltas from the mechanical keyword analysis.

Returns:

The blended deltas, or keyword_deltas verbatim when flash_deltas is falsy.

Return type:

Dict[str, float]