!i,dZddlmZddlZddlZddlmZddlmZm Z eje Z dZ dZGdd Zy) u7 core/merge/semantic_merge_interceptor_v2.py SwarmMergeInterceptor — sends conflicting SwarmResults to Opus 4.6 for semantic resolution and returns a single merged dict. Note on naming -------------- This is the Track A counterpart to ``SemanticMergeInterceptor`` in ``core/merge/semantic_merge_interceptor.py`` (Track B). Track B operates on RFC 6902 StateDelta patches with ``ConflictDetector``. THIS class operates on ``SwarmResult`` output dicts with ``SwarmConflictDetector``. The class is named ``SwarmMergeInterceptor`` (not ``SemanticMergeInterceptor``) to avoid name collision. Flow ---- 1. ``SwarmConflictDetector.get_conflicts(results)`` 2a. No conflicts → dict-union of all results' outputs (no Opus call) 2b. Conflicts → for each conflicting (result_a, result_b, key), format MERGE_PROMPT and call Opus 4.6 → parse JSON response → insert resolved_value 3. Assemble final dict, attach ``_merge_reasoning`` when conflicts existed. 4. (optional) ``reduce_and_commit()`` writes merged dict to Redis under ``aiva:results:{session_id}`` with a 600-second TTL so AIVA can consume it. Opus is ONLY called when at least one conflict exists. # VERIFICATION_STAMP # Story: 6.02 # Verified By: parallel-builder # Verified At: 2026-02-25 # Tests: 14/14 # Coverage: 100% ) annotationsN)Any)SwarmConflictDetector SwarmResultiXa You are the Genesis Semantic Reducer. Multiple swarm workers have produced conflicting outputs. SESSION: {session_id} WORKER A ({worker_a}): {output_a} WORKER B ({worker_b}): {output_b} CONFLICT ON KEY: {conflict_key} Analyze both outputs and return the most semantically correct resolution. Respond ONLY with valid JSON: {{"resolved_value": "...", "reasoning": "...", "winner": "A|B|MERGE"}} cZeZdZdZdd dZd dZ d dZ d dZd dZy)SwarmMergeInterceptoraj Resolves conflicts between SwarmResult objects using Opus 4.6. Parameters ---------- opus_client: Client used to call Opus 4.6 for conflict resolution. Must expose one of: - ``generate_content_async(prompt) -> response`` (async preferred) - ``generate_content(prompt) -> response`` (sync fallback) Response objects must have a ``.text`` attribute or be str-coercible. Opus is ONLY called when at least one conflict exists. redis_client: Optional Redis client (e.g. ``redis.Redis`` or ``redis.StrictRedis``). When provided, ``reduce_and_commit()`` writes the merged result to ``aiva:results:{session_id}`` via ``setex`` with ``RESULT_TTL`` seconds. When ``None``, ``reduce_and_commit()`` silently skips the Redis write and still returns the merged dict. Nc>||_||_t|_yN)opusredisr _detector)self opus_client redis_clients A/mnt/e/genesis-system/core/merge/semantic_merge_interceptor_v2.py__init__zSwarmMergeInterceptor.__init__ds ! .0c K|jj|}i}|D]}|j|j|s|Sg}|D]\}}}tj |j |jtj|j|jtj|j|} |j| |||d{\} } | ||<|j| ||d<|S7&w)u Merge *results* into a single resolved dict. Algorithm --------- 1. Run SwarmConflictDetector.get_conflicts(results). 2. Compute base merge: union of all output dicts (later results overwrite earlier ones for non-conflicting keys). 3. For each conflict: a. Format MERGE_PROMPT with the two conflicting results and key. b. Call Opus 4.6 for the resolved_value. c. On JSON parse failure → fall back to the value from the result with the higher confidence score. d. Store per-conflict reasoning in ``_merge_reasoning``. 4. Apply resolved values over the base merge. 5. Return final dict. If any conflicts existed, ``_merge_reasoning`` key is present. Parameters ---------- results: List of SwarmResult objects, possibly spanning multiple workers for the same session. Returns ------- dict Merged output dict. Contains ``_merge_reasoning`` key only when at least one conflict was resolved via Opus. ) session_idworker_aoutput_aworker_boutput_b conflict_key)promptresult_aresult_brN_merge_reasoning) r get_conflictsupdateoutput MERGE_PROMPTformatr worker_namejsondumps_resolve_conflictappend) rresults conflictsmergedresult reasoning_logrrrrresolved_valueentrys rreducezSwarmMergeInterceptor.reducems@NN009 "$ )F MM&-- ( )M%' 09 ( ,Hh !((#..!--HOO4!--HOO4) )F+/*@*@!!) +A+% !NE$2F<   '% ((&3!" %sCDD'DcK|j|d{}|j7|jjd|tt j ||S7Iw)a Merge *results* and write the final dict to Redis for AIVA consumption. Algorithm --------- 1. Calls ``reduce(results)`` to obtain the merged dict. 2. If ``redis_client`` is set, serialises the merged dict to JSON and calls ``redis.setex(key, RESULT_TTL, value)`` where:: key = f"aiva:results:{session_id}" RESULT_TTL = 600 (10 minutes) 3. Returns the merged dict (identical to what ``reduce()`` returned). Notes ----- - If ``redis_client`` is ``None`` (not injected), the Redis write is silently skipped and ``reduce()`` still returns its result. - ``setex`` is used (not ``set`` + ``expire``) so the key write and TTL assignment are atomic. Parameters ---------- results: List of SwarmResult objects to merge. session_id: Unique session identifier. The Redis key written is ``aiva:results:{session_id}``. Returns ------- dict The fully merged output dict (same object returned by ``reduce()``). Nz aiva:results:)r0r setex RESULT_TTLr%r&)rr)rr+s rreduce_and_commitz'SwarmMergeInterceptor.reduce_and_commits[N{{7++ :: ! JJ   |, 6"   ,sA#A!A A#cK |j|d{}tj|}|d}|jdd}|jdd} |dk(r|jj||}n!|dk(r|jj||}||| |j |j d } || fS7#t $r} tjd |t| j| |j|jkDr|} n|} | jj|} || j d | d |j |j d } | | fcYd} ~ Sd} ~ wwxYww) a$ Call Opus and parse its JSON response for one conflicting key. Returns ------- (resolved_value, reasoning_entry) On JSON parse failure, falls back to the value from the result with the higher confidence score (result_a wins on tie). Nr.winnerMERGE reasoningAB)keyr6r8rrunSwarmMergeInterceptor: Opus resolution failed for key %r (%s: %s) — falling back to higher-confidence resultzOpus parse failure: z; used higher-confidence result) _call_opusr%loadsgetr!r$ Exceptionloggerwarningtype__name__ confidence)rrrrr opus_textparsedr.r6r8r/excfallback_resultfallback_values rr'z'SwarmMergeInterceptor._resolve_conflictsr". )"oof55IZZ *F#$45NZZ'2F ;3I}!)!4!4\>!R3!)!4!4\>!R$ &$00$00 E"5( (+6. ) NN@S ""  ""X%8%88"*"*,3377 EN#)553C58WX$00$00 E"5( (- )sFE(CB>B$C=E(>C E% BE E%E( E%%E(cKt|jdr$|jj|d{}n|jj|}t|dr |jSt |S7Cw)z Dispatch *prompt* to the injected Opus client. Supports both async (``generate_content_async``) and sync (``generate_content``) client interfaces for compatibility with different SDK versions and test mocks. generate_content_asyncNtext)hasattrr rLgenerate_contentrMstr)rrresponses rr=z SwarmMergeInterceptor._call_opus'sf 4996 7!YY==fEEHyy11&9H 8V $== 8} Fs5A=A;AA=r )rrrrreturnNone)r)list[SwarmResult]rRdict)r)rTrrPrRrU) rrPrrrrrrPrRztuple[Any, dict])rrPrRrP) rD __module__ __qualname____doc__rr0r4r'r=rrrrOsp(1CJ."..  .h?)?)?) ?)  ?)  ?)Brr)rX __future__rr%loggingtypingrcore.merge.swarm_resultrr getLoggerrDrAr3r"rrYrrr_sH"H# F   8 $   $ggr