`ik*dZddlmZddlZddlmZmZddlmZm Z m Z eje Z dZeGddZGd d Zy) u, core/coherence/occ_commit.py OCCCommitEngine — Barrier Sync + OCC Write. Orchestrates the full commit cycle for a multi-agent swarm round: 1. BARRIER: wait for all expected workers to submit their StateDelta proposals to the StagingArea (uses StagingArea.wait_for_all). 2. MERGE: pass collected deltas through the SemanticMergeInterceptor to resolve contradictions and produce a single merged patch. 3. OCC WRITE: commit the merged patch to RedisMasterState with OCC retry logic (max 3 attempts on version conflict). 4. SAGA: record outcome via saga_writer if provided. 5. LEDGER: emit a merge event to ColdLedger on success if provided. OCC retry loop (MAX_RETRIES = 3): On each attempt, the engine re-reads the current state snapshot (version + data) from RedisMasterState and re-tries commit_patch. If all retries are exhausted the commit returns OccCommitResult(success=False, saga_status="conflict_exhausted"). Dependency injection: All collaborators are injected via the constructor. None of them are required at import time — pass mocks in tests. Usage:: engine = OCCCommitEngine( staging_area=staging, merge_interceptor=merger, master_state=rms, saga_writer=saga, # optional cold_ledger=ledger, # optional ) result = await engine.execute_commit(session_id="sess-abc", expected_workers=3) assert result.success # VERIFICATION_STAMP # Story: 6.06 # Verified By: parallel-builder # Verified At: 2026-02-25 # Tests: 14/14 # Coverage: 100% ) annotationsN) dataclassfield)AnyListOptionalcbeZdZUdZded<eeZded<dZded <dZ ded <d Z d ed <y)OccCommitResultu Result of an OCCCommitEngine.execute_commit() call. Attributes: success: True when the patch was committed successfully. merged_patch: The merged patch list produced by the merge interceptor. Empty list when success=False. version: The new version number after a successful commit (i.e. old_version + 1). The last-known version on failure. retries: Number of OCC retry attempts made (0-based: 0 means the first attempt succeeded or failed without a conflict retry). saga_status: Saga outcome string: "completed" — committed successfully. "merge_failed" — merge interceptor returned failure. "conflict_exhausted" — all OCC retries failed. "unknown" — initial/unset value (should not appear in a finished result). boolsuccess)default_factoryz List[Any] merged_patchrintversionretriesunknownstr saga_statusN) __name__ __module__ __qualname____doc____annotations__rlistrrrr2/mnt/e/genesis-system/core/coherence/occ_commit.pyr r Cs;(M#D9L)9GSGS K rr cXeZdZdZ d ddZ d dZd dZd dZy) OCCCommitEngineu Barrier-sync + OCC commit coordinator for multi-agent swarm rounds. Constructor args: staging_area: StagingArea — collects worker deltas in Redis. merge_interceptor: Object with an async merge(deltas, current_state, version) method that returns an object with .success and .merged_patch attributes. master_state: RedisMasterState — versioned OCC state store. Must expose: - async get_snapshot(session_id) -> (version, data) - async commit_patch(session_id, version, patch) -> CommitResult(success, new_version, conflict) saga_writer: Optional object with async close_saga(session_id, status). cold_ledger: Optional object with sync write_event(session_id, event, data). NcJ||_||_||_||_||_y)N)stagingmergermaster saga_writerledger)self staging_areamerge_interceptor master_stater% cold_ledgers r__init__zOCCCommitEngine.__init__ws($ ' " &! rc\Ktjd|||jj||dd{}tjdt ||d}t t D]\}|jj|d{\}}|jj|||d{}|jsCtjd|||j|dd{td g||d cS|jj|||j d{}|jre|d z} tj#d || ||j|d d{|j%|dtd|j | |d cStjd||d zt _tj'dt ||j|dd{td g|t d S777e7+777(w)a Run the full barrier-sync + OCC commit cycle. Steps: 1. Wait for all expected_workers to submit deltas (barrier). 2. Merge all deltas via merge_interceptor. 3. Attempt OCC commit with up to MAX_RETRIES retries on conflict. 4. Record outcome to saga_writer (if provided). 5. Emit merge event to cold_ledger on success (if provided). Args: session_id: Session identifier shared by all workers. expected_workers: How many worker delta submissions to wait for. Returns: OccCommitResult describing the outcome of the commit cycle. z5OCCCommitEngine: waiting for %d workers on session %si`) timeout_msNz5OCCCommitEngine: collected %d delta(s) for session %srz8OCCCommitEngine: merge failed on session %s (attempt %d) merge_failedF)r rrrrz@OCCCommitEngine: committed session %s at version %d (attempt %d) completedsaga_committedTz:OCCCommitEngine: OCC conflict on session %s, attempt %d/%dzBBB&!!### . #'++":":G\%>%>#M$$%k  # &&z;???'' 4DE& !-!:!:'# +  NNL!   a5 t  J   z+?@@@,   a $&PSC@4 As9H,HAH,H'H,;H < # I3 s- AAA)NN) r(rr)rr*rr% Optional[Any]r+rSreturnNone)rBrrCrrTr )rBrrNrrTrU)rBrrRrrTrU)rrrrr,rJr=r@rrrr r es~,&*%) " " " " # " # "  "$t t t   t trr )r __future__rlogging dataclassesrrtypingrrr getLoggerrr4r9r r rrrr[s[,\#(&&   8 $   !! !Bppr