Skip to content

symfonic.capabilities.memory.phases.merge_classification

merge_classification

Classification identity preservation when duplicate memories are merged.

carry_classification

carry_classification(canonical: Any, absorbed: Any, props: dict[str, Any]) -> bool

Preserve one classification identity; refuse two that disagree.

Source code in src/symfonic/capabilities/memory/phases/merge_classification.py
def carry_classification(canonical: Any, absorbed: Any, props: dict[str, Any]) -> bool:
    """Preserve one classification identity; refuse two that disagree."""
    left, left_error = classification_identity(
        dict(getattr(canonical, "properties", None) or {}),
        layer=getattr(canonical, "layer", None),
    )
    right, right_error = classification_identity(
        dict(getattr(absorbed, "properties", None) or {}),
        layer=getattr(absorbed, "layer", None),
    )
    if left_error or right_error or (left is not None and right is not None and left != right):
        return False
    if left is None and right is not None:
        existing = props.get(METADATA_KEY)
        producer = dict(existing) if isinstance(existing, Mapping) else {}
        producer.update(memory_category=right[0], subject=right[1])
        props[METADATA_KEY] = producer
    return True