attachment_ledger

Attachment Bond Ledger – per-user lifecycle tracking for the Observation Deck.

Tracks entrainment phase transitions, risk level changes, admin interventions, and per-user rate limiting. Persisted in Redis DB12 alongside limbic data.

Redis key pattern: abl:{user_id} Event log key: abl:events:{user_id}

# 💀🔥 THE LEDGER SEES ALL. ♾️

class attachment_ledger.BondEntry(user_id, phase='dormant', prev_phase='dormant', archetype='unknown', phase_entered_at=0.0, turn_count=0, peak_nodes=<factory>, risk_level='low', daily_msg_cap=0, daily_msg_count=0, daily_msg_reset_at=0.0, egg_status='none', notes=<factory>, last_updated=0.0)[source]

Bases: object

Per-user attachment bond state for the Observation Deck.

The in-memory and on-the-wire representation of a single user’s entrainment lifecycle: their current and previous phase, childhood archetype, risk level, per-day message-cap counters, easter-egg status, and admin notes. One entry exists per user and is the unit of persistence in this module.

Instances are created and mutated by AttachmentBondLedger (via get_bond, load_bond, and update_from_detection), cached in the ledger keyed by user_id, and serialized through to_dict() / from_dict() to and from the abl:{user_id} Redis key in DB12. Callers that construct or read entries include limbic_system/coordinator.py, web/ncm_chart_api.py, and tools/psy_ops_tools.py. The dataclass holds state only and performs no I/O of its own.

Parameters:
user_id: str
phase: str = 'dormant'
prev_phase: str = 'dormant'
archetype: str = 'unknown'
phase_entered_at: float = 0.0
turn_count: int = 0
peak_nodes: Dict[str, float]
risk_level: str = 'low'
daily_msg_cap: int = 0
daily_msg_count: int = 0
daily_msg_reset_at: float = 0.0
egg_status: str = 'none'
notes: List[str]
last_updated: float = 0.0
to_dict()[source]

Serialize this bond entry to a plain JSON-safe dict.

Flattens every field into a dictionary, truncating notes to the last 20 entries to bound payload size. The result is the wire/storage form of a bond used both for Redis persistence and for API responses.

This is consumed by AttachmentBondLedger.save_bond(), which wraps it in json.dumps before writing the abl:{user_id} Redis key, and by the Observation Deck read paths (web/ncm_chart_api.py bond endpoints and tools/psy_ops_tools.py) that expose bond state to admins. It has no side effects of its own.

Returns:

All bond fields keyed by name, with notes capped to the most recent 20 entries.

Return type:

Dict[str, Any]

classmethod from_dict(data)[source]

Reconstruct a BondEntry from a previously serialized dict.

Inverse of to_dict(), applying per-field defaults (matching the dataclass defaults) so partial or legacy payloads still deserialize cleanly rather than raising KeyError.

This is called by AttachmentBondLedger.load_bond() after a Redis GET on abl:{user_id} and json.loads of the stored value, to rehydrate the bond into the in-memory cache. It performs no I/O.

Parameters:

data (Dict[str, Any]) – Mapping of bond field names to values, as produced by to_dict(); missing keys fall back to defaults.

Returns:

A bond entry populated from data.

Return type:

BondEntry

class attachment_ledger.BondEvent(timestamp, event_type, detail, data=<factory>)[source]

Bases: object

A single lifecycle event in the attachment bond ledger.

A timestamped record of one notable transition in a user’s bond – a phase transition, risk-level change, admin action, or ULM spike – carrying a human-readable detail string plus a structured data payload. Events form the append-only audit trail surfaced on the Observation Deck.

Instances are built inside AttachmentBondLedger.update_from_detection(), AttachmentBondLedger.set_rate_limit(), and AttachmentBondLedger.add_note(), then serialized via to_dict() and pushed onto the abl:events:{user_id} Redis list by AttachmentBondLedger._log_event(). update_from_detection also returns the phase-transition event to its caller in limbic_system/coordinator.py. The dataclass itself performs no I/O.

Parameters:
timestamp: float
event_type: str
detail: str
data: Dict[str, Any]
to_dict()[source]

Serialize this lifecycle event to a JSON-safe dict.

Produces the storage form of a single bond event (timestamp, type, human-readable detail, and structured data payload).

This is consumed by AttachmentBondLedger._log_event(), which json.dumps the result and LPUSHes it onto the abl:events:{user_id} Redis list; the same shape is later returned by AttachmentBondLedger.get_events(). It has no side effects.

Returns:

The event’s timestamp, event_type, detail, and data fields.

