Index: flexoentity/domain.py =================================================================== --- flexoentity/domain.py (revision 8aa20c772729e2042bc331b986df06947d18fcdd) +++ flexoentity/domain.py (revision 5c72356eddc16f354a461320a48eaf1e34225257) @@ -1,4 +1,4 @@ from dataclasses import dataclass -from flexoentity.flexo_entity import FlexoEntity, EntityType, EntityState +from flexoentity.flexo_entity import FlexOID, FlexoEntity, EntityType, EntityState @@ -12,9 +12,22 @@ @classmethod def default(cls): - return cls(domain="GEN", entity_type=EntityType.DOMAIN, state=EntityState.DRAFT) + return cls(domain="GEN") def __post_init__(self): super().__post_init__() + # Generate FlexOID only if missing + if not getattr(self, "flexo_id", None): + domain_code = self.domain or "GEN" + self.flexo_id = FlexOID.safe_generate( + domain=domain_code, + entity_type=EntityType.DOMAIN.value, # 'D' + estate=EntityState.DRAFT.value, # 'D' + text=self.text_seed, + version=1, + ) + + + @property def text_seed(self) -> str: Index: flexoentity/flexo_entity.py =================================================================== --- flexoentity/flexo_entity.py (revision 8aa20c772729e2042bc331b986df06947d18fcdd) +++ flexoentity/flexo_entity.py (revision 5c72356eddc16f354a461320a48eaf1e34225257) @@ -1,11 +1,62 @@ """ -versioned_entity.py — Shared base class for all Flex-O entities. -Integrates with hardened id_factory.py and content fingerprint tracking. +flexo_entity.py - I represent the mutable counterpart of the immutable FlexOID. + +I exist to connect an entity’s immutable identity with its mutable provenance +and content. My FlexOID encodes what an entity *is* - its domain, type, +version, and lifecycle state - while I record *who* created it, *who* owns it +now, and *where* it came from. + +Structure +────────── +- My `flexo_id` is the canonical source of truth. + It deterministically encodes: + - domain (prefix) + - entity type (single letter, e.g. I, C, M) + - date + hash (prefix stability) + - version (numeric counter) + - state (single capital letter) + +- My derived properties (`state`, `entity_type`, `version`) read from that ID. + They are never stored separately, so I cannot become inconsistent. + +- My attributes `origin`, `originator_id`, and `owner_id` express authorship + and workflow context: + - `origin` links to the FlexOID of the entity I was derived from + - `originator_id` identifies the original creator (UUID) + - `owner_id` identifies the current responsible user or process (UUID) + +- My optional `domain` field is a semantic hint, useful for filtering or + multi-tenant setups, but the canonical domain remains encoded in my ID. + +Design principles +────────────────── +I separate immutable identity from mutable context: + + ┌────────────────────────────┐ + │ FlexOID │ + │ – who I am │ + │ – lifecycle state │ + └────────────▲───────────────┘ + │ + ┌────────────┴───────────────┐ + │ FlexoEntity (this module) │ + │ – who created or owns me │ + │ – where I came from │ + │ – what content I hold │ + └────────────────────────────┘ + +This means: + - My identity never changes; only my content or ownership does. + - My lifecycle transitions (approve, sign, publish) are represented by new + FlexOIDs, not by mutating state attributes. + - My audit trail and access control live in the provenance layer, not the ID. + +I am the living object behind a FlexOID — mutable, traceable, and accountable. """ + import json -import re from uuid import UUID from enum import Enum -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from typing import Optional from abc import ABC, abstractmethod @@ -67,8 +118,36 @@ @dataclass(kw_only=True) class FlexoEntity(ABC): + """ + I represent a mutable entity identified by an immutable FlexOID. + + My `flexo_id` is the single source of truth for what I am: + - it encodes my domain, entity type, version, and lifecycle state. + I never store those values separately, but derive them on demand. + + I extend the raw identity with provenance and authorship data: + - `origin` → the FlexOID of the entity I was derived from + - `originator_id` → the UUID of the first creator + - `owner_id` → the UUID of the current responsible person or system + - `domain` → optional semantic domain hint (for filtering or indexing) + + I am designed to be traceable and reproducible: + - My identity (`flexo_id`) never mutates. + - My lifecycle transitions are represented by new FlexOIDs. + - My provenance (origin, originator_id, owner_id) can change over time + as the entity moves through workflows or ownership transfers. + + Behavior + ───────── + - `state` and `entity_type` are read-only views derived from my FlexOID. + - `to_dict()` and `from_dict()` serialize me with human-readable fields + while preserving the canonical FlexOID. + - Subclasses (such as questions, media items, or catalogs) extend me with + domain-specific content and validation, but not with new identity rules. + + I am the living, editable layer that connects identity with accountability. + """ + domain: str - entity_type: EntityType subtype: str = "GENERIC" - state: EntityState flexo_id: Optional[FlexOID] = field(default=None) fingerprint: str = field(default_factory=str) @@ -77,4 +156,10 @@ origin: Optional[str] = field(default=None) + def with_new_owner(self, new_owner: UUID): + """Return a clone of this entity with a different owner.""" + copy = replace(self) + copy.owner_id = new_owner + return copy + @property @abstractmethod @@ -83,4 +168,12 @@ raise NotImplementedError("Subclasses must define text_seed property") + @property + def state(self) -> EntityState: + return EntityState(self.flexo_id.state_code) + + @property + def entity_type(self) -> EntityType: + return EntityType(self.flexo_id.entity_type) + def canonical_seed(self) -> str: return canonical_seed(self.text_seed) @@ -98,18 +191,32 @@ def __post_init__(self): """ - Generate ID and content fingerprint. - - All entities must carry a `.domain` attribute exposing a domain code string. - This may be a `Domain` instance or a temporary wrapper used by the `Domain` - class itself to avoid circular initialization. - """ - - self.flexo_id = FlexOID.safe_generate(self.domain_code(), - self.entity_type.value, - self.state.value, - self.text_seed, - 1) - seed = canonical_seed(self.text_seed) - self.fingerprint = hashlib.blake2s(seed.encode("utf-8"), digest_size=8).hexdigest().upper() + Optionally generate a new FlexOID and fingerprint if none exist. + + I check for subclass attributes `ENTITY_TYPE` and `INITIAL_STATE`. + If they exist, I use them to generate a draft FlexOID. + Otherwise, I skip ID generation — leaving it to the subclass. + """ + + # If already has a FlexOID, do nothing + if getattr(self, "flexo_id", None): + return + + # Skip if subclass doesn’t declare ENTITY_TYPE / INITIAL_STATE + etype = getattr(self.__class__, "ENTITY_TYPE", None) + if not etype: + # No info available, do nothing + return + + # Generate a new draft FlexOID + self.flexo_id = FlexOID.safe_generate( + self.domain_code(), + etype.value, + EntityState.DRAFT.value, + self.text_seed, + 1, + ) + + # Compute fingerprint + self.fingerprint = self._compute_fingerprint() def __str__(self): @@ -135,14 +242,10 @@ abbrev, fullname = (lambda p: (p[0], p[1] if len(p) > 1 else ""))(domain.split("_", 1)) domain_obj = Domain( - domain=abbrev, - entity_type=EntityType.DOMAIN, - state=EntityState.DRAFT, # default when reconstructing context + domain=abbrev ) obj = cls( domain=domain_obj, - entity_type=EntityType[data["entity_type"]], - state=EntityState[data["state"]], - ) - obj.flexo_id = FlexOID.from_string(data["flexo_id"]) + ) + obj.flexo_id = FlexOID(data["flexo_id"]) obj.fingerprint = data.get("fingerprint", "") obj.origin = data.get("origin") @@ -207,6 +310,5 @@ # special case: marking obsolete if target_state == EntityState.OBSOLETE: - self.flexo_id = FlexOID.with_state(self.flexo_id, "O") - self.state = target_state + self.flexo_id = self.flexo_id.with_state(EntityState.OBSOLETE.value) return @@ -258,10 +360,19 @@ self.origin = self.flexo_id # optional: keep audit trail self.flexo_id = new_fid - self.state = EntityState.APPROVED return self raise ValueError("Only drafts can be approved") + def sign(self): - self._transition(EntityState.APPROVED_AND_SIGNED) + """ + Mark entity as approved and signed without bumping version. + Only changes the state letter in the FlexOID. + """ + if self.state != EntityState.APPROVED: + raise ValueError("Only approved entities can be signed.") + + # Change only the trailing state letter (A → S) + self.flexo_id = self.flexo_id.with_state(EntityState.APPROVED_AND_SIGNED.value) + return self def publish(self): @@ -279,5 +390,4 @@ ) - new_version = self.flexo_id.version + 1 new_fid = FlexOID.safe_generate( self.domain_code(), @@ -285,10 +395,9 @@ EntityState.PUBLISHED.value, self.text_seed, - version=new_version + version=self.version ) self.origin = self.flexo_id self.flexo_id = new_fid - self.state = EntityState.PUBLISHED return self @@ -302,4 +411,5 @@ if self.state != EntityState.OBSOLETE: self._transition(EntityState.OBSOLETE) + return self def clone_new_base(self): @@ -309,20 +419,86 @@ self.domain_code(), self.entity_type.value, - self.state.value, + EntityState.DRAFT.value, self.text_seed, ) - self.state = EntityState.DRAFT + return self # ─────────────────────────────────────────────────────────────── # Integrity verification # ─────────────────────────────────────────────────────────────── + @staticmethod + def debug_integrity(entity): + """ + Return a dict with all values used by verify_integrity so we can see + exactly why a check passes/fails. + """ + info = {} + try: + fid = entity.flexo_id + info["fid"] = str(fid) + info["fid_domain"] = fid.domain + info["fid_type"] = fid.entity_type + info["fid_state"] = fid.state_code + info["fid_hash_part"] = fid.hash_part + info["version"] = fid.version + + # Derived views + info["derived_domain_code"] = getattr(entity, "domain_code")() if hasattr(entity, "domain_code") else None + info["derived_entity_type_value"] = entity.entity_type.value + info["derived_state_value"] = entity.state.value + + # Seeds & fingerprints + info["text_seed"] = entity.text_seed + exp_fp = entity._compute_fingerprint() + info["expected_fingerprint"] = exp_fp + info["stored_fingerprint"] = getattr(entity, "fingerprint", None) + + # Expected ID hash from current seed (should match fid.hash_part if content didn't change) + exp_hash = FlexOID._blake_hash(canonical_seed(f"{fid.domain}:{fid.entity_type}:{entity.text_seed}")) + info["expected_id_hash_from_seed"] = exp_hash + + # Quick mismatches + info["mismatch_fp"] = (info["stored_fingerprint"] != exp_fp) + info["mismatch_hash"] = (fid.hash_part != exp_hash) + info["mismatch_type"] = (entity.entity_type.value != fid.entity_type) + info["mismatch_state"] = (entity.state.value != fid.state_code) + info["mismatch_domain"] = (info["derived_domain_code"] is not None and info["derived_domain_code"] != fid.domain) + except Exception as e: + info["error"] = repr(e) + return info @staticmethod def verify_integrity(entity) -> bool: - """Verify that an entity’s content fingerprint matches its actual content.""" - expected_fp = hashlib.blake2s( - canonical_seed(entity.text_seed).encode("utf-8"), digest_size=8 - ).hexdigest().upper() - return expected_fp == entity.fingerprint + """ + Verify that an entity's FlexOID, state, and fingerprint are consistent. + + Returns False if: + - flexo_id format is invalid + - derived state/type mismatch the ID + - fingerprint doesn't match the canonical text seed + """ + try: + if not isinstance(entity.flexo_id, FlexOID): + return False + + # Validate domain and ID coherence + if entity.domain and entity.domain_code() != entity.flexo_id.domain: + return False + if entity.entity_type.value != entity.flexo_id.entity_type: + return False + if entity.state.value != entity.flexo_id.state_code: + return False + + # --- Fingerprint validation --- + if hasattr(entity, "fingerprint") and entity.fingerprint: + seed = canonical_seed(entity.text_seed) + expected_fp = entity._compute_fingerprint() + if expected_fp != entity.fingerprint: + return False + + return True + + except Exception as e: + return False def allowed_transitions(self) -> list[str]: Index: tests/conftest.py =================================================================== --- tests/conftest.py (revision 8aa20c772729e2042bc331b986df06947d18fcdd) +++ tests/conftest.py (revision 5c72356eddc16f354a461320a48eaf1e34225257) @@ -4,5 +4,5 @@ from dataclasses import dataclass, field from typing import List -from flexoentity import FlexoEntity, EntityType, EntityState, Domain +from flexoentity import FlexOID, FlexoEntity, EntityType, EntityState, Domain @pytest.fixture @@ -37,14 +37,26 @@ class SingleChoiceQuestion(FlexoEntity): """A minimal stub to test FlexoEntity integration.""" + ENTITY_TYPE = EntityType.ITEM + text: str = "" options: List[AnswerOption] = field(default_factory=list) + def __post_init__(self): + # If no FlexOID yet, generate a draft ID now. + if not getattr(self, "flexo_id", None): + domain_code = ( + self.domain.domain if isinstance(self.domain, Domain) else "GEN" + ) + self.flexo_id = FlexOID.safe_generate( + domain=domain_code, + entity_type=EntityType.ITEM.value, # 'I' + estate=EntityState.DRAFT.value, # 'D' + text=self.text_seed or self.text, + version=1, + ) @classmethod def default(cls): - return cls(domain=Domain(domain="GEN", - entity_type=EntityType.DOMAIN, - state=EntityState.DRAFT), - state=EntityState.DRAFT, entity_type=EntityType.ITEM) + return cls(domain=Domain(domain="GEN")) def to_dict(self): @@ -68,5 +80,5 @@ @classmethod def from_dict(cls, data): - obj = cls( + obj = cls(domain=Domain(domain="GEN"), text=data.get("text", ""), options=[AnswerOption.from_dict(o) for o in data.get("options", [])], @@ -74,8 +86,5 @@ # restore FlexoEntity core fields obj.domain = data.get("domain") - obj.entity_type = EntityType[data.get("etype")] if "etype" in data else EntityType.ITEM - obj.state = EntityState[data.get("state")] if "state" in data else EntityState.DRAFT if "flexo_id" in data: - from flexoentity import FlexOID obj.flexo_id = FlexOID.parsed(data["flexo_id"]) return obj @@ -87,7 +96,7 @@ @pytest.fixture def sample_question(): - return SingleChoiceQuestion(domain=Domain.default(), - text="What is 2 + 2?", - options=[], - entity_type=EntityType.ITEM, - state=EntityState.DRAFT) + q = SingleChoiceQuestion(domain=Domain.default(), + text="What is 2 + 2?", + options=[]) + q._update_fingerprint() + return q Index: tests/test_id_lifecycle.py =================================================================== --- tests/test_id_lifecycle.py (revision 8aa20c772729e2042bc331b986df06947d18fcdd) +++ tests/test_id_lifecycle.py (revision 5c72356eddc16f354a461320a48eaf1e34225257) @@ -20,15 +20,24 @@ assert q.flexo_id.version == 1 - -def test_signing_bumps_version(sample_question): +def test_signing_does_not_bump_version(sample_question): q = sample_question q.approve() - v_before = str(q.flexo_id) + before = q.flexo_id q.sign() + after = q.flexo_id + + # state changed assert q.state == EntityState.APPROVED_AND_SIGNED - assert str(q.flexo_id) != v_before + + # version unchanged + assert before.version == after.version + + # only suffix letter differs + assert before.prefix == after.prefix + assert before.state_code == "A" + assert after.state_code == "S" -def test_publish_bumps_version(sample_question): +def test_publish_does_not_bump_version(sample_question): q = sample_question q.approve() @@ -37,5 +46,5 @@ q.publish() assert q.state == EntityState.PUBLISHED - assert q.flexo_id.version == v_before + 1 + assert q.flexo_id.version == v_before @@ -70,4 +79,5 @@ # simulate tampering q.text = "Tampered text" + print(FlexoEntity.debug_integrity(q)) assert not FlexoEntity.verify_integrity(q) @@ -112,4 +122,6 @@ for _ in range(FlexOID.MAX_VERSION - 1): q.bump_version() + + # Next one must raise with pytest.raises(RuntimeError, match="mark obsolete"): - q.sign() + q.bump_version() Index: tests/test_id_stress.py =================================================================== --- tests/test_id_stress.py (revision 8aa20c772729e2042bc331b986df06947d18fcdd) +++ tests/test_id_stress.py (revision 5c72356eddc16f354a461320a48eaf1e34225257) @@ -82,21 +82,4 @@ -def test_version_ceiling_enforcement(sample_question): - """Simulate approaching @999 to trigger obsolescence guard.""" - q = sample_question - q.approve() - - # artificially bump version number to near ceiling - q.flexo_id = FlexOID.from_oid_and_version(q.flexo_id, 998) - - # 998 → 999 is allowed - q.sign() - assert q.flexo_id.version == 999 - - # 999 → 1000 should raise RuntimeError - with pytest.raises(ValueError): - q.publish() - - def test_massive_lifecycle_simulation(sample_question): """