i%dZddlmZddlZddlZddlZddlZddlmZddl m Z ddl m Z ddl mZejeZeGdd ZGd d Zy) u core/merge/semantic_merge_interceptor.py SemanticMergeInterceptor — orchestrates conflict detection and Opus 4.6 resolution for concurrent StateDelta patches. Flow ---- 1. ConflictDetector.detect(deltas) → ConflictReport 2a. No conflicts → ConflictDetector.fast_merge(deltas) → MergeResult(used_opus=False) 2b. Conflicts detected → MergePromptBuilder.build(…) → Opus 4.6 → parsed JSON Opus response: {"resolved_patch": [...], "resolution_rationale": "..."} On parse failure → fast_merge(non_conflicting_deltas) (partial fallback) 3. Append merge event to data/observability/events.jsonl Opus is ONLY called when ConflictDetector reports at least one conflict. ) annotationsN) dataclass)Optional)ConflictDetector)MergePromptBuildercLeZdZUdZded<ded<ded<dZded <d Zd ed <y) MergeResultz3Outcome of a SemanticMergeInterceptor.merge() call.boolsuccesslist merged_patch used_opusNz Optional[str]resolution_rationalegfloat latency_ms)__name__ __module__ __qualname____doc____annotations__rr>/mnt/e/genesis-system/core/merge/semantic_merge_interceptor.pyr r 's*= MO*.-.Jrr cbeZdZdZ d ddZ d d dZd dZ d dZy) SemanticMergeInterceptora Orchestrates conflict detection + Opus conflict resolution for StateDelta patches submitted concurrently by swarm workers. Parameters ---------- opus_client: Optional client used to call Opus 4.6. Must expose one of: - ``generate_content_async(prompt) -> response`` (preferred async) - ``generate_content(prompt) -> response`` (sync fallback) Response objects must have a ``.text`` attribute *or* be directly str-coercible. Inject ``None`` to run in test-only mode where Opus is never reachable (conflicts will fall back to partial merge). events_path: Path to the JSONL file where merge events are appended. Defaults to ``"data/observability/events.jsonl"``. Nc\t|_t|_||_||_y)N)rdetectorrprompt_builder opus_client events_path)selfrr s r__init__z!SemanticMergeInterceptor.__init__Js) )* 02&&rc Ktj}|jj|}|jsc|jj |}tj|z dz}|j t|ddd|td|d|St|j} |jj||||} |j| d{} tj| } | d} t| t s!t#d t%| j&| j)d d } tj|z dz}|j t||dd|td| d| | S7#t*$r}t,j/d t%|j&|t|j0|jj |j0}tj|z dz}|j t||dd|td|d|cYd}~Sd}~wwxYww)a@ Merge *deltas* into a single resolved patch. Parameters ---------- deltas: List of StateDelta objects (or dicts with a ``patch`` key). current_state: The master state dict at the time of the merge request. version: Current state version number (forwarded to MergePromptBuilder). Returns ------- MergeResult ``success`` is always True (partial merge is used on Opus failure). ``used_opus`` reflects whether Opus was actually invoked. irTF) delta_countconflict_countresolvedrr)r r rrNresolved_patchz#resolved_patch must be a list, got r)r r rrru[Opus merge failed (%s: %s) — falling back to partial merge of %d non-conflicting delta(s))time monotonicrdetect has_conflicts fast_merge _record_eventlenr conflicting_pathsrbuild _call_opusjsonloads isinstancer ValueErrortyperget Exceptionloggerwarningnon_conflicting_deltas)r!deltas current_stateversionstartreportmergedlatencyr%promptopus_response_textparsedr' rationaleexcs rmergezSemanticMergeInterceptor.mergeXsT0 %%f-##]]--f5F~~'%/47G   K "  #"  V5565 ((.. wF(,v'>!> ZZ 23F#$45Nnd3 9$~:N:W:W9XY(.zz2H"'MI~~'%/47G   K-"  +%."  %"?4  NN.S ""F112  ]]--f.K.KLF~~'%/47G   K-"  #"  % sJB4I72F)F*B)FIF IB*I I IIIc,K|j tdt|jdr$|jj|d{}n|jj |}t|dr |j St |S7Cw)a Dispatch *prompt* to the injected Opus client. Supports both async (``generate_content_async``) and sync (``generate_content``) interfaces to stay compatible with different SDK versions and test mocks. Raises ------ RuntimeError If no ``opus_client`` was provided at construction time. Exception Any exception propagated from the underlying client call. NuNSemanticMergeInterceptor: no opus_client provided — cannot resolve conflictsgenerate_content_asynctext)r RuntimeErrorhasattrrKgenerate_contentrLstr)r!rDresponses rr2z#SemanticMergeInterceptor._call_opuss    #`  4##%= >!--DDVLLH''88@H 8V $== 8}MsA BBABc ,d||||t|dd} |j}tjj |s tjj d|}tj tjj|dt|dd 5}|jtj|d zd d d y #1swYy xYw#t$r+} tjd |j| Yd } ~ y d } ~ wwxYw) z Append a single merge-event record to ``self.events_path`` as JSONL. The write is best-effort: any I/O error is logged as a warning but never propagated (event logging must never break the merge pipeline). semantic_merge)eventr$r%r&rrz/mnt/e/genesis-systemT)exist_okazutf-8)encoding Nz9SemanticMergeInterceptor: failed to write event to %s: %s)roundr ospathisabsjoinmakedirsdirnameopenwriter3dumpsr9r:r;) r!r$r%r&rrrUr fhrHs rr.z&SemanticMergeInterceptor._record_events&&, " A.   **K77==- ggll+[  KK 4t Dk39 3RE*T12 3 3 3  NNK     s6B C"(C CCCC D(!DD)Nzdata/observability/events.jsonl)r rP))r=r r>dictr?intreturnr )rDrPrhrP) r$rgr%rgr&r rr rrrhNone)rrrrr"rIr2r.rrrrr7s(<''$ fff f  fX<%%% %  %  % %rr)r __future__rr3loggingr[r) dataclassesrtypingrcore.merge.conflict_detectorrcore.merge.merge_prompt_builderr getLoggerrr:r rrrrrqs^$#  !9>   8 $  PPr