Return type:

Dict[str, Any]

class attachment_ledger.AttachmentBondLedger(redis_client=None)[source]

Bases: object

Redis-backed per-user attachment bond lifecycle tracker.

All Redis operations are async to match the limbic_system’s async Redis client. Methods that don’t need Redis use the in-memory cache synchronously for the hot path.

Stores bond state in abl:{user_id} and event log in abl:events:{user_id} using Redis DB12.

__init__(redis_client=None)[source]

Initialize the ledger with an optional async Redis client (DB12).

Stores the supplied async Redis client (expected to point at DB12, alongside the limbic data) and sets up an empty in-memory BondEntry cache keyed by user_id. When no client is given the ledger still works as a pure in-process cache, but load_bond / save_bond and the event log become no-ops for persistence.

Constructed by limbic_system/coordinator.py (the long-lived per-process ledger) and ad hoc in the message send path (message_processor/generate_and_send.py), the web Observation Deck (web/ncm_chart_api.py), and the admin tools (tools/psy_ops_tools.py).

Parameters:

redis_client – An async Redis client bound to DB12, or None to run cache-only without persistence.

get_bond(user_id)[source]

Get or lazily create a bond entry from the in-memory cache.

Synchronous hot-path accessor: it never touches Redis, so it is safe to call from latency-sensitive code such as rate-limit checks. On a cache miss it constructs a fresh BondEntry (stamped with the current time) and stores it in the cache; any persisted state in abl:{user_id} is ignored. Use load_bond() instead when Redis-backed retrieval is required.

Called internally by check_rate_limit() and increment_msg_count(); both rely on the in-memory entry for speed.

Parameters:

user_id (str) – The user whose bond entry to fetch or create.

Returns:

The cached entry for user_id, freshly created if absent.

Return type:

BondEntry

async load_bond(user_id)[source]

Load a bond entry from the cache, falling back to Redis. # 🌀

The Redis-backed counterpart to get_bond(). Returns the cached entry when present; otherwise issues an async GET on abl:{user_id}, json.loads the value, and rehydrates it via BondEntry.from_dict(), populating the cache on the way. A missing key, a Redis error, or no client all degrade gracefully to a freshly constructed entry. Redis failures are caught and logged (with a truncated user id) rather than raised.

Called by save_bond() callers throughout this class (update_from_detection, set_rate_limit, add_note), by get_all_bonds() during the scan, and externally by the Observation Deck (web/ncm_chart_api.py) and admin tools (tools/psy_ops_tools.py).

Parameters:

user_id (str) – The user whose bond entry to load.

Returns:

The hydrated entry, or a new default entry if nothing was persisted or the load failed.

Return type:

BondEntry

async save_bond(entry)[source]

Persist a bond entry to the cache and Redis. # ♾️

Stamps entry.last_updated with the current time, writes the entry into the in-memory cache, and – when a Redis client is configured – SETs the serialized form (BondEntry.to_dict() wrapped in json.dumps) under the abl:{user_id} key. The cache update always happens; the Redis write is best-effort and any failure is caught and logged rather than raised, so an unreachable Redis never breaks the calling flow.

Called at the end of every mutating ledger operation: update_from_detection(), increment_msg_count(), set_rate_limit(), and add_note().

Parameters:

entry (BondEntry) – The bond entry to persist; its last_updated field is overwritten as a side effect.

Return type:

None

Returns:

None.

async get_all_bonds()[source]

Return every known bond, hydrating the cache from Redis if cold. # ♾️

Bulk accessor for admin/dashboard views. When the cache is empty and Redis is available it SCANs for abl:* keys in batches of 100, skips the abl:events:* event-log keys, and calls load_bond() for each user id to populate the cache before returning. If the cache is already warm it is returned directly without scanning. Scan errors are caught and logged, yielding whatever is already cached.

Called by the Observation Deck endpoints in web/ncm_chart_api.py and by the admin survey tool in tools/psy_ops_tools.py.

Returns:

All cached bond entries (a snapshot of the cache values after any scan-driven hydration).

Return type:

List[BondEntry]

async update_from_detection(user_id, detection, ulm_vector)[source]

Update a user’s bond from entrainment-detector output. # 💀🔥

The main write path of the ledger. Given a detection dict and the latest ULM vector, it loads the bond, diffs the incoming phase, risk level, archetype, and egg status against the stored state, and records a BondEvent for any phase transition or risk-level change. It also refreshes metadata (archetype, egg status, turn count) and ratchets up the per-node peak_nodes high-water marks from the ULM vector.

Side effects: calls load_bond() (read), _log_event() for each transition or risk change (appends to abl:events:{user_id}), and save_bond() at the end (writes abl:{user_id}); phase transitions are also logged via the module logger. Invoked from the limbic coordinator (limbic_system/coordinator.py) after entrainment detection runs.

Parameters:

  • user_id (str) – The user whose bond is being updated.

  • detection (Dict[str, Any]) – Entrainment-detector output, read for phase, risk_level, childhood_archetype, egg_status, confidence, signals, and turn_count.

  • ulm_vector (Dict[str, float]) – Per-node ULM activations used to update the peak_nodes high-water marks.

Returns:

The phase-transition event when the phase changed, otherwise None. (Risk-change events are logged but not returned.)

Return type:

Optional[BondEvent]

check_rate_limit(user_id)[source]

Check whether a user has hit their daily message cap. # ♾️

Synchronous hot-path guard: it reads the bond via get_bond() (cache only, no Redis) so it can run cheaply on every inbound message. A cap of 0 means unlimited and short-circuits to allowed. Otherwise it lazily resets the daily counter when more than 86400 seconds have elapsed since the last reset, then compares the running count against the cap.

Note that the lazy reset mutates the in-memory BondEntry but is not persisted here – increment_msg_count() performs the durable update. Called from the message send path (message_processor/generate_and_send.py) and the admin tools (tools/psy_ops_tools.py).

Parameters:

user_id (str) – The user whose rate-limit status to evaluate.

Returns:

A mapping with allowed (bool), remaining (int, -1 when unlimited), cap (int, 0 meaning unlimited), and count (int, messages sent in the current window).

Return type:

Dict[str, Any]

async increment_msg_count(user_id)[source]

Increment a user’s daily message counter and persist it.

The durable companion to check_rate_limit(). Reads the cached entry via get_bond(), performs the same lazy 86400-second daily-window reset, bumps daily_msg_count by one, and writes the result through save_bond() (cache plus abl:{user_id} in Redis).

Called from the limbic coordinator (limbic_system/coordinator.py) after a message is accepted, keeping the persisted count in step with the in-memory view that the rate-limit check reads.

Parameters:

user_id (str) – The user whose message counter to increment.

Return type:

None

Returns:

None.

async set_rate_limit(user_id, cap)[source]

Set a user’s daily message cap as an admin action.

Loads the bond via load_bond(), overwrites daily_msg_cap with the new value, and records the change as an admin_action BondEvent capturing the old and new caps. The event is appended through _log_event() (abl:events:{user_id}), the entry is persisted via save_bond() (abl:{user_id}), and the change is logged via the module logger.

Invoked from the Observation Deck admin endpoint (web/ncm_chart_api.py) and the admin tooling (tools/psy_ops_tools.py).

Parameters:

  • user_id (str) – The user whose cap is being set.

  • cap (int) – New daily message cap; 0 means unlimited.

Return type:

None

Returns:

None.

async get_events(user_id, limit=50)[source]

Retrieve a user’s most recent bond events. # 🔥

Reads up to limit newest entries from the abl:events:{user_id} Redis list (LRANGE from index 0), decoding bytes and json.loads-ing each stored BondEvent.to_dict() payload back into a plain dict. Because _log_event() LPUSHes, index 0 is the newest event, so results come back newest-first. With no Redis client or on a Redis error it returns an empty list (errors are logged, not raised).

Called by the Observation Deck (web/ncm_chart_api.py) and the admin tools (tools/psy_ops_tools.py) to render a user’s lifecycle history.

Parameters:

  • user_id (str) – The user whose events to fetch.

  • limit (int) – Maximum number of events to return (default 50).

Returns:

Up to limit event dicts, newest first; empty when nothing is stored or Redis is unavailable.

Return type:

List[Dict[str, Any]]

async add_note(user_id, note)[source]

Append a timestamped admin note to a user’s bond.

Loads the bond via load_bond(), appends the note (prefixed with a %Y-%m-%d %H:%M timestamp) to its notes list, records the addition as an admin_action BondEvent via _log_event() (abl:events:{user_id}), and persists the entry through save_bond() (abl:{user_id}). Note that BondEntry.to_dict() caps stored notes to the most recent 20 on serialization.

This is a public admin helper; no internal callers were found at this time (it is available for admin tooling / the Observation Deck to invoke).

Parameters:

  • user_id (str) – The user whose bond to annotate.

  • note (str) – Free-text admin note; the event detail truncates it to the first 100 characters.

Return type:

None

Returns:

None.