core.game_event_stream module

Durable Redis-Stream transport for game art and turn-complete events.

The game agents used to consume Redis Pub/Sub notifications. Pub/Sub has no pending-entry list, so a process restart between publish and handling silently lost a whole-turn side effect. This module provides a small transactional outbox plus an at-least-once consumer:

  • one stable logical event_id maps to exactly one stream entry;

  • replaying the same payload returns the original stream receipt;

  • reusing the identity with a different payload is rejected;

  • consumers reclaim stale pending entries and ACK only after the handler has either completed or reached an explicitly terminal, non-replayable outcome.

  • pending/executing records stay persistent; terminal records receive an eight-day TTL only after both stream ACK and deletion succeed.

The streams intentionally have no MAXLEN trim. A count-based trim can delete entries that a consumer group has not processed yet. Operations may add an age reaper later, but only after checking every group’s delivered and pending watermarks.

class core.game_event_stream.DurableGameEventOutbox(redis, *, terminal_ttl_seconds=691200)

Bases: object

Atomically deduplicate and append game events to their Redis Stream.

Parameters:

  • redis (Any)

  • terminal_ttl_seconds (int)

async enqueue(event_type, event_id, payload)

Append exactly one entry for event_id or return its old receipt.

Return type:

GameEventEnqueueResult

Parameters:
class core.game_event_stream.DurableGameStreamConsumer(redis, *, stream, group, consumer_name, handler, min_idle_ms=60000, heartbeat_ms=20000, block_ms=1000, batch_size=8, terminal_ttl_seconds=691200, terminal_retention_keys=None)

Bases: object

Consumer-group runner with stale reclaim and ACK-after-processing.

Parameters:

  • redis (Any)

  • stream (str)

  • group (str)

  • consumer_name (str)

  • handler (GameEventHandler)

  • min_idle_ms (int)

  • heartbeat_ms (int)

  • block_ms (int)

  • batch_size (int)

  • terminal_ttl_seconds (int)

  • terminal_retention_keys (TerminalRetentionKeys | None)

async ensure_group()

Create the group at the beginning so pre-start entries are consumed.

Return type:

None

async stop()

Request that run() exit after the current blocking read.

Return type:

None

async run()

Continuously reclaim and consume until stopped or cancelled.

Return type:

None

async process_once()

Reclaim stale PEL entries, read new entries, and process a batch.

Return type:

int

exception core.game_event_stream.GameEventConflict

Bases: GameEventError

A logical event identity was reused for a different payload.

class core.game_event_stream.GameEventEnqueueResult(status, event_id, stream, stream_id, payload_fingerprint)

Bases: object

Receipt for one atomic game-event enqueue.

Parameters:

  • status (str)

  • event_id (str)

  • stream (str)

  • stream_id (str)

  • payload_fingerprint (str)

status: str
event_id: str
stream: str
stream_id: str
payload_fingerprint: str
as_dict()

Return a checkpoint/msgpack-safe receipt mapping.

Return type:

dict[str, str]

exception core.game_event_stream.GameEventError

Bases: RuntimeError

Base error for the durable game-event transport.

exception core.game_event_stream.GameEventTerminalError

Bases: GameEventError

The event cannot be replayed safely and should be terminally ACKed.

core.game_event_stream.default_game_event_id(event_type, payload)

Derive an id for non-checkpoint callers from the complete payload.

Return type:

str

Parameters:
core.game_event_stream.game_event_id(event_type, logical_operation_id)

Derive a bounded stable event id from a checkpoint/tool operation id.

Return type:

str

Parameters:

  • event_type (str)

  • logical_operation_id (str)