orchestrata_ultimate.cxp.deterministic_ids

Reference › orchestrata_ultimate › cxp › deterministic_ids

generate_deterministic_id returns a UUID v5 derived from a packet's content, so the same packet content always yields the same id. The key is the sender, recipient, subject, thread id and ISO timestamp joined with | and hashed in a fixed CXP namespace. The id is therefore reproducible from the packet's fields. The Postman sync router (postman.sync_router) is its importer. See cxp.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/cxp/deterministic_ids.py

"""Deterministic CXP packet ID generation using UUID v5.

Generates reproducible message_id values from packet content fields,
so that the same logical packet always produces the same UUID —
regardless of when or how many times it is created.

This solves the UUID mismatch between:
  - Runtime auditor (random uuid4, stored in the graph)
  - JSONL reconstruction (hash-based, different from runtime)

By using UUID v5 (SHA-1, RFC 4122) with a fixed namespace, both
paths produce identical IDs for identical packet content.

Scope: spike-only. The production CXPPacket class retains uuid4
for concurrent routing where packets may share identical fields.
"""

import uuid
from datetime import datetime

from orchestrata_ultimate.core.base.traces import traces

# Fixed namespace for CXP packet IDs (RFC 4122 UUID v5 namespace)
CXP_NAMESPACE = uuid.UUID("a1b2c3d4-e5f6-7890-abcd-ef1234567890")


@traces(
    spec="004",
    story="US-001",
    reqs=[
        "REQ-004-US-001-002",
    ],
)
def generate_deterministic_id(
    from_agent: str,
    to_agent: str,
    subject: str,
    thread_id: str,
    timestamp: datetime,
) -> str:
    """Generate a deterministic UUID v5 from packet content fields.

    The canonical key is: from_agent|to_agent|subject|thread_id|timestamp_iso

    Args:
        from_agent: Source agent identifier (e.g. "U:Ox").
        to_agent: Destination agent identifier (e.g. "U:Cd").
        subject: Packet subject/type.
        thread_id: Conversation thread UUID.
        timestamp: Packet creation timestamp.

    Returns:
        UUID v5 string derived from the content fields.
    """
    if hasattr(timestamp, "isoformat"):
        ts_str = timestamp.isoformat()
    else:
        ts_str = str(timestamp)
    key = f"{from_agent}|{to_agent}|{subject}|{thread_id}|{ts_str}"
    return str(uuid.uuid5(CXP_NAMESPACE, key))