]i'dZddlmZddlZddlmZddlmZmZm Z m Z m Z ddl m Z eGddZGd d Zy) uy core/coherence/redis_master_state.py RedisMasterState — Versioned OCC (Optimistic Concurrency Control) state store. Uses Redis WATCH + MULTI/EXEC (pipeline transaction) to ensure that concurrent agents cannot corrupt shared state. The OCC pattern: 1. WATCH the key (register a "dirty" detector on the key) 2. GET current state, verify the caller's expected version matches 3. MULTI (begin transaction) 4. SET new state (version+1, patched data) 5. EXEC (commit) — Redis returns None if the watched key was modified between WATCH and EXEC, triggering a WatchError in redis-py If two agents concurrently read version 5 and both try to write version 6, only the first EXEC succeeds; the second raises WatchError → CommitResult(conflict=True). State structure stored in Redis: {"version": , "data": } Simple patch application (not full RFC 6902 — StateDelta handles that): {"op": "replace", "path": "/key", "value": v} → data[key] = v {"op": "add", "path": "/key", "value": v} → data[key] = v {"op": "remove", "path": "/key"} → del data[key] The "path" uses the key name without a leading "/" — i.e. "replace", path "/name" strips the leading slash and uses "name" as the dict key. # VERIFICATION_STAMP # Story: 6.02 # Verified By: parallel-builder # Verified At: 2026-02-25 # Tests: 9/9 # Coverage: 100% ) annotationsN) dataclass)AnyDictListOptionalTuple) WatchErrorc0eZdZUdZded<ded<ded<y) CommitResultaResult of a commit_patch or initialize_state operation. Attributes: success: True if the write committed successfully. new_version: The version number after a successful commit (version+1). 0 when success=False. conflict: True when a concurrent writer modified the key between our WATCH and EXEC, causing the transaction to abort. boolsuccessint new_versionconflictN)__name__ __module__ __qualname____doc____annotations__:/mnt/e/genesis-system/core/coherence/redis_master_state.pyr r 4sMNrr ceZdZdZdZd dZd dZeddZeddZ eddZ ddZ dd Z dd Z y )RedisMasterStatea Versioned, OCC-protected Redis state store for multi-agent coherence. Each session's state is stored at: genesis:state:master: Value format (JSON): {"version": , "data": } Usage:: rms = RedisMasterState(redis_client) version, data = await rms.get_snapshot("sess-abc") result = await rms.commit_patch("sess-abc", version, patch_ops) if result.conflict: # Retry: re-read snapshot and rebuild patch ... zgenesis:state:master:c||_y)zm Args: redis_client: An async redis.asyncio.Redis instance (or compatible mock). N)redis)self redis_clients r__init__zRedisMasterState.__init__`s " rc"|j|S)z(Build the Redis key for a given session.) KEY_PREFIX)r session_ids r_keyzRedisMasterState._keyks//":,//rcddl}|j|}|D]g}|jd}|jdd}|jd}|dvr|jd||<P|d k(sV|j |di|S) u Apply a simplified patch list to a data dict. Operations supported: "replace" — set data[key] = value (key must already exist is NOT enforced here; simple replace also creates the key) "add" — set data[key] = value "remove" — delete data[key] The "path" field is an RFC 6902-style JSON Pointer (e.g. "/name"). The leading "/" is stripped to derive the flat dict key. Returns a new dict — does NOT mutate the input. rNoppath/)replaceaddvalueremove)copydeepcopygetlstrippop)datapatchr.resultop_dictr&r'keys r_apply_patch_simplez$RedisMasterState._apply_patch_simpleos t$ &GT"B;;vr*D++c"C''%kk'2s x 3% & rc6tj||ddS)zSerialize state to JSON string.versionr3),:) separators)jsondumpsr:s r_encodezRedisMasterState._encodeszzgt<TTrc@tj|}|d|dfS)z+Deserialize JSON string to (version, data).r;r3)r?loads)rawstates r_decodezRedisMasterState._decodes% 3i %-00rcK|j|}|jj|d{}|sdifS|j|S7w)a Return the current (version, data) for a session. For a fresh session that has never been initialized, returns (0, {}). Args: session_id: Unique identifier for the session. Returns: Tuple of (version: int, data: dict). Nr)r$rr0rF)rr#r7rDs r get_snapshotzRedisMasterState.get_snapshotsKii #JJNN3''r7N||C  (s0AAAc0K|j|}|jj4d{} |j|d{|j |d{}|r|j |\}}ndi}}||k7r7|j d{tdddcdddd{S|j||} |dz} |j| | } |j|j|| |jd{td| dcdddd{S777777.7#t$r#tdddcYcdddd{7SwxYw#1d{7swYyxYww)ud Apply patch to state using Optimistic Concurrency Control. OCC flow: 1. WATCH key — Redis monitors it for external changes. 2. GET current state. 3. Check that current version == expected version. If not → version mismatch → return conflict (not a WatchError, just a version check failure — still conflict=True). 4. Apply patch to produce new data. 5. MULTI (begin transaction). 6. SET new state (version+1, new_data). 7. EXEC — commits if key has not been modified since WATCH. WatchError from redis-py → conflict=True. Args: session_id: Unique identifier for the session. version: The version the caller read from get_snapshot(). Must match the current version in Redis for the commit to proceed. patch: List of simplified patch operations (see _apply_patch_simple). Returns: CommitResult with success, new_version, conflict fields. NrFTrrr)r$rpipelinewatchr0rFunwatchr r8rAmultisetexecuter ) rr#r;r4r7piperDcurrent_version current_datanew_datarnew_raws r commit_patchzRedisMasterState.commit_patchs>ii #::&&(" Q" QD! Qjjo%%!HHSM)48LL4E1O\45r\O#g-,,.(('1tT!" Q" Q" Q& 33L%H%k ,,{H= g&lln$$#DkTYZ=" Q" Q" Q&*)" Q8%9" Q@ Q#Eq4PPE" Q" Q" Q@ QA" Q" Q" Qs0FEFFE E E%E&7EE E/ F;E <FAEE E1 F=E>FEE E FEFE>)F* F6E97F=E>>FFF FFcK|j|}|jd|}|jj||dd{}|rt dddSt dddS7"w)a Set the initial state for a session (version=1). Uses Redis SET NX (set-if-not-exists) to ensure idempotency. If the key already exists, returns conflict=True. Args: session_id: Unique identifier for the session. initial_data: The initial data dict to store. Returns: CommitResult(success=True, new_version=1, conflict=False) on success. CommitResult(success=False, new_version=0, conflict=True) if key exists. rKT)nxNFrJr)r$rArrPr )rr# initial_datar7rVr5s rinitialize_statez!RedisMasterState.initialize_statesi&ii #,,q,/zz~~c7t~<< !eL L 1tL L =sAA,A*#A,N)rrreturnNone)r#strr\r^)r3Dict[str, Any]r4List[Dict[str, Any]]r\r_)r;rr3r_r\r^)rDz bytes | strr\Tuple[int, Dict[str, Any]])r#r^r\ra)r#r^r;rr4r`r\r )r#r^rZr_r\r )rrrrr"r r$ staticmethodr8rArFrHrWr[rrrrrJs&)J"0<UU11!$CQCQCQ$ CQ  CQJMM%M  Mrr)r __future__rr? dataclassesrtypingrrrrr redis.exceptionsr r rrrrrgsF#J# !33'     *EMEMr