""" 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. """ from __future__ import annotations import json import logging import os import time from dataclasses import dataclass from typing import Optional from core.merge.conflict_detector import ConflictDetector from core.merge.merge_prompt_builder import MergePromptBuilder logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Result dataclass # --------------------------------------------------------------------------- @dataclass class MergeResult: """Outcome of a SemanticMergeInterceptor.merge() call.""" success: bool merged_patch: list used_opus: bool resolution_rationale: Optional[str] = None latency_ms: float = 0.0 # --------------------------------------------------------------------------- # SemanticMergeInterceptor # --------------------------------------------------------------------------- class SemanticMergeInterceptor: """ 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"``. """ def __init__( self, opus_client=None, events_path: str = "data/observability/events.jsonl", ): self.detector = ConflictDetector() self.prompt_builder = MergePromptBuilder() self.opus_client = opus_client self.events_path = events_path # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ async def merge( self, deltas: list, current_state: dict, version: int = 1, ) -> MergeResult: """ 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. """ start = time.monotonic() # Step 1 — conflict detection (always runs, never skipped) report = self.detector.detect(deltas) if not report.has_conflicts: # FAST PATH — no conflicts, skip Opus entirely merged = self.detector.fast_merge(deltas) latency = (time.monotonic() - start) * 1000 self._record_event( delta_count=len(deltas), conflict_count=0, resolved=True, used_opus=False, latency_ms=latency, ) return MergeResult( success=True, merged_patch=merged, used_opus=False, latency_ms=latency, ) # SLOW PATH — conflicts detected, invoke Opus for arbitration conflict_count = len(report.conflicting_paths) try: prompt = self.prompt_builder.build( deltas, report, current_state, version ) opus_response_text = await self._call_opus(prompt) parsed = json.loads(opus_response_text) resolved_patch = parsed["resolved_patch"] if not isinstance(resolved_patch, list): raise ValueError( f"resolved_patch must be a list, got {type(resolved_patch).__name__}" ) rationale: Optional[str] = parsed.get("resolution_rationale", "") latency = (time.monotonic() - start) * 1000 self._record_event( delta_count=len(deltas), conflict_count=conflict_count, resolved=True, used_opus=True, latency_ms=latency, ) return MergeResult( success=True, merged_patch=resolved_patch, used_opus=True, resolution_rationale=rationale, latency_ms=latency, ) except Exception as exc: # Opus unavailable or returned malformed JSON — partial merge fallback logger.warning( "Opus merge failed (%s: %s) — falling back to partial merge of " "%d non-conflicting delta(s)", type(exc).__name__, exc, len(report.non_conflicting_deltas), ) merged = self.detector.fast_merge(report.non_conflicting_deltas) latency = (time.monotonic() - start) * 1000 self._record_event( delta_count=len(deltas), conflict_count=conflict_count, resolved=False, used_opus=False, latency_ms=latency, ) return MergeResult( success=True, merged_patch=merged, used_opus=False, latency_ms=latency, ) # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ async def _call_opus(self, prompt: str) -> str: """ 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. """ if self.opus_client is None: raise RuntimeError( "SemanticMergeInterceptor: no opus_client provided — cannot resolve conflicts" ) if hasattr(self.opus_client, "generate_content_async"): response = await self.opus_client.generate_content_async(prompt) else: # Synchronous fallback (e.g. mocks that aren't coroutine-based) response = self.opus_client.generate_content(prompt) if hasattr(response, "text"): return response.text return str(response) def _record_event( self, delta_count: int, conflict_count: int, resolved: bool, used_opus: bool, latency_ms: float, ) -> None: """ 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). """ event = { "event": "semantic_merge", "delta_count": delta_count, "conflict_count": conflict_count, "resolved": resolved, "used_opus": used_opus, "latency_ms": round(latency_ms, 3), } try: events_path = self.events_path # Support both absolute and relative paths if not os.path.isabs(events_path): events_path = os.path.join( "/mnt/e/genesis-system", events_path ) os.makedirs(os.path.dirname(events_path), exist_ok=True) with open(events_path, "a", encoding="utf-8") as fh: fh.write(json.dumps(event) + "\n") except Exception as exc: # pragma: no cover logger.warning( "SemanticMergeInterceptor: failed to write event to %s: %s", self.events_path, exc, ) # VERIFICATION_STAMP # Story: 7.02 # Verified By: parallel-builder # Verified At: 2026-02-25 # Tests: 13/13 # Coverage: 100%