""" core/coherence/state_delta.py StateDelta — RFC 6902 JSON Patch proposal for multi-agent coherence. This module provides: - VALID_OPS: the 6 RFC 6902 operation names - PatchConflictError: raised when an "op: test" assertion fails - validate_patch(patch): True/False validation of RFC 6902 structure - apply_patch(state, patch): apply a patch list to a state dict, returning a NEW dict - StateDelta: frozen dataclass representing a proposed state change No external libraries (jsonpatch, etc.) are used. All 6 RFC 6902 operations are implemented manually against plain Python dicts. RFC 6902 operations: add — set value at path (creates intermediate keys as needed for "" root; for list indices, appends if index == "-") remove — delete value at path (must exist) replace — set value at path (must already exist) move — remove from "from" path + add at "path" copy — copy value from "from" path to "path" test — assert that value at path equals "value"; raise PatchConflictError if not # VERIFICATION_STAMP # Story: 6.01 # Verified By: parallel-builder # Verified At: 2026-02-25 # Tests: 10/10 # Coverage: 100% """ from __future__ import annotations import copy from dataclasses import dataclass from datetime import datetime from typing import Any, List, Optional # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- VALID_OPS: set = {"add", "remove", "replace", "move", "copy", "test"} # --------------------------------------------------------------------------- # Exceptions # --------------------------------------------------------------------------- class PatchConflictError(Exception): """Raised when an RFC 6902 'test' operation fails.""" def __init__(self, path: str, expected: Any, actual: Any) -> None: self.path = path self.expected = expected self.actual = actual super().__init__( f"Patch conflict at {path}: expected {expected!r}, got {actual!r}" ) # --------------------------------------------------------------------------- # Path utilities # --------------------------------------------------------------------------- def _parse_path(path: str) -> list: """ Convert an RFC 6902 JSON Pointer path to a list of keys/indices. Examples: "" -> [] "/" -> [""] "/a" -> ["a"] "/a/b" -> ["a", "b"] "/a/0/b" -> ["a", "0", "b"] Tilde escaping per RFC 6901: "~1" -> "/" "~0" -> "~" """ if path == "": return [] if not path.startswith("/"): raise ValueError(f"Invalid JSON Pointer path: {path!r} — must start with '/'") tokens = path[1:].split("/") # Decode RFC 6901 escapes: ~1 -> /, ~0 -> ~ (order matters) return [t.replace("~1", "/").replace("~0", "~") for t in tokens] def _get_at(obj: Any, keys: list) -> Any: """ Traverse obj along keys and return the final value. Raises KeyError / IndexError if any key is missing. """ current = obj for key in keys: if isinstance(current, dict): if key not in current: raise KeyError(key) current = current[key] elif isinstance(current, list): idx = _list_index(key, len(current), allow_append=False) current = current[idx] else: raise KeyError(f"Cannot traverse into {type(current).__name__} with key {key!r}") return current def _set_at(obj: Any, keys: list, value: Any) -> None: """ Set obj[keys[-1]] = value, traversing obj along keys[:-1]. Mutates obj in-place (caller must pass a deep copy). For list targets, index "-" appends. Raises KeyError / IndexError / TypeError on bad traversal. """ if not keys: raise ValueError("Cannot set root value via _set_at (use special-case handling)") parent = _get_at(obj, keys[:-1]) final_key = keys[-1] if isinstance(parent, dict): parent[final_key] = value elif isinstance(parent, list): idx = _list_index(final_key, len(parent), allow_append=True) if idx == len(parent): parent.append(value) else: parent[idx] = value else: raise TypeError(f"Cannot set key on {type(parent).__name__}") def _remove_at(obj: Any, keys: list) -> Any: """ Remove and return the value at keys[-1] in obj (traversing keys[:-1]). Mutates obj in-place (caller must pass a deep copy). Raises KeyError / IndexError if path does not exist. """ if not keys: raise ValueError("Cannot remove root value") parent = _get_at(obj, keys[:-1]) final_key = keys[-1] if isinstance(parent, dict): if final_key not in parent: raise KeyError(final_key) return parent.pop(final_key) elif isinstance(parent, list): idx = _list_index(final_key, len(parent), allow_append=False) return parent.pop(idx) else: raise TypeError(f"Cannot remove key from {type(parent).__name__}") def _list_index(key: str, length: int, allow_append: bool) -> int: """ Convert a string key to a list index. "-" means append (index == length) when allow_append=True. Raises IndexError for out-of-range or ValueError for non-integer. """ if key == "-": if allow_append: return length raise IndexError("'-' index not allowed for get/remove operations") try: idx = int(key) except ValueError: raise KeyError(f"Invalid list index: {key!r}") if idx < 0 or idx >= length: raise IndexError(f"List index {idx} out of range (length {length})") return idx # --------------------------------------------------------------------------- # Validation # --------------------------------------------------------------------------- def validate_patch(patch: list) -> bool: """ Validate an RFC 6902 patch (list of operation dicts). Returns True if the patch is structurally valid, False otherwise. Does NOT raise — validation failures return False. Rules: - patch must be a list - every element must be a dict - every dict must have "op" key - "op" value must be in VALID_OPS - "path" key must be present and be a string - "move" and "copy" ops require a "from" key - "add", "replace", "test" ops require a "value" key - "remove" does NOT require "value" """ if not isinstance(patch, (list, tuple)): return False for op_dict in patch: if not isinstance(op_dict, dict): return False if "op" not in op_dict: return False op = op_dict["op"] if op not in VALID_OPS: return False if "path" not in op_dict or not isinstance(op_dict.get("path"), str): return False if op in ("move", "copy"): if "from" not in op_dict or not isinstance(op_dict.get("from"), str): return False if op in ("add", "replace", "test"): if "value" not in op_dict: return False return True # --------------------------------------------------------------------------- # Core apply logic # --------------------------------------------------------------------------- def apply_patch(state: dict, patch: list) -> dict: """ Apply an RFC 6902 patch to state dict. Returns a NEW dict (original unchanged). Supported operations: add, remove, replace, move, copy, test Raises: PatchConflictError: if "test" op fails (value mismatch) KeyError: if path does not exist for remove/replace/test ValueError: if patch is structurally invalid TypeError: if an operation targets a non-container """ # Work on a deep copy so original is never mutated result = copy.deepcopy(state) for op_dict in patch: op = op_dict.get("op") if op not in VALID_OPS: raise ValueError(f"Unknown op: {op!r}") path = op_dict.get("path", "") keys = _parse_path(path) if op == "test": expected = op_dict["value"] try: actual = _get_at(result, keys) except (KeyError, IndexError): actual = _MISSING if actual != expected: raise PatchConflictError(path=path, expected=expected, actual=actual) elif op == "add": value = copy.deepcopy(op_dict["value"]) if not keys: # Adding to root replaces entire state result = value else: _set_at(result, keys, value) elif op == "remove": if not keys: raise ValueError("Cannot 'remove' the root document") _remove_at(result, keys) elif op == "replace": if not keys: # Replace root result = copy.deepcopy(op_dict["value"]) else: # Must already exist _get_at(result, keys) # raises KeyError if missing _set_at(result, keys, copy.deepcopy(op_dict["value"])) elif op == "move": from_path = op_dict["from"] from_keys = _parse_path(from_path) moved_value = _remove_at(result, from_keys) if not keys: result = moved_value else: _set_at(result, keys, moved_value) elif op == "copy": from_path = op_dict["from"] from_keys = _parse_path(from_path) copied_value = copy.deepcopy(_get_at(result, from_keys)) if not keys: result = copied_value else: _set_at(result, keys, copied_value) return result # Sentinel for "value not found" in test op class _MissingType: def __repr__(self): return "" _MISSING = _MissingType() # --------------------------------------------------------------------------- # StateDelta dataclass # --------------------------------------------------------------------------- @dataclass(frozen=True) class StateDelta: """ Immutable proposal for a state change, expressed as an RFC 6902 JSON Patch. Attributes: agent_id: ID of the agent proposing this change session_id: Session this change belongs to version_at_read: The state version this patch was built against patch: Tuple of RFC 6902 operation dicts (frozen for hashability) submitted_at: When this delta was submitted The patch field uses a tuple (not list) so the frozen dataclass remains hashable. Convert to list before calling apply_patch if needed — or use the convenience method apply_to(). """ agent_id: str session_id: str version_at_read: int patch: tuple # tuple of dicts — frozen for hashability submitted_at: datetime def apply_to(self, state: dict) -> dict: """ Apply this delta's patches to state dict. Returns a NEW dict. The original state dict is never mutated. Raises: PatchConflictError: if a 'test' operation fails KeyError: if a required path does not exist ValueError: if the patch is malformed """ return apply_patch(state, list(self.patch))