from __future__ import annotations import secrets import time import uuid from typing import Any, MutableMapping def new_governance_uid() -> str: """Return an RFC 9562 UUIDv7-compatible identifier.""" timestamp_ms = int(time.time_ns() // 1_000_000) & ((1 << 48) - 1) random_a = secrets.randbits(12) random_b = secrets.randbits(62) value = ( (timestamp_ms << 80) | (0x7 << 76) | (random_a << 64) | (0b10 << 62) | random_b ) return str(uuid.UUID(int=value)) def ensure_governance_uid(properties: MutableMapping[str, Any]) -> str: """Attach a UUIDv7 to mutable node properties and preserve it on retries.""" existing = properties.get("uid") if existing: try: parsed = uuid.UUID(str(existing)) except (ValueError, AttributeError, TypeError) as exc: raise ValueError("governance uid must be a valid UUIDv7") from exc if parsed.version != 7 or parsed.variant != uuid.RFC_4122: raise ValueError("governance uid must be a valid UUIDv7") value = str(parsed) else: value = new_governance_uid() properties["uid"] = value return value