Update contradiction_resolution.py

codex/contradiction_resolution.py

@@ -1 +1,65 @@
-print("Hello World")
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Literal, Optional
+
+
+Decision = Literal["prefer_a", "prefer_b", "defer", "merge"]
+
+
+@dataclass
+class Contradiction:
+    a: Any
+    b: Any
+    context: str
+    decision: Decision
+    rationale: str
+    timestamp: str
+
+
+def resolve_contradiction(
+    a: Any,
+    b: Any,
+    context: str,
+    policy: Decision = "merge",
+    log_path: Optional[Path] = None,
+) -> Any:
+    """
+    Resolve a contradiction between values `a` and `b`.
+
+    Parameters
+    ----------
+    policy : {"prefer_a","prefer_b","defer","merge"}
+        Simple policy. "merge" tries dict merge; otherwise returns chosen side.
+
+    Returns
+    -------
+    Any
+        Chosen/merged result.
+    """
+    timestamp = datetime.utcnow().isoformat()
+    rationale = "policy=" + policy
+
+    if policy == "prefer_a":
+        result = a
+    elif policy == "prefer_b":
+        result = b
+    elif policy == "defer":
+        result = {"deferred": True, "a": a, "b": b}
+    else:  # merge
+        if isinstance(a, dict) and isinstance(b, dict):
+            result = {**b, **a}  # a overrides b
+            rationale = "merged dicts with a overriding b"
+        else:
+            result = a if a is not None else b
+            rationale = "fallback merge (prefer non-None)"
+
+    record = Contradiction(a, b, context, policy, rationale, timestamp)
+    if log_path:
+        log_path.parent.mkdir(parents=True, exist_ok=True)
+        with log_path.open("a", encoding="utf-8") as fh:
+            fh.write(json.dumps(record.__dict__) + "\n")
+    return result