conflict.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/round/conflict.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_round_close_escalation.py"
"""
Merge-conflict resolution policy for round-close.
 
Purpose
-------
Decide, for a set of conflicted paths from a staging merge, whether the
round-close job may auto-resolve (generated views only) or must abort
(OI-210 abort-by-default).
 
Responsibilities
----------------
- Hold the tight allowlist of auto-resolvable generated-file prefixes.
- Classify a conflict set into resolvable vs blocking paths.
- Default to abort whenever any path is outside the allowlist.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - decision events
    L3 - per-path classification
    L4 - deep tracing
 
Contracts
---------
- An empty conflict set is never "resolvable"; callers treat it as clean.
- The allowlist is prefix-based and tight (views only) by default.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- dataclasses (the frozen ConflictDecision and ConflictPolicy records)
- enum (the ConflictStrategy enum)
"""
from __future__ import annotations
 
from dataclasses import dataclass, field
from enum import Enum
from typing import Sequence
 
from oc_diagnostics import _dbg
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
# Q-1 lean: keep the allowlist tight — generated registry views only.
DEFAULT_ALLOWLIST_PREFIXES: tuple[str, ...] = ("oc_chat/registry/views/",)
 
 
class ConflictStrategy(str, Enum):
    """How the round-close job should treat a conflict set."""
 
    ABORT = "abort"
    TAKE_CURRENT = "take_current"
 
 
@dataclass(frozen=True)
class ConflictDecision:
    """The classification result for one conflict set."""
 
    strategy: ConflictStrategy
    resolvable_paths: tuple[str, ...] = field(default_factory=tuple)
    blocking_paths: tuple[str, ...] = field(default_factory=tuple)
 
    @property
    def is_clean(self) -> bool:
        """True when there were no conflicts to resolve."""
 
        return not self.resolvable_paths and not self.blocking_paths
 
 
@dataclass(frozen=True)
class ConflictPolicy:
    """Prefix-allowlisted, abort-by-default conflict policy."""
 
    allowlist_prefixes: tuple[str, ...] = DEFAULT_ALLOWLIST_PREFIXES
 
    def classify(self, conflicted_paths: Sequence[str]) -> ConflictDecision:
        """Classify ``conflicted_paths`` into a resolution decision."""
 
        func = "classify"
        paths = tuple(p.strip() for p in conflicted_paths if p.strip())
        if not paths:
            _dbg("OC_CHAT.MAIN", func, "no conflicts", 4)
            return ConflictDecision(strategy=ConflictStrategy.TAKE_CURRENT)
        resolvable = tuple(p for p in paths if self._allowlisted(p))
        blocking = tuple(p for p in paths if not self._allowlisted(p))
        if blocking:
            _dbg("OC_CHAT.MAIN", func, f"abort: {len(blocking)} blocking", 2)
            return ConflictDecision(
                strategy=ConflictStrategy.ABORT,
                resolvable_paths=resolvable,
                blocking_paths=blocking,
            )
        _dbg("OC_CHAT.MAIN", func, f"take_current: {len(resolvable)} views", 2)
        return ConflictDecision(
            strategy=ConflictStrategy.TAKE_CURRENT,
            resolvable_paths=resolvable,
        )
 
    def _allowlisted(self, path: str) -> bool:
        normalized = path.replace("\\", "/")
        return normalized.startswith(self.allowlist_prefixes)