5iW)dZddlmZddlZddlZddlmZddlmZddlm Z eje Z dZ eGdd ZGd d Zy) uEventSourcingStream — Immutable event append and replay for Genesis. Provides an event sourcing layer on top of the ColdLedger (L4 Postgres). Events are written append-only to ColdLedger and optionally published to a Redis Stream (``genesis:events``) for real-time consumption by subscribers. Usage:: from core.storage.event_sourcing import EventSourcingStream, GenesisEvent import uuid from datetime import datetime stream = EventSourcingStream(cold_ledger=ledger, redis_client=redis) event = GenesisEvent( id=str(uuid.uuid4()), session_id="session-uuid", event_type="dispatch_start", payload={"agent": "forge"}, version=1, created_at=datetime.utcnow(), ) stream.append(event) all_events = stream.replay("session-uuid") state = stream.get_current_state("session-uuid") Design notes: - Version ordering is the authoritative ordering mechanism (not created_at). Callers are responsible for assigning monotonically increasing version numbers. - Redis XADD is best-effort: a Redis failure logs an error but does NOT raise so ColdLedger (the source of truth) is never rolled back. - get_current_state folds events sequentially in version order — each event's payload is shallow-merged into the state dict (later versions win on key collision). - NO SQLite anywhere in this file (Genesis Rule 7). ) annotationsN) dataclass)datetime)Optionalzgenesis:eventscNeZdZUdZded<ded<ded<ded<ded <d ed <y ) GenesisEventuTyped representation of one event in the immutable event log. Fields ------ id : str UUID4 string — unique identifier for this event record. session_id : str UUID string of the owning Genesis session. event_type : str Short label describing the event, e.g. ``"dispatch_start"``. payload : dict Arbitrary JSON-serialisable dict containing event data. version : int Monotonically increasing integer assigned by the caller. Determines replay order — lower versions are applied first. created_at : datetime Wall-clock timestamp when the event was created (UTC). strid session_id event_typedictpayloadintversionr created_atN)__name__ __module__ __qualname____doc____annotations__4/mnt/e/genesis-system/core/storage/event_sourcing.pyrr9s(& GOO M Lrrc`eZdZdZd d dZd dZd dZ ddZddZddZ e dd Z y)EventSourcingStreamaAppend-only event stream backed by ColdLedger (L4) with optional Redis publish. Args: cold_ledger: A ``ColdLedger`` instance exposing ``write_event()`` and ``get_events()``. redis_client: Optional Redis client. When provided, each appended event is also published to the ``genesis:events`` Redis Stream via ``XADD``. Redis failures are logged but never raised. Nc ||_||_yN)ledgerredis)self cold_ledger redis_clients r__init__zEventSourcingStream.__init__fs! ! rc"|j|j|jjd|j}|j j |j|j||j|j||yy)ufWrite an event to ColdLedger and publish to Redis Stream. The ColdLedger write is mandatory and will raise on failure. The Redis XADD is best-effort — any exception is caught, logged, and does NOT prevent the ColdLedger record from persisting. Args: event: A fully populated ``GenesisEvent`` instance. event_idrr)r r rN) r rr isoformatrr write_eventr r r_publish_to_redis)r eventenriched_payloads rappendzEventSourcingStream.appendns}}**446 mm   ''''$   :: !  " "5*: ; "rc|jj|}|Dcgc]}|j|}}|jd|Scc}w)atReturn all events for a session, ordered by version ascending. Reads from ColdLedger (L4). Events are sorted by their embedded ``version`` field extracted from the stored payload. Args: session_id: UUID string of the session to replay. Returns: Ordered list of ``GenesisEvent`` instances (may be empty). c|jSr)r)es rz,EventSourcingStream.replay..s !))r)key)r get_events _row_to_eventsort)r r rowsroweventss rreplayzEventSourcingStream.replaysO{{%%j159:c$$$S):: + , ;sAcn|j|}|Dcgc]}|j|k\s|c}Scc}w)u%Return events with version >= from_version, ordered by version ascending. Useful for incremental state reconstruction — e.g. when a snapshot exists at version N and only events after N need to be replayed. Args: session_id: UUID string of the session to replay. from_version: Lower bound (inclusive) — events with version strictly less than this are excluded. Returns: Filtered, ordered list of ``GenesisEvent`` instances (may be empty). )r8r)r r from_version all_eventsr/s rreplay_from_versionz'EventSourcingStream.replay_from_versions1 [[, %Cal)BCCCs22c|j|}i}|D]D}|jjDcic] \}}|dvr||}}}|j|F|Scc}}w)uFold all events into a single state dict. Events are applied sequentially in version order. Each event's ``payload`` (excluding the internal bookkeeping keys ``event_id``, ``version``, ``created_at``) is shallow-merged into the accumulating state dict — later versions override earlier ones on key collision. Args: session_id: UUID string of the session whose state to compute. Returns: Merged state dict. Returns ``{}`` for an unknown or empty session. r%)r8ritemsupdate)r r r7stater*kv user_payloads rget_current_statez%EventSourcingStream.get_current_states}Z( 'E"MM//1AqAA1L LL & ' sAc t |jjt|j|j|j t |jtj|dy#t$rA}tjd|j|j|j|Yd}~yd}~wwxYw)zIPublish event to Redis Stream via XADD. Failures are logged, not raised.)r&r r rrurEventSourcingStream: Redis XADD failed for event %s (session=%s, version=%d) — event is in ColdLedger. Error: %sN) rxaddREDIS_STREAM_KEYr r r r rjsondumps Exceptionloggererror)r r*r+excs rr)z%EventSourcingStream._publish_to_rediss  JJOO %"'"2"2"'"2"2"5==1#zz*:;    LLQ       sA*A-- B767B22B7c6|jdi}t|trtj|}t |jdd}|jdxs|jd}t|trt j|}n't|t r|}nt j}|jdxst|jdd}t|t|dt|d ||| S) aConvert a raw ColdLedger row dict into a ``GenesisEvent``. ColdLedger rows have keys: ``id``, ``session_id``, ``event_type``, ``payload``, ``created_at``. The ``payload`` dict contains the enriched data we stored in ``append()``, including the original ``version`` and ``event_id``. Handles the case where psycopg2 has already parsed the JSONB column into a Python dict (normal production path) as well as the string fallback (some test/mock paths). rrrrr&r r r )r r r rrr) get isinstancer rHloadsrr fromisoformatutcnowr)r6rrcreated_at_rawrr&s rr3z!EventSourcingStream._row_to_events'')R( gs #jj)Ggkk)Q/0!\2Kcggl6K nc *!//?J  1'J!*J;;z*Dc#''$2C.D3|,-3|,-!   rr)returnNone)r*rrVrW)r r rVlist[GenesisEvent])r r r:rrVrX)r r rVr )r*rr+r rVrW)r6r rVr) rrrrr#r,r8r<rDr) staticmethodr3rrrrrU"<>"DD-0D D&<.( ( rr)r __future__rrHlogging dataclassesrrtypingr getLoggerrrKrGrrrrrr_s_#J# !   8 $$  Bt t r