The CZOA/CZOI User Guide

1. Introduction

What is CZOI?

The CZOI toolkit is the reference Python implementation of the Constrained Zoned-Object Architecture (CZOA) — a unified framework for building systems that are simultaneously:

• Secure — enforce strict, auditable access policies.
• Intelligent — adapt to changing conditions through neural components.
• Organisation-aligned — mirror the hierarchical structure of real enterprises.
• Integrable — compose heterogeneous subsystems into a coherent federation without sacrificing local autonomy.

The core idea is simple but powerful: every organizational unit is a zone, and every zone is itself a full system. A hospital is a zone. So is its Emergency department. So is a legacy mainframe wrapped in an adapter zone.

Three fragmentations, one solution

1. Intelligence vs. security. AI models and access-control systems have evolved separately, with different abstractions, correctness criteria, and timescales.
2. Local autonomy vs. global governance. Subunits want to evolve at their own rate; governance wants a single policy.
3. Heterogeneous systems vs. coherent operations. Real organizations run on dozens of incompatible systems.

CZOA addresses all three with a single mechanism: recursive composition.

Who is this guide for?

Software engineers, system architects, integration architects, and researchers who need to build enterprise access-control systems, adaptive systems, system-of-systems architectures, post-merger IT landscapes, multi-tenant platforms, and federated consortia.

What you need to know

• Python: comfort with classes, decorators, and async (asyncio).
• Basics of RBAC: what roles and permissions are.
• No prior knowledge of CZOA, UniLog, or category theory is required.

Reading this guide

Sections 3–7 are the core tutorial — read them in order. Sections 8–13 are feature guides. Section 14 is the integration guide. Sections 16–20 are reference material. Every code snippet in this guide is self-contained and runnable.

2. Installation

Requirements

• Python 3.9 or later
• The unilog-toolkit package (installed automatically)

Standard install

pip install czoi-toolkit

With optional features

pip install czoi-toolkit[neural]     # scikit-learn for RoleMiner
pip install czoi-toolkit[embedding]  # sentence-transformers for semantic embeddings
pip install czoi-toolkit[dev]        # pytest, ruff, mypy

Verify the installation

import czoi
print(czoi.__version__)   # 1.1.0

From source

git clone https://github.com/hongxueharriswang/czoi-toolkit.git
cd czoi-toolkit
pip install -e ".[dev,neural,embedding]"

3. Your First CZOI System

from czoi import (
    Application, CZOABuilder, Decision, Operation, Role, User,
)

# 1. Create the system and its zones
builder = CZOABuilder("AcmeCorp")
company = builder.root
hr      = builder.add_zone("HR", parent=company, atomic=True)

# 2. Define an application and its operations
hr_app = Application("HRSystem", zone=hr)
view_employee = hr_app.add_operation(Operation("view_employee"))
edit_employee = hr_app.add_operation(Operation("edit_employee"))
hr.add_application(hr_app)

# 3. Create roles
manager = Role("Manager", zone=hr,
               base_permissions=[view_employee, edit_employee])
assistant = Role("Assistant", zone=hr, base_permissions=[view_employee])
manager.add_junior(assistant)
hr.add_role(manager)
hr.add_role(assistant)

# 4. Create users (containment principle enforced automatically)
alice = User("alice", roles={"Assistant"})
bob   = User("bob",   roles={"Manager"})
company.add_user(alice); hr.add_user(alice)
company.add_user(bob);   hr.add_user(bob)

# 5. Check permissions
engine = builder.permission_engine
print(engine.decide(alice, view_employee, hr).name)   # ALLOW
print(engine.decide(alice, edit_employee, hr).name)   # INCONCLUSIVE
print(engine.decide(bob,   edit_employee, hr).name)   # ALLOW

What just happened: CZOABuilder wired the runtime; add_junior() established intra-zone seniority; add_user() enforced containment; decide() returned one of ALLOW, DENY, or INCONCLUSIVE.

4. Core Concepts

The 10‑tuple

S = (Z, R, U, A, O, N, E, Γ, Φ, Δ)

Three kinds of zone

• CompositeZone (the default) — may contain child zones.
• AtomicZone — a leaf; add_zone raises TypeError.
• AdapterZone — a leaf wrapping a legacy or external system through a declared interface.

Applications vs Operations

Permissions are granted on operations, never on applications. This preserves the orthogonality of the ten components and makes operations the atomic interface contract between subsystems.

5. Building Zone Hierarchies

builder = CZOABuilder("University")
uni       = builder.root
coe       = builder.add_zone("CollegeOfEngineering", parent=uni)
cs        = builder.add_zone("CSDepartment", parent=coe, atomic=True)
registrar = builder.add_zone("Registrar", parent=uni, atomic=True)

The containment principle

U_child ⊆ U_parent

Enforced automatically. Registering a user in a child zone requires registering them in every ancestor first.

Properties on zones

cs.properties.set("capacity", 500, type_hint="int")
cs.properties.get("capacity")   # 500

# Type hints are enforced:
cs.properties.set("capacity", "many", type_hint="int")
# TypeError: capacity: expected int, got 

6. Roles, Users, and Operations

doctor = Role("Doctor", zone=hospital,
              base_permissions=[prescribe, view_patient, order_lab])
hospital.add_role(doctor)

# Seniority (intra-zone inheritance):
attending.add_junior(resident)

# Granting and revoking:
hospital.grant(role, operation)   # invalidates cache
hospital.revoke(role, operation)  # invalidates cache

Creating users

alice = User(
    "alice",
    roles={"Doctor"},
    attributes={"employee_id": "E-1234", "shift": "day"},
    credentials={"password_hash": "..."},
)
for z in zone.ancestry():
    z.add_user(alice)

7. The Permission Calculus

from czoi import Decision
decision = engine.decide(user, operation, zone)
# Decision.ALLOW / Decision.DENY / Decision.INCONCLUSIVE

Three values:

• ALLOW — a covering role was found and constraints passed.
• DENY — a covering role was found but a constraint forbids the operation, or the recursion reached the root with no covering role.
• INCONCLUSIVE — no covering role in this zone. A parent might still grant.

Caching

print(engine.stats())
# {'hits': 43, 'misses': 1180, 'parent_lookups': 4, 'denies': 0}

Enabling audit

from czoi import PermissionEngine
builder.permission_engine = PermissionEngine(
    cache_enabled=True, audit_enabled=True,
)
for zone in builder.root.walk():
    zone.set_permission_engine(builder.permission_engine)

Each DecisionRecord has timestamp, user, operation, zone, decision, and stage.

8. Constraints with UniLog

The four families — Γ = (I, T, G, C) — Identity, Trigger, Goal, Access — are expressed in UniLang.

builder.add_access_constraint("""
    signature {
        sort User, Role;
        constant Doctor : Role;
        constant Auditor : Role;
        predicate hasRole(u: User, r: Role);
    }
    forall u: User .
        not (hasRole(u, Doctor) and hasRole(u, Auditor))
""")

Built-in predicates

9. Daemons and Signals

from czoi import Daemon, DaemonSignal

class BatteryDaemon(Daemon):
    def __init__(self, robots, threshold=0.2, parent=None):
        super().__init__("BatteryDaemon", parent=parent, interval=1.0)
        self.robots = robots
        self.threshold = threshold

    def monitor(self):
        for r in self.robots:
            batt = r.attributes.get("battery", 1.0)
            if batt < self.threshold:
                self.emit_signal(DaemonSignal.STATE_WARNING,
                                 {"robot": r.name, "battery": batt})

Running daemons

for step in range(60):
    sim.step()
    builder.daemon_manager.tick()   # synchronous

# Or, for production:
# await builder.daemon_manager.run(duration=300.0)

10. Neural Components

from czoi import Predictor, AnomalyDetector, RoleMiner
import numpy as np

# Predictor
predictor = Predictor("sepsis", threshold=0.85)
predictor.fit(X, y, epochs=500, lr=0.05)
predictor.predict({"hr": 0.8, "temp": 0.9, "lactate": 0.7})

# AnomalyDetector
detector = AnomalyDetector("access", input_dim=5, latent_dim=3)
detector.fit(X_normal, epochs=300, lr=0.05)
detector.is_anomalous(x)

# RoleMiner
miner = RoleMiner(latent_dim=4, min_cluster_size=2, seed=42)
result = miner.mine(X_binary, operation_names, min_support=0.5)

11. Semantic Embeddings

from czoi import EmbeddingService

emb = EmbeddingService(dimension=64)              # hash-based
emb = EmbeddingService(model_name="all-MiniLM-L6-v2",
                       use_transformer=True)      # or transformer-backed

v_op   = emb.embed_operation(operation)
v_role = emb.embed_role(role)
score  = emb.similarity(v_op, v_role)

# Train the global alignment functor with contrastive loss:
emb.train_alignment(positives=[(a, b)], negatives=[(a, c)])

12. Adaptive Access Control

from czoi import NeuralContribution

def surge_hook(user, operation, zone, base_decision):
    if (operation.name == "prescribe"
            and user.has_role("SeniorNurse")
            and zone.properties.get("surge_active", False)):
        return Decision.ALLOW
    return base_decision

builder.permission_engine.set_neural_contribution(
    NeuralContribution(surge_hook)
)

The hook receives the base decision and can return the base (unchanged), Decision.ALLOW (elevate), or Decision.DENY (lock down).

Safety guarantees (Theorem 7)

1. It preserves monotonicity.
2. It satisfies all identity and access constraints.
3. It maintains a complete audit trail.

13. Audit and Observability

for r in builder.permission_engine.audit[-10:]:
    print(f"{r.timestamp:%H:%M:%S}  {r.user:<12} "
          f"{r.operation:<28} @{r.zone:<16} "
          f"→ {r.decision.name:<14} via {r.stage}")

Example output:

14:23:01  alice        EMR.prescribe                 @Emergency       → ALLOW          via local
14:23:02  bob          EMR.prescribe                 @Emergency       → INCONCLUSIVE   via parent-override
14:23:03  bob          EMR.dispense                  @Emergency       → ALLOW          via local

Engine statistics

stats = builder.permission_engine.stats()
# {'hits': 43, 'misses': 1180, 'parent_lookups': 4, 'denies': 12}

14. Integrating Heterogeneous Systems

This section is the integration guide. It shows how to wrap legacy systems as adapter zones, federate independent subsystems under shared governance, discover semantically equivalent roles across subsystems, and verify that integration is sound.

Adapter zones

from czoi import AdapterZone, Operation, Role

class LegacyHRAdapter:
    def __init__(self, connection_string):
        self.conn = legacy_connect(connection_string)

    def execute(self, operation_name, user_name, kwargs):
        if operation_name == "LegacyHR.view_employee":
            return self.conn.read_employee(kwargs["employee_id"])
        elif operation_name == "LegacyHR.edit_employee":
            return self.conn.update_employee(
                kwargs["employee_id"], kwargs["changes"])
        raise ValueError(f"Unknown operation: {operation_name}")

    def describe_interface(self):
        return {"operations": ["view_employee", "edit_employee"],
                "roles": ["HRViewer", "HREditor"]}

legacy_hr = AdapterZone(
    name="LegacyHR",
    adapter=LegacyHRAdapter("sql://legacy-hr.internal"),
)
legacy_hr.expose_operations({
    "view_employee": Operation("view_employee"),
    "edit_employee": Operation("edit_employee"),
})

hr_viewer = Role("HRViewer", zone=legacy_hr,
                 base_permissions=[legacy_hr.operations["view_employee"]])
hr_editor = Role("HREditor", zone=legacy_hr,
                 base_permissions=list(legacy_hr.operations.values()))
hr_editor.add_junior(hr_viewer)
legacy_hr.add_role(hr_viewer)
legacy_hr.add_role(hr_editor)

From the federation's perspective, legacy_hr is indistinguishable from a natively built zone. The legacy system has not been modified in any way.

Federating independent subsystems

from czoi import FederationBuilder

fed = FederationBuilder("MergedHealthAuthority")
fed.adopt_subsystem(legacy_hr_a, name="LegacyRegionA")
fed.adopt_subsystem(legacy_hr_b, name="LegacyRegionB")
fed.adopt_subsystem(shared_services, name="SharedServices")

fed.share_constraint("""
    signature {
        sort User, Role;
        constant HRViewer : Role;
        constant FinanceApprover : Role;
        predicate hasRole(u: User, r: Role);
    }
    forall u: User .
        not (hasRole(u, HRViewer) and hasRole(u, FinanceApprover))
""")

fed.add_cross_zone_daemon(FederationHealthDaemon())

engine = fed.permission_engine
print(engine.decide(alice, view_employee, legacy_hr_a).name)

Discovering role equivalence across subsystems

def discover_role_equivalences(zone_a, zone_b, threshold=0.75):
    """Find semantically equivalent roles across two subsystems."""
    equivalences = []
    for role_a in zone_a.roles.values():
        vec_a = emb.embed_role(role_a)
        for role_b in zone_b.roles.values():
            vec_b = emb.embed_role(role_b)
            sim = emb.similarity(vec_a, vec_b)
            if sim > threshold:
                equivalences.append((role_a, role_b, sim))
    return sorted(equivalences, key=lambda x: -x[2])

Cross-zone integration relations

Two sibling zones compose via a cross-zone integration relation — a 4-tuple (γ, C, δ, ε):

• γ — inter-zone role mappings.
• C — shared constraints.
• δ — cross-zone daemons.
• ε — cross-zone embedding alignment.

fed.integrate(
    from_zone=legacy_hr_a,
    to_zone=legacy_hr_b,
    gamma=[(legacy_hr_a.roles["HRViewer"],
            legacy_hr_b.roles["HRViewer"], 1.0)],
    shared_constraints=[
        """forall u: User .
           not (hasRole(u, HRViewer) and hasRole(u, FinanceApprover))""",
    ],
    cross_zone_daemons=[CrossRegionDriftDaemon()],
    embedding_alignment=True,
)

The Integration Theorem

Let S₁ and S₂ be two independent, soundly integrated CZOI systems with no shared operations. Then the composed federation S = S₁ ⊕ S₂ under a new common parent satisfies:

1. S is a well-formed CZOI system.
2. Every identity constraint of S₁ and S₂ holds in S.
3. Every access constraint of S₁ and S₂ holds in S.
4. Any additional constraint on the parent is enforced for both subsystems.

Corollary (Post-Merger Integration Guarantee). In a post-merger scenario where two legacy organizations are integrated under a new parent authority, if both legacy systems are soundly integrated CZOI systems, the merger preserves each legacy system's security guarantees while enabling explicit cross-system policy at the authority level.

Three integration patterns

End-to-end integration example

from czoi import (
    AdapterZone, CZOABuilder, FederationBuilder, Operation, Role, User,
)

# 1. Wrap the legacy HR system
legacy_hr = AdapterZone(name="LegacyHR", adapter=MyHRAdapter())
legacy_hr.expose_operations({"view_employee": Operation("view_employee")})

# 2. Build the new finance system
finance_builder = CZOABuilder("Finance")
finance_app = Application("FinanceApp", zone=finance_builder.root)
approve = finance_app.add_operation(Operation("approve_invoice"))
finance_builder.root.add_application(finance_app)
finance_builder.root.add_role(
    Role("FinanceApprover", zone=finance_builder.root,
         base_permissions=[approve]))

# 3. Federate under a shared parent
fed = FederationBuilder("MergedCorp")
fed.adopt_subsystem(legacy_hr, name="LegacyHR")
fed.adopt_subsystem(finance_builder, name="Finance")

# 4. Share a separation-of-duty constraint
fed.share_constraint("""
    forall u: User .
        not (hasRole(u, HRViewer) and hasRole(u, FinanceApprover))
""")

# 5. Create users and query
engine = fed.permission_engine
alice = User("alice", roles={"HRViewer"})
fed.root.add_user(alice); legacy_hr.add_user(alice)
print(engine.decide(alice, view_employee, legacy_hr).name)   # ALLOW

Verifying integration soundness

from czoi.federation import verify_integration

report = verify_integration(fed)
print(report.summary())
# Integration Report
# ------------------
# Well-formedness       : OK
# Identity preservation : OK
# Access preservation   : OK
# Constraint flow       : OK
# Cross-zone daemons    : OK
# Audit completeness    : OK

15. Persistence and Web Frameworks

The toolkit is persistence-agnostic. Zones are runtime objects; persistence is a separate concern.

1. Runtime — CZOI zones, roles, operations.
2. Persistence — ordinary ORM models that reference zones by name.
3. Service — glue code that reads persisted data, drives the runtime, writes results.

Example: Django

from django.db import models

class Substation(models.Model):
    name = models.CharField(max_length=64, unique=True)
    voltage_kv = models.FloatField()
    capacity_mw = models.FloatField()

class SensorReading(models.Model):
    substation = models.ForeignKey(Substation, on_delete=models.CASCADE)
    timestamp = models.DateTimeField(auto_now_add=True, db_index=True)
    current_a = models.FloatField()
    voltage_kv = models.FloatField()

Example: FastAPI

from fastapi import FastAPI, Depends, HTTPException
from czoi import Decision

app = FastAPI()
service = GridService()

@app.post("/dispatch/{operation}")
def dispatch(operation: str, user: str = Depends(current_user)):
    decision = service.dispatch(user, operation)
    if decision is not Decision.ALLOW:
        raise HTTPException(403, detail=decision.name)
    return {"status": "ok"}

Persistence in a federation

The federation itself only persists the integration relations (γ, C, δ, ε), the shared constraints, the cross-subsystem audit trail, and the role equivalence mappings. Subsystems keep their own data.

16. Complete Worked Examples

Example A — Native hospital system

from czoi import (
    Application, CZOABuilder, Daemon, DaemonSignal, Decision,
    Operation, Predictor, Role, User,
)

builder = CZOABuilder("RegionalHealthAuthority")
region = builder.root
hosp_a = builder.add_zone("CityHospitalA", parent=region)
emerg_a = builder.add_zone("Emergency", parent=hosp_a, atomic=True)

app = Application("EMR", zone=hosp_a)
prescribe = app.add_operation(Operation("prescribe"))
dispense = app.add_operation(Operation("dispense"))
hosp_a.add_application(app)

attending = Role("AttendingPhysician", zone=hosp_a,
                 base_permissions=[prescribe])
senior_nurse = Role("SeniorNurse", zone=hosp_a,
                    base_permissions=[dispense])
nurse = Role("Nurse", zone=hosp_a, base_permissions=[dispense])
senior_nurse.add_junior(nurse)

for r in (attending, senior_nurse, nurse):
    hosp_a.add_role(r)

alice = User("alice", roles={"AttendingPhysician"})
bob = User("bob", roles={"SeniorNurse"})
for u in (alice, bob):
    for z in emerg_a.ancestry():
        z.add_user(u)

# ... (see tutorial Chapter 15 for the full example)

Example B — Post-merger federation

Two legacy health authorities are integrated under a new parent authority. Each retains its own legacy EMR; a shared services zone provides cross-region workflows. The full example is in the tutorial Chapter 15 and the repository's examples/post_merger.py.

17. Cookbook

Find all zones where a user has a role

def zones_with_role(user, role_name):
    return [z for z in builder.root.walk()
            if role_name in user.roles and role_name in z.roles]

List every operation a user can perform

def accessible_operations(user, zone):
    engine = builder.permission_engine
    return [op.qualified_name for op in zone.operations.values()
            if engine.decide(user, op, zone) is Decision.ALLOW]

Snapshot the system to JSON

import json

def snapshot(builder):
    def zone_repr(zone):
        return {
            "name": zone.name,
            "roles": list(zone.roles.keys()),
            "operations": list(zone.operations.keys()),
            "users": list(zone.users.keys()),
            "properties": zone.properties.as_dict(),
            "children": [zone_repr(c) for c in zone.zones.values()],
        }
    return zone_repr(builder.root)

Trace a permission decision step by step

def trace_decision(engine, user, operation, zone):
    for z in [zone] + zone.ancestry()[::-1]:
        decision = engine.evaluate_local(user, operation, z)
        print(f"  @{z.name}: {decision.name}")
        if decision is not Decision.INCONCLUSIVE:
            break

Discover semantically equivalent roles across subsystems

def find_equivalences(zone_a, zone_b, threshold=0.75):
    results = []
    for role_a in zone_a.roles.values():
        vec_a = emb.embed_role(role_a)
        for role_b in zone_b.roles.values():
            vec_b = emb.embed_role(role_b)
            sim = emb.similarity(vec_a, vec_b)
            if sim > threshold:
                results.append((role_a.name, role_b.name, sim))
    return sorted(results, key=lambda x: -x[2])

Federate two subsystems

from czoi import FederationBuilder

fed = FederationBuilder("MergedCorp")
fed.adopt_subsystem(subsystem_a, name="A")
fed.adopt_subsystem(subsystem_b, name="B")
fed.share_constraint("""...""")
fed.integrate(from_zone=subsystem_a, to_zone=subsystem_b,
              gamma=[(role_a, role_b, 1.0)])
fed.add_cross_zone_daemon(FederationHealthDaemon())

18. Troubleshooting

ZoneContainmentError

You tried to register a user in a child zone without registering them in the parent first.

# Wrong
emergency.add_user(alice)

# Right
for z in emergency.ancestry():
    z.add_user(alice)

TypeError: AtomicZone cannot have child zones

You tried to add a child to a leaf. Make the zone composite (the default) or don't add children to it.

UniLogBridgeError

pip install unilog-toolkit

Decision.INCONCLUSIVE when you expect DENY

INCONCLUSIVE means "no covering role." DENY means "covering role, but a constraint blocked it." If you want boolean semantics:

allowed = engine.decide(user, op, zone) is Decision.ALLOW

Cache staleness

# Ensure you used the zone-level method
hospital.grant(role, operation)   # invalidates cache
# Not:
role.grant(operation)             # does NOT invalidate cache

Integration-specific issues

Conflicting operations: rename or use adapter zone prefixes. Role equivalence not discovered: lower the embedding threshold or train the alignment functor on domain-specific pairs. Integration validation failure: run verify_integration(fed) to identify the specific constraint.

19. Best Practices

Structure

• Model the organisation, not the code.
• Use atomic zones generously for anything that will never grow.
• Name operations with verb_noun: view_patient, not viewPatient.

Roles

• One role per job function.
• Prefer seniority over duplication.
• Keep role hierarchies shallow — depth 3 is usually enough.

Operations

• Atomic operations, not composite actions.
• Fine granularity for security-critical operations.
• Application grouping is for humans, not for permissions.

Constraints

• Encode policy, not implementation.
• One constraint per rule.
• Test constraints in isolation.

Daemons

• One responsibility per daemon.
• Signal, don't act.
• Use per-daemon intervals.

Neural components

• Normalise features. Always.
• Fit on normal data only for anomaly detection.
• Version your models.

Testing

• Test with a fresh builder per test.
• Test ALLOW, DENY, and INCONCLUSIVE separately.
• Test constraints explicitly.

Integration

• Start with adapter zones, not refactoring.
• Use the alignment functor to discover role equivalence, but review manually.
• Prefer shared constraints over duplicated ones.
• Add cross-zone daemons from day one.
• Verify integration soundness programmatically after every change.
• Document the integration relation — the (γ, C, δ, ε) tuple is the contract.
• Preserve local autonomy explicitly — it's a property to protect, not a cost to minimize.
• Test integration across subsystem boundaries in both directions.

Performance

• Keep cache_enabled=True and monitor stats()["hits"].
• Avoid wide fan-outs — prefer smaller roles with seniority relations.
• Deep trees are fine; recursion depth adds ~0.02 ms per level.
• Cross-zone decisions add ~0.1 ms per subsystem boundary crossed.

20. API Reference

CZOABuilder

FederationBuilder

AdapterZone

ZoneBase

PermissionEngine

ConstraintManager

Daemon, DaemonSignal, DaemonManager

Predictor, AnomalyDetector, RoleMiner

EmbeddingService

Integration utilities

from czoi.federation import verify_integration, discover_role_equivalences

report = verify_integration(fed)
print(report.summary())

equivalences = discover_role_equivalences(zone_a, zone_b,
                                          embedding_service=emb,
                                          threshold=0.75)

Exceptions