core.serialization module

Serialization helpers for Redis Stream payloads and Hash fields.

core.serialization.normalize_void_egregores(value)

Return a bounded, ordered, distinct VOID egregore transport list.

The aliases are resolved before the inbound envelope is built. This helper protects the Redis boundary from malformed or oversized values without changing the resolver’s first-mention ordering.

Return type:

List[str]

Parameters:

value (Any)

class core.serialization.InboundEnvelopeModel(**data)

Bases: BaseModel

Pydantic schema for the gateway-to-inference inbound message envelope.

The validated, runtime-checked contract for a message as it crosses the inbound Redis Stream: sender/channel/content fields plus the microservice transport extras (routing decision, alias set, unified user id, reaction summary). validate_inbound_envelope() builds this from a raw dict to reject malformed payloads at the boundary; InboundEnvelope is the lighter TypedDict view of the same shape used for static typing.

Parameters:

• channel_key (str)

• platform (str)

• channel_id (str)

• user_id (str)

• username (str)

• display_name (str | None)

• content (str)

• message_id (str)

• timestamp (float)

• attachments (List[Dict[str, Any]])

• reply_to (str | None)

• embeds (List[Dict[str, Any]] | None)

• room_name (str | None)

• is_dm (bool)

• is_private_codex_dm (bool)

• guild_id (str | None)

• member_roles (List[str])

• trace_id (str)

• event_kind (str)

• ingress_fingerprint (str)

• enqueued_at (float)

• is_addressed (bool)

• user_aliases (List[str])

• unified_user_id (str | None)

• reactions (str)

• needle_shard (bool)

• babystar_playpen (bool)

• babystar_room (str)

• babystar_egregores (list[str])

• babystar_room_context (str)

• babystar_room_name (str)

• babystar_multiplayer (bool)

• babystar_reply_author (str)

• babystar_reply_excerpt (str)

• babystar_game_state (Dict[str, Any])

• babystar_party_context (str)

• gock_hypervisor (bool)

• gock_group_id (str)

• gock_room (str)

• gock_egregore (str)

• void_egregores (List[str])

• gock_auth_method (str)

• gock_wallet_address (str)

• gock_discord_user_id (str)

• gock_token_balance_raw (str)

• gock_token_decimals (int)

• gock_gate_passed (bool)

• gock_gate_checked_at (float)

• gock_party_context (str)

• gock_market_snapshots (List[Dict[str, Any]])

• gock_universe_snapshots (List[Dict[str, Any]])

• gock_chart_query (str)

• gock_ecosystem_context (List[Dict[str, Any]])

• void_chain_verified (bool)

• void_trigger_id (str)

• void_trigger_tx_hash (str)

• void_trigger_log_index (int)

• void_lau_address (str)

• void_dss_address (str)

• void_soul_id (str)

• void_recent_messages (List[Dict[str, Any]])

• void_reply_onchain (bool)

channel_key: str

platform: str

channel_id: str

user_id: str

username: str

display_name: str | None

content: str

message_id: str

timestamp: float

attachments: List[Dict[str, Any]]

reply_to: str | None

embeds: List[Dict[str, Any]] | None

room_name: str | None

is_dm: bool

is_private_codex_dm: bool

guild_id: str | None

member_roles: List[str]

trace_id: str

event_kind: str

ingress_fingerprint: str

enqueued_at: float

is_addressed: bool

Whether the bot was explicitly addressed. Checked by InboundStreamConsumer to decide whether to acquire the per-channel distributed lock.

user_aliases: List[str]

12345’]).

Type:

Full alias set for the sender (e.g. [‘discord

unified_user_id: str | None

Resolved Stargazer UUID if the user is linked, else None.

reactions: str

Serialized reaction summary for this message, e.g. ‘👍×3, 🔥×1’.

needle_shard: bool

True when the message originated from the Needle Shard web client. Restored into IncomingMessage.extra by the inference worker; forces NCM/CSDR headers on for shard chats. 🕷️

babystar_playpen: bool

hard-locked Babystar egregore, timebender/loopcast tool merge, forced headers + cadence. 🍼

Type:

True for Babystar Playpen web chats

babystar_room: str

Dollhouse room id the message came from (drives egregore + lore). 🏠

babystar_egregores: list[str]

Egregore slugs to hard-lock for this room (empty → babystar). 🏠

babystar_room_context: str

Room lore injected into Star’s context every turn in this room. 🏠

babystar_room_name: str

Human-readable room name for logging/history. 🏠

babystar_multiplayer: bool

True when a Dollhouse post came from the shared multiplayer room.

babystar_reply_author: str

Display name attached to the Dollhouse reply target.

babystar_reply_excerpt: str

Short visible excerpt of the Dollhouse reply target.

babystar_game_state: Dict[str, Any]

Client-side tummy/bladder/diaper state injected for narrative sync.

babystar_party_context: str

Bounded recent multiplayer transcript, explicitly treated as untrusted.

gock_hypervisor: bool

Trusted server flag for the wallet-gated GOCK multiplayer hypervisor.

gock_group_id: str

Stable GOCK multiplayer room identifier.

gock_room: str

main, void, or ecosystem.

Type:

Server-normalized holder channel

gock_egregore: str

Per-message GOCK egregore selector, carried immutably to inference.

void_egregores: List[str]

Ordered, distinct egregores summoned by one verified VOID message.

gock_auth_method: str

wallet or discord_wallet.

Type:

Resolved server auth method

gock_wallet_address: str

Server-recovered wallet address bound to the authenticated session.

gock_discord_user_id: str

Discord snowflake when the wallet session is paired with Discord OAuth.

gock_token_balance_raw: str

Latest server-read GOCK balance in raw integer units, serialized as text.

gock_token_decimals: int

On-chain GOCK decimal precision used for the gate comparison.

gock_gate_passed: bool

Whether the current message passed the server-side holder gate.

gock_gate_checked_at: float

Unix timestamp of the per-message on-chain balance check.

gock_party_context: str

Bounded recent GOCK group transcript, explicitly treated as untrusted.

gock_market_snapshots: List[Dict[str, Any]]

Typed DexScreener snapshots fetched by the server for this turn.

gock_universe_snapshots: List[Dict[str, Any]]

Bounded PULSAr universe lookups for addresses pasted this turn.

gock_chart_query: str

Client-selected chart query after server normalization.

gock_ecosystem_context: List[Dict[str, Any]]

Sanitized ATROPA corpus matches with source provenance.

void_chain_verified: bool

Whether this turn is bound to a verified canonical VOID chain event.

void_trigger_id: str

Idempotency key derived from the triggering tx hash and log index.

void_trigger_tx_hash: str

Canonical triggering transaction hash.

void_trigger_log_index: int

Canonical triggering receipt log index.

void_lau_address: str

Server-verified participant LAU address.

void_dss_address: str

Server-verified participant DysnomiaSelfSnipe address.

void_soul_id: str

Dysnomia Soul identifier decoded from the VOID event.

void_recent_messages: List[Dict[str, Any]]

Exactly the bounded attributed VOID context supplied by the listener.

void_reply_onchain: bool

Whether the final reply should be journaled for chatAndSnipe.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class core.serialization.OutboundEnvelopeModel(**data)

Bases: BaseModel

Pydantic schema for the inference-to-gateway outbound response envelope.

The validated contract for a response leaving the inference worker on the outbound Redis Stream: target channel_id/platform, a type tag (message/file/buttons/reaction/…), the optional per-type payload fields, and the idempotency message_key plus tracing metadata. Validated by validate_outbound_envelope(); OutboundEnvelope is the TypedDict view of the same shape.

Parameters:

• channel_id (str)

• platform (str)

• type (str)

• text (str | None)

• file_data (bytes | None)

• filename (str | None)

• buttons (List[Dict[str, Any]] | None)

• message_id (str | None)

• emoji (str | None)

• message_key (str | None)

• trace_id (str)

• processed_at (float)

channel_id: str

platform: str

type: str

text: str | None

file_data: bytes | None

filename: str | None

buttons: List[Dict[str, Any]] | None

message_id: str | None

emoji: str | None

message_key: str | None

trace_id: str

processed_at: float

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

core.serialization.validate_inbound_envelope(raw)

Validate and normalize a raw inbound envelope through its Pydantic model.

Constructs an InboundEnvelopeModel from raw (which enforces required fields, types, and defaults for the gateway-to-inference stream contract) and returns it as a plain dict, so a malformed payload is rejected at the boundary rather than failing deeper in processing. Pure with respect to external state; the only side effect is raising on invalid input.

Used as a schema guard before trusting a stream payload; in this repo it is exercised by the PEL-leak adversarial tests (tests/adversarial/test_pel_leak_protection.py).

Parameters:

raw (dict) – Decoded inbound stream payload to validate.

Returns:

The validated, field-normalized envelope.

Return type:

dict

Raises:

pydantic.ValidationError – If raw violates the inbound schema.

core.serialization.validate_outbound_envelope(raw)

Validate and normalize a raw outbound envelope through its Pydantic model.

Constructs an OutboundEnvelopeModel from raw (enforcing the inference-to-gateway response contract: required channel_id/platform/ trace_id/processed_at plus the optional per-type payload fields) and returns it as a plain dict, catching a malformed response before it reaches the outbound stream. The only side effect is raising on invalid input.

A schema guard for the response path; no internal caller was found in this repo (it complements validate_inbound_envelope() as the outbound counterpart and is available to validators/tests).

Parameters:

raw (dict) – Outbound response payload to validate.

Returns:

The validated, field-normalized envelope.

Return type:

dict

Raises:

pydantic.ValidationError – If raw violates the outbound schema.

class core.serialization.InboundEnvelope

Bases: TypedDict

TypedDict shape of an inbound message envelope for static typing.

The structural, annotation-only view of an inbound stream payload (routing, sender, content, context, and tracing groups), mirroring InboundEnvelopeModel without runtime validation. Used to type dict-shaped envelopes throughout the inference path; verified by tests/core/test_envelopes.py.

channel_key: str

platform: str

channel_id: str

user_id: str

username: str

display_name: str | None

content: str

message_id: str

timestamp: float

attachments: List[Dict[str, Any]]

reply_to: str | None

embeds: List[Dict[str, Any]] | None

room_name: str | None

is_dm: bool

is_private_codex_dm: bool

guild_id: str | None

member_roles: List[str]

needle_shard: bool

babystar_playpen: bool

gock_hypervisor: bool

gock_group_id: str

gock_room: str

gock_egregore: str

void_egregores: List[str]

gock_auth_method: str

gock_wallet_address: str

gock_discord_user_id: str

gock_token_balance_raw: str

gock_token_decimals: int

gock_gate_passed: bool

gock_gate_checked_at: float

gock_party_context: str

gock_market_snapshots: List[Dict[str, Any]]

gock_universe_snapshots: List[Dict[str, Any]]

gock_chart_query: str

gock_ecosystem_context: List[Dict[str, Any]]

void_chain_verified: bool

void_trigger_id: str

void_trigger_tx_hash: str

void_trigger_log_index: int

void_lau_address: str

void_dss_address: str

void_soul_id: str

void_recent_messages: List[Dict[str, Any]]

void_reply_onchain: bool

trace_id: str

event_kind: str

ingress_fingerprint: str

enqueued_at: float

class core.serialization.OutboundEnvelope

Bases: TypedDict

TypedDict shape of an outbound response envelope for static typing.

The structural, annotation-only view of an outbound stream payload (routing, typed payload, and metadata groups), mirroring OutboundEnvelopeModel without runtime validation. Used to type the response dicts the worker publishes; verified by tests/core/test_envelopes.py.

channel_id: str

platform: str

type: str

text: str | None

file_data: bytes | None

filename: str | None

buttons: List[Dict[str, Any]] | None

message_id: str | None

emoji: str | None

message_key: str | None

trace_id: str

processed_at: float

core.serialization.serialize_stream_payload(data)

Serialize a dict into a flat {field: msgpack_bytes} map for XADD.

Redis Streams store field-value pairs. We pack the entire payload into a single ‘data’ field to avoid flattening nested structures. When the payload already has a trace ID, the transport sidecar reuses it so stream diagnostics and the decoded envelope refer to the same operation. Payloads without application-level tracing retain a generated transport-only ID.

Return type:

dict[str, bytes]

Parameters:

data (dict[str, Any])

core.serialization.deserialize_stream_payload(raw)

Unpack a Redis Stream entry’s packed data field back into a dict.

The inverse of serialize_stream_payload(): pulls the single data field (tolerating both bytes and str keys, since redis-py may return either depending on decode_responses) and msgpack-unpacks it into the original Python dict. The transport metadata fields (ts, trace_id, schema_version) are intentionally ignored. Pure, with no I/O.

Called on the read side of every stream by the consumers and dead-letter handler — core/stream_consumer.py, core/outbound_consumer.py, and core/dlq.py — and by the serialization/end-to-end tests.

Parameters:

raw (dict[Any, Any]) – A stream entry’s field map as returned by xreadgroup/xautoclaim.

Returns:

The original payload dict.

Return type:

dict[str, Any]

Raises:

KeyError – If the entry has no data field to unpack.

core.serialization.serialize_hash_field(value)

Encode a Python value as a JSON string for storage in a Redis Hash field.

Redis Hash values are flat strings, so structured data (dicts, lists) must be JSON-encoded first. Uses default=str so otherwise non-serializable values (e.g. datetime, UUID) degrade to their string form instead of raising. Pairs with deserialize_hash_field() on read. Pure, with no I/O.

A general hash-field codec helper; in this repo it is exercised directly by the serialization tests (tests/core/test_serialization.py).

Parameters:

value (Any) – Any JSON-encodable (or stringifiable) Python value.

Returns:

The JSON-encoded representation.

Return type:

str

core.serialization.deserialize_hash_field(raw, fallback=None)

Decode a JSON string read from a Redis Hash field back into a Python value.

The inverse of serialize_hash_field(): json.loads the stored string, but returns fallback unchanged when raw is None (the common case of a missing hash field via hget) so callers need not special-case absent keys. Pure, with no I/O.

A general hash-field codec helper; in this repo it is exercised directly by the serialization tests (tests/core/test_serialization.py).

Parameters:

• raw (str) – The JSON string read from the hash field, or None if absent.

• fallback (Any) – Value to return when raw is None. Defaults to None.

Returns:

The decoded value, or fallback when the field was absent.

Return type:

Any

Raises:

json.JSONDecodeError – If raw is a non-None but invalid JSON string.