&i+dZddlmZddlZddlZddlmZddlmZddlm Z ddl Z ddl Z ddl Z eGddZ Gd d Zy) u6Cold Ledger — L4 Postgres read/write client for Genesis persistent context. Provides a typed, connection-pooled interface over the three Cold Ledger tables: - events : append-only event log per session - swarm_sagas : orchestration saga records with full lifecycle status Usage:: from core.storage.cold_ledger import ColdLedger, SwarmSaga ledger = ColdLedger(connection_params={ "host": "...", "port": 5432, "user": "...", "password": "...", "dbname": "genesis" }) event_id = ledger.write_event("session-uuid", "dispatch_start", {"agent": "forge"}) events = ledger.get_events("session-uuid", event_type="dispatch_start") saga_id = ledger.write_saga(SwarmSaga(...)) saga = ledger.get_saga(saga_id) ledger.close() Rules enforced (Genesis hardwired): - NO SQLite anywhere in this file - All SQL uses parameterised queries (%s placeholders — never f-strings) - Connection pool uses getconn/putconn in try/finally (no connection leaks) - All writes to the events table are append-only (no UPDATE) ) annotationsN) dataclass)datetime)OptionalcXeZdZUdZded<ded<ded<ded<d ed <ded <d ed <y) SwarmSagaz9Typed representation of one row in the swarm_sagas table.strsaga_id session_iddictorchestrator_daglistproposed_deltaszOptional[dict]resolved_statestatusr created_atN)__name__ __module__ __qualname____doc____annotations__1/mnt/e/genesis-system/core/storage/cold_ledger.pyrr1s,C LO"" KrrceZdZdZd dZdZddZ ddZ d ddZddZ dd Z dd Z dd Z e dd Zy) ColdLedgeruTyped Postgres client for the Genesis Cold Ledger (L4 storage layer). Thread-safe via ThreadedConnectionPool. Every method acquires a connection from the pool and returns it in a ``try/finally`` block — connection leaks are structurally impossible. Args: connection_params: Dict accepted by ``psycopg2.connect`` — keys: ``host``, ``port``, ``user``, ``password``, ``dbname``. Optionally ``sslmode`` and other libpq parameters. cZ||_tjjdi||_y)N) )_paramspsycopg2poolThreadedConnectionPool)selfconnection_paramss r__init__zColdLedger.__init__Ps&( MM 0 0 L:K L rc6|jjS)zCReturn a connection from the pool (caller must putconn in finally).)r"getconnr$s r_acquirezColdLedger._acquireZsyy  ""rc:|jj|y)z3Return a connection to the pool without closing it.N)r"putconn)r$conns r_releasezColdLedger._release^s $rc pttj}d}|j} |j 5}|j ||||t j|fddd|j|j||S#1swY,xYw#|j|wxYw)uAppend one event record to the events table. The write is strictly append-only — no UPDATE path exists. Args: session_id: UUID string of the owning session. event_type: Short type label, e.g. ``"dispatch_start"``. payload: Arbitrary JSON-serialisable dict stored as JSONB. Returns: The new event's UUID string (UUID4). zVINSERT INTO events (id, session_id, event_type, payload) VALUES (%s, %s::uuid, %s, %s)N) r uuiduuid4r*cursorexecutejsondumpscommitr.)r$r event_typepayloadevent_idsqlr-curs r write_eventzColdLedger.write_eventfs$tzz|$ - }}  # z:tzz'7JK  KKM MM$    MM$ s#B"*B+B"BB""B5Nc|d}|||f}nd}||f}|j} |jtjj5}|j |||j }ddd|j|D cgc] } t| c} S#1swY3xYw#|j|wxYwcc} w)uRead events for a session, ordered by creation time ascending. Args: session_id: UUID string of the session to query. event_type: Optional filter — only return events of this type. limit: Maximum number of rows to return (default 100). Returns: List of dicts with keys: ``id``, ``session_id``, ``event_type``, ``payload``, ``created_at``. NzSELECT id, session_id, event_type, payload, created_at FROM events WHERE session_id = %s::uuid AND event_type = %s ORDER BY created_at ASC LIMIT %szSELECT id, session_id, event_type, payload, created_at FROM events WHERE session_id = %s::uuid ORDER BY created_at ASC LIMIT %scursor_factory) r*r2r!extrasRealDictCursorr3fetchallr.r ) r$r r7limitr:paramsr-r;rowsrows r get_eventszColdLedger.get_eventss"  ! !*e4F !%(F}} HOO,J,JK &s C(||~ & MM$ &**cS ** & & MM$ +s)*B- #B!/B- C!B*&B--Cc Hd}|jtj|jnd}|j} |j 5}|j ||j |jtj|jtj|j||j|jfddd|j|j||j S#1swY6xYw#|j|wxYw)uPersist a SwarmSaga record to the swarm_sagas table. Args: saga: Fully populated ``SwarmSaga`` dataclass instance. Returns: The ``saga.saga_id`` string (unchanged — returned for caller convenience). zINSERT INTO swarm_sagas (saga_id, session_id, orchestrator_dag, proposed_deltas, resolved_state, status, created_at) VALUES (%s::uuid, %s::uuid, %s, %s, %s, %s, %s)N)rr4r5r*r2r3r r r rrrr6r.)r$sagar: resolved_jsonr-r;s r write_sagazColdLedger.write_sagas ? 04/B/B/NDJJt** +TX }}  #   4#8#89 4#7#78%    KKM MM$ ||!   MM$ s%DAN) r*r2r!r@rAr3fetchoner. _row_to_sagar )r$r r:r-r;rFs rget_sagazColdLedger.get_sagas ( }} HOO,J,JK %s C',lln % MM$  ;  c++ % % MM$ s"*B$$B"B$B!B$$B7cd}|j} |jtjj5}|j ||f|j }ddd|j|Dcgc]}|jt|c}S#1swYBxYw#|j|wxYwcc}w)zRetrieve all sagas associated with a session, ordered by creation time. Args: session_id: UUID string of the owning session. Returns: List of ``SwarmSaga`` dataclass instances (may be empty). zSELECT saga_id, session_id, orchestrator_dag, proposed_deltas, resolved_state, status, created_at FROM swarm_sagas WHERE session_id = %s::uuid ORDER BY created_at ASCr>N) r*r2r!r@rAr3rBr.rNr )r$r r:r-r;rErFs rget_sagas_by_sessionzColdLedger.get_sagas_by_sessions ' }} HOO,J,JK &s C*/||~ & MM$ 8<=!!$s),==  & & MM$ =s(*B/$B#"B/?!C#B,(B//CcR|jr|jjyy)zBShut down the connection pool, closing all underlying connections.N)r"closeallr)s rclosezColdLedger.closes 99 II    rc d}tt|dt|d||dxsi||dxsg||d|d|d S) zConvert a raw database row dict into a typed SwarmSaga instance. Handles JSONB columns that psycopg2 returns as already-parsed Python objects (dicts/lists) and also the string-fallback case for tests. cb|yt|ttfr|Stj|S)N) isinstancer rr4loads)vals r _parse_jsonz,ColdLedger._row_to_saga.._parse_json,s+{#d|, ::c? "rr r r rrrr)r r r rrrr)rr )rFrZs rrNzColdLedger._row_to_saga$sr #I'3|,-(-?)@AGR',=(>?E2&s+;'<=x=<(  r)r%r returnNone)r[r\)r r r7r r8r r[r )Nd)r r r7z Optional[str]rCintr[r)rIrr[r )r r r[zOptional[SwarmSaga])r r r[r)rFr r[r)rrrrr&r*r.r<rGrKrOrQrT staticmethodrNrrrrrCs  # !!! ! !L%) -+-+"-+ -+  -+f$L,6><!  rr)r __future__rr4r0 dataclassesrrtypingrr! psycopg2.poolpsycopg2.extrasrrrrrresL<# !     "x x r