CAP Triangle:

        C (Consistency)
       / \
      /   \
     /     \
    / Choice \
   /  Region  \
  A --------- P
(Availability) (Partition Tolerance)

coomia-dip Choice: AP + Eventual Consistency
  - Guarantee availability and partition tolerance
  - Achieve eventual consistency through event-driven design
  - Provide strong consistency options on critical paths

Consistency Requirements Matrix:

Scenario                    Consistency     Reason
------------------------------------------------------
Ontology Schema changes     Strong          Schema affects everything
World create/delete         Strong          World is isolation boundary
User auth/authorization     Strong          Security has no grace period
Object instance CRUD        Eventual        Brief inconsistency OK
Derived property recalc     Eventual        Briefly stale values OK
Rule evaluation trigger     Eventual        ms-level delay OK
Inter-World branch data     Snapshot        Branches fully isolated
Audit logs                  Eventual        Just don't lose data

Data Ownership:

onto-control (Control Layer):
  +-- Ontology Schema (ObjectType, LinkType, ActionType)
  +-- World metadata (World, Branch, Tag)
  +-- Users and permissions (User, Role, Permission)
  +-- Data source: PostgreSQL (strong consistency)

onto-data (Data Layer):
  +-- Object instances
  +-- Link instances
  +-- Time series data
  +-- Data source: Doris + Iceberg/Nessie (eventual consistency)

onto-intelligence (Reasoning & Decision Layer + Agent Runtime Layer):
  +-- Rule definitions and evaluation results
  +-- Decision definitions and execution results
  +-- Derived property definitions and computed values
  +-- Data source: PostgreSQL + Redis (mixed)

Cross-process consistency = Event-driven + Compensation

Nessie Version Tree:

main (World: prod)
  |
  +-- commit-001: Initialize ObjectTypes
  +-- commit-002: Add InventoryRecord
  +-- commit-003: Inventory data sync
  |
  +---- branch: staging (World: staging)
  |     +-- commit-004: Add SupplierScore property
  |     +-- commit-005: Modify scoring logic
  |
  +---- branch: dev-feature-x (World: dev)
        +-- commit-006: Experimental ObjectType
        +-- commit-007: Test data

class NessieOCCWriter:
    """Concurrent-safe writes using Nessie OCC"""

    def __init__(self, nessie_client):
        self.nessie = nessie_client

    async def write_with_occ(
        self,
        branch: str,
        table_id: str,
        data: Any,
        max_retries: int = 3,
    ):
        """Optimistic concurrent write"""
        for attempt in range(max_retries):
            try:
                # 1. Read current branch HEAD hash
                branch_ref = await self.nessie.get_reference(branch)
                current_hash = branch_ref.hash

                # 2. Execute write operation
                operations = self._prepare_operations(table_id, data)

                # 3. Commit with expected_hash
                # If hash changed (someone else committed), commit fails
                await self.nessie.commit(
                    branch=branch,
                    operations=operations,
                    expected_hash=current_hash,  # OCC key
                    message=f"Update {table_id}",
                )
                return  # Success

            except NessieConflictException:
                # 4. Conflict: another writer committed before us
                if attempt < max_retries - 1:
                    await asyncio.sleep(0.1 * (2 ** attempt))
                    continue
                else:
                    raise ConcurrencyConflictError(
                        f"Failed after {max_retries} retries "
                        f"on branch {branch}"
                    )

Why OCC over pessimistic locking:

Scenario                OCC              Pessimistic Lock
------------------------------------------------------------
Low contention (normal) Very fast, no     Must acquire/release
                        lock waiting      locks
High contention (rare)  Retry overhead    Blocking waits
Deadlock risk           None              Present
Distributed complexity  Low               High (distributed lock)
Throughput              High              Limited by lock granularity

coomia-dip scenario analysis:
  - Different World writes are fully isolated -> no contention
  - Same World concurrent writes -> low contention (different objects)
  - Same object concurrent writes -> extremely rare

Conclusion: OCC is the optimal choice

Kafka Ordering Guarantees:

Guarantee 1: Messages within a Partition are strictly ordered
  Partition 0: [msg-1] -> [msg-2] -> [msg-3]  <- strictly ordered

Guarantee 2: Same Key routes to same Partition
  Key = "WH-001::PROD-ABC"
  -> hash("WH-001::PROD-ABC") % num_partitions = Partition 3
  -> All change events for this entity go to Partition 3

Guarantee 3: Consumer reads Partition in order
  Consumer reads Partition 3: [evt-1] -> [evt-2] -> [evt-3]

Partition Key Strategy (varies by Topic):

Topic                   Partition Key             Reason
------------------------------------------------------------
cdc.raw.*               source_pk                 Same entity ordered
onto.changes.objects    world_id + object_pk      Same world+object ordered
onto.changes.links      world_id + link_pk        Same world+link ordered
domain.alerts           world_id + alert_type     Same alert type ordered
domain.decisions        world_id + decision_id    Same decision ordered
domain.actions          world_id + action_id      Same action ordered

Cross-Entity Causal Ordering Problem:

Scenario: Object A change triggers Object B change
  - A's change in Partition 1
  - B's change in Partition 5
  - Consumer X processes Partition 5 first (B's change)
  - Consumer Y processes Partition 1 later (A's change)
  - Result: B processed before A!

Solution: Events carry causal chain information

{
  "event_id": "evt-002",
  "caused_by": "evt-001",     // Causal relationship marker
  "causal_order": 2,          // Causal sequence number
  "world_id": "world-prod"
}

Processing logic:
  if event.caused_by is not None:
    if not is_processed(event.caused_by):
      defer(event, retry_after=100ms)
    else:
      process(event)

class IdempotentConsumer:
    """Idempotent consumer"""

    def __init__(self, processed_store):
        self.processed = processed_store  # Redis Set

    async def consume(self, event: Event):
        # 1. Check if already processed
        event_key = f"processed:{event.event_id}"
        if await self.processed.exists(event_key):
            logger.info(f"Skipping duplicate: {event.event_id}")
            return

        # 2. Process event
        await self._process(event)

        # 3. Mark as processed (TTL = 7 days)
        await self.processed.set(event_key, "1", ex=7 * 24 * 3600)

    async def _process(self, event: Event):
        """Actual processing logic (must be idempotent)"""
        # Use upsert instead of insert
        # Use conditional update instead of blind update
        await self.store.upsert(
            key=event.object_pk,
            value=event.after_properties,
            version=event.version,  # Version check
        )

World Isolation Architecture:

World: prod (main branch)
  +----------------------------------------+
  | Nessie Branch: main                    |
  | Doris Database: world_prod             |
  | Kafka Consumer Group: cg-world-prod    |
  |                                        |
  | ObjectTypes: [complete prod Schema]    |
  | Objects: [production data]             |
  | Rules: [production rules]              |
  | Decisions: [production decisions]      |
  +----------------------------------------+

World: staging (staging branch)
  +----------------------------------------+
  | Nessie Branch: staging                 |
  | Doris Database: world_staging          |
  | Kafka Consumer Group: cg-world-staging |
  |                                        |
  | ObjectTypes: [staging Schema]          |
  | Objects: [staging data]                |
  | Rules: [rules under test]             |
  | Decisions: [decisions under test]      |
  +----------------------------------------+

Isolation Guarantees:
  - Different World data is completely isolated
  - Modifying staging does not affect prod
  - Only Merge operations propagate changes between Worlds

Saga Pattern (creating an object instance):

Step 1: onto-control -- Validate ObjectType exists
  Success -> Step 2
  Failure -> Return error

Step 2: onto-data -- Write to Doris
  Success -> Step 3
  Failure -> No compensation needed (Step 1 has no side effects)

Step 3: onto-data -- Write to Iceberg
  Success -> Step 4
  Failure -> Compensate Step 2 (delete from Doris)

Step 4: onto-data -- Publish change event to Kafka
  Success -> Complete
  Failure -> Compensate Step 3 (delete from Iceberg)
             Compensate Step 2 (delete from Doris)

class ObjectCreateSaga:
    """Object creation Saga"""

    async def execute(self, request: CreateObjectRequest):
        compensations = []

        try:
            # Step 1: Validate Schema
            await self.control_client.validate_object_type(
                request.world_id, request.object_type
            )

            # Step 2: Write to Doris
            doris_result = await self.doris_writer.insert(
                request.world_id, request.object_type,
                request.properties
            )
            compensations.append(
                lambda: self.doris_writer.delete(
                    request.world_id, doris_result.pk
                )
            )

            # Step 3: Write to Iceberg
            iceberg_result = await self.iceberg_writer.append(
                request.world_id, request.object_type,
                request.properties
            )
            compensations.append(
                lambda: self.iceberg_writer.delete(
                    request.world_id, iceberg_result.file_id
                )
            )

            # Step 4: Publish event
            await self.event_publisher.publish(
                "onto.changes.objects",
                ObjectChangeEvent(
                    world_id=request.world_id,
                    object_type=request.object_type,
                    change_type="CREATE",
                    after_properties=request.properties,
                )
            )
            return doris_result

        except Exception as e:
            # Execute compensations in reverse order
            for compensation in reversed(compensations):
                try:
                    await compensation()
                except Exception as comp_error:
                    logger.error(f"Compensation failed: {comp_error}")
            raise

Three-Way Merge:

Find the common ancestor (Merge Base), compare three versions:

  Merge Base (fork point)
       |
  +----+----+
  |         |
  v         v
Source    Target
(staging) (prod)

For each object/property:
  Base    Source  Target  -> Operation
  -----------------------------------------------
  A=1     A=1    A=1     -> No change (all same)
  A=1     A=2    A=1     -> Take Source (Source modified)
  A=1     A=1    A=3     -> Take Target (Target modified)
  A=1     A=2    A=3     -> Conflict! Needs resolution
  -       A=2    -       -> Take Source (Source added)
  A=1     -      A=1     -> Delete (Source deleted)
  A=1     A=2    -       -> Conflict! One modified, one deleted

class WorldMergeService:
    """World three-way merge service"""

    async def merge(
        self,
        source_world: str,
        target_world: str,
        merge_strategy: str = "AUTO",
    ) -> MergeResult:

        # 1. Find Merge Base
        base_ref = await self.nessie.find_merge_base(
            source_world, target_world
        )

        # 2. Get three version snapshots
        base_snapshot = await self._get_snapshot(base_ref)
        source_snapshot = await self._get_snapshot(source_world)
        target_snapshot = await self._get_snapshot(target_world)

        # 3. Compare object by object
        conflicts = []
        operations = []

        all_keys = set(
            list(base_snapshot.keys()) +
            list(source_snapshot.keys()) +
            list(target_snapshot.keys())
        )

        for key in all_keys:
            base_val = base_snapshot.get(key)
            source_val = source_snapshot.get(key)
            target_val = target_snapshot.get(key)

            result = self._three_way_compare(
                base_val, source_val, target_val
            )

            if result.is_conflict:
                conflicts.append(Conflict(
                    key=key, base=base_val,
                    source=source_val, target=target_val,
                    conflict_type=result.conflict_type,
                ))
            elif result.has_change:
                operations.append(result.operation)

        # 4. Handle conflicts
        if conflicts:
            if merge_strategy == "AUTO":
                for conflict in conflicts:
                    operations.append(MergeOperation(
                        key=conflict.key,
                        value=conflict.source,
                        resolution="SOURCE_WINS",
                    ))
            elif merge_strategy == "MANUAL":
                return MergeResult(
                    status="CONFLICTS",
                    conflicts=conflicts,
                    merge_id=generate_merge_id(),
                )

        # 5. Apply merge operations to Target
        await self._apply_operations(target_world, operations)

        return MergeResult(
            status="MERGED",
            operations_applied=len(operations),
            conflicts_resolved=len(conflicts),
        )

Conflict Resolution Strategies:

Strategy 1: SOURCE_WINS
  - Source value overwrites Target
  - Use case: staging -> prod deployment

Strategy 2: TARGET_WINS
  - Target value preserved
  - Use case: prod data must not be overwritten

Strategy 3: MANUAL
  - Return conflict list
  - User resolves each in the UI
  - Use case: important merge operations

Strategy 4: TIMESTAMP_WINS
  - Take the more recently updated value
  - Use case: data synchronization scenarios

Conflict Types:
  +------------------------------------------+
  | MODIFY_MODIFY: Both sides modified same  |
  | MODIFY_DELETE: One modified, one deleted  |
  | SCHEMA_CONFLICT: ObjectType def conflict  |
  | RULE_CONFLICT: Rule definition conflict   |
  +------------------------------------------+

Cross-process consistency (no distributed transactions):

Scenario: Modifying an ObjectType property

1. onto-control updates Schema (PostgreSQL transaction)
   +-- Publishes event: schema.updated

2. onto-data receives event, updates storage layer mapping
   +-- Doris schema change
   +-- Iceberg Schema Evolution
   +-- Publishes event: storage.schema.updated

3. onto-intelligence receives event, updates derived property deps
   +-- Updates DAG graph
   +-- Triggers affected derived property recalculation

Consistency Window:
  Step 1 complete -> Step 2 complete: ~200ms
  Step 2 complete -> Step 3 complete: ~500ms

During this window:
  - New Schema is defined, but storage is still updating
  - Queries may return data with old Schema
  - This is acceptable eventual consistency

class VersionVector:
    """Version vector - detect cross-process consistency"""

    def __init__(self):
        self.versions = {
            "control": 0,
            "data": 0,
            "intelligence": 0,
        }

    def increment(self, process: str):
        self.versions[process] += 1

    def is_consistent(self) -> bool:
        """Check if all three processes are consistent"""
        values = list(self.versions.values())
        return max(values) - min(values) <= 1

    def get_lagging_process(self) -> str | None:
        """Find the lagging process"""
        max_ver = max(self.versions.values())
        for process, ver in self.versions.items():
            if ver < max_ver:
                return process
        return None

Periodic Consistency Checks (every 5 minutes):

1. Schema consistency check:
   control_schema = onto-control.get_schema(world_id)
   data_schema = onto-data.get_schema(world_id)
   if control_schema.version != data_schema.version:
     trigger_schema_sync(world_id)

2. Object count consistency:
   control_count = onto-control.get_object_count(world_id, type)
   data_count = onto-data.get_object_count(world_id, type)
   if abs(control_count - data_count) > threshold:
     trigger_reconciliation(world_id, type)

3. Derived property freshness:
   for derived_prop in all_derived_properties:
     last_calc = get_last_calculation_time(derived_prop)
     if now() - last_calc > max_staleness:
       trigger_recalculation(derived_prop)

Consistency Selection Matrix:

                    Read Latency Requirement
                    Low(< 10ms)  Med(< 100ms)  High(< 1s)
Write Freq  -----------------------------------------------
  Low       | Strong        Strong         Eventual
(< 10/s)    | (query primary) (query primary) (event-driven)
            |
  Med       | Eventual      Eventual        Eventual
(< 100/s)   | (cache+inval) (event-driven)  (event-driven)
            |
  High      | Eventual      Eventual        Eventual
(> 100/s)   | (CQRS)       (CQRS)          (batch)

coomia-dip scenarios:
  Ontology Schema:   Low write x Low latency -> Strong
  Object queries:    Med write x Low latency -> Eventual (cache)
  Derived props:     Med write x Med latency -> Eventual (event)
  Audit logs:        High write x High latency -> Eventual (batch)

CQRS (Command Query Responsibility Segregation):

Write Path (Command):
  Client -> Gateway -> onto-data -> Doris (write)
                                      |
                                 Kafka Event
                                      |
                                 onto-data -> update read model

Read Path (Query):
  Client -> Gateway -> onto-data -> Doris (read)
                                      ^
                                 Read model (materialized views)

Benefits:
  - Write and read paths can be optimized independently
  - Read path can use materialized views
  - Writes don't block reads

Scenarios where Actions may be executed multiple times:

1. Temporal Worker crashes and restarts -> Activity retry
2. Network timeout -> Client unsure if succeeded -> Retry
3. Kafka duplicate message consumption -> Duplicate Action trigger
4. Manual replay -> Ops team re-triggers

If Actions are not idempotent:
  Create purchase order x 3 = 3 duplicate orders = financial disaster

class IdempotentActionExecutor:
    """Idempotent action executor"""

    def __init__(self, idempotency_store):
        self.store = idempotency_store  # Redis

    async def execute(
        self,
        action_id: str,
        idempotency_key: str,
        action_fn,
        params: dict,
    ):
        # 1. Check for existing result
        existing = await self.store.get(idempotency_key)
        if existing:
            logger.info(
                f"Action {action_id} already executed, "
                f"returning cached result"
            )
            return existing

        # 2. Acquire execution lock
        lock = await self.store.acquire_lock(
            f"lock:{idempotency_key}", timeout=60)

        if not lock:
            return await self._wait_for_result(idempotency_key)

        try:
            # 3. Execute Action
            result = await action_fn(**params)

            # 4. Store result (TTL = 7 days)
            await self.store.set(
                idempotency_key, result, ex=7 * 24 * 3600)

            return result
        finally:
            await self.store.release_lock(lock)

    def generate_idempotency_key(
        self, action_type: str, params: dict
    ) -> str:
        """Generate idempotency key"""
        key_parts = [
            action_type,
            params.get("world_id", ""),
            params.get("object_type", ""),
            params.get("object_pk", ""),
        ]
        content = "|".join(key_parts)
        return hashlib.sha256(content.encode()).hexdigest()

Consistency Monitoring Dashboard:

Metric 1: Event Processing Lag
  kafka_consumer_lag{topic="onto.changes.*"}
  Alert: lag > 1000 -> WARNING
  Alert: lag > 10000 -> CRITICAL

Metric 2: Cross-Process Version Difference
  version_vector_diff{process_pair="control-data"}
  Alert: diff > 5 -> WARNING

Metric 3: Derived Property Freshness
  derived_property_staleness_seconds
  Alert: staleness > 60s -> WARNING

Metric 4: Merge Conflict Rate
  world_merge_conflict_rate
  Alert: rate > 10% -> Review workflow needed

Metric 5: Idempotency Cache Hit Rate
  idempotency_cache_hit_rate
  Normal: < 1% (occasional retries)
  Alert: > 10% (system may have issues)