K501-AIONARC - Part 01 - Codebase of the Transition-Centric System

A developer released Part 1 of the K501-AIONARC codebase, the Transition-Centric System, featuring canonical serialization and identity generation utilities. The code includes a strict RFC 8785 JCS engine for deterministic hashing and a graph model for transition-based state management.

Here is Part 1 of the complete Google Gemini interpretation from the first phase of K501-AIONARC codebase, the Transition-Centric System . This first batch contains the fundamental utilities and the core graph models. This version of code is not implemeted or tested jet, its just a referenz and exploration, a Evolution of the code used in RealDeal Build. src/k501/utils/canonical serialization.py Python """ Canonical Serialization: Strict JCS RFC 8785 Engine for K501-Aionarc. Guarantees mathematically unique byte sequences for deterministic IDs. """ import json import hashlib from typing import Any def canonical serialize obj: Any - str: """ Serializes an object to canonical JSON according to RFC 8785 JCS. """ def serialize value: Any - str: if value is None: return "null" if isinstance value, bool : return "true" if value else "false" if isinstance value, int : return str value if isinstance value, float : Float normalization: prevents discrepancies between 1 and 1.0 if value.is integer : return str int value return f"{value:.10f}".rstrip '0' .rstrip '.' if isinstance value, str : Standard JSON escaping, forced to UTF-8 plaintext return json.dumps value, ensure ascii=False if isinstance value, list : return " " + ",".join serialize item for item in value + " " if isinstance value, dict : Key sorting strictly by Unicode code-points sorted keys = sorted value.keys , key=lambda k: k.encode 'utf-8' return "{" + ",".join f'"{k}":{ serialize value k }' for k in sorted keys + "}" raise TypeError f"Invalid data type for canonical serialization: {type value }" return serialize obj def canonical hash obj: Any - str: """Generates a SHA-256 hash of the canonically serialized object.""" canonical str = canonical serialize obj return hashlib.sha256 canonical str.encode 'utf-8' .hexdigest src/k501/utils/identity.py Python """ Identity Engine: Deterministic ID generation based on Transition truth. """ from typing import Any, Dict from .canonical serialization import canonical hash def generate transition id content: Dict str, Any - str: """ Generates a deterministic Transition ID. Format: transition:{sha256 24} """ hash value = canonical hash content return f"transition:{hash value :24 }" def generate state id from transition transition id: str, object type: str, axiom id: str, namespace: str - str: """ Derives the StateObject ID directly from its parent transition. Guarantees that no state can exist without an origin transition. """ content = { "transition id": transition id, "object type": object type.lower , "axiom id": axiom id, "namespace": namespace, } hash value = canonical hash content return f"{object type.lower }:{hash value :24 }" def generate edge id source id: str, edge type: str, target id: str, attributes: Dict str, Any - str: """Generates a deterministic ID for relational graph edges.""" content = { "source id": source id, "edge type": edge type.lower , "target id": target id, "attributes": attributes, } hash value = canonical hash content return f"edge:{hash value :24 }" src/k501/graph/models.py Python """ Graph Models: Transition-centric node and edge data structures. """ from dataclasses import dataclass, field from datetime import datetime, timezone from enum import Enum from typing import Any, Dict, Optional class TransitionType Enum : CREATE = "create" FREEZE = "freeze" ARCHIVE = "archive" SUPERSEDE = "supersede" VALIDATE = "validate" REJECT = "reject" CONTEST = "contest" ACCEPT = "accept" DERIVE = "derive" class EdgeType Enum : DERIVED FROM = "derived from" ANCESTOR = "ancestor" VALIDATED BY = "validated by" AUDITED BY = "audited by" BELONGS TO = "belongs to" REFERENCES = "references" SUPERSEDES = "supersedes" @dataclass frozen=True class TransitionNode: node id: str Corresponds to transition id transition type: TransitionType before state id: Optional str after state id: str timestamp: datetime attributes: Dict str, Any = field default factory=dict @classmethod def create cls, transition id: str, transition type: TransitionType, before state id: Optional str , after state id: str, attributes: Optional Dict str, Any = None, - "TransitionNode": return cls node id=transition id, transition type=transition type, before state id=before state id, after state id=after state id, timestamp=datetime.now timezone.utc , attributes=attributes or {}, def to dict self - Dict str, Any : return { "node id": self.node id, "transition type": self.transition type.value, "before state id": self.before state id, "after state id": self.after state id, "timestamp": self.timestamp.isoformat , "attributes": self.attributes, } @dataclass frozen=True class TransitionEdge: source id: str edge type: EdgeType target id: str attributes: Dict str, Any = field default factory=dict @classmethod def create cls, source id: str, edge type: EdgeType, target id: str, attributes: Optional Dict str, Any = None - "TransitionEdge": return cls source id=source id, edge type=edge type, target id=target id, attributes=attributes or {} def to dict self - Dict str, Any : return { "source id": self.source id, "edge type": self.edge type.value, "target id": self.target id, "attributes": self.attributes, } src/k501/graph/store.py Python """ Provenance Graph Store: SQLite persistence and traversal queries. Operates on the Transition-Centric model. """ import json import sqlite3 from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional from .models import TransitionNode, TransitionEdge, TransitionType, EdgeType class ProvenanceGraph: def init self, db path: str = "data/graph/graph.sqlite" : self.db path = Path db path self.db path.parent.mkdir parents=True, exist ok=True self. init db def init db self - None: """Initializes the transition-centric SQLite schema.""" with sqlite3.connect str self.db path as conn: cursor = conn.cursor Transition nodes cursor.execute """ CREATE TABLE IF NOT EXISTS nodes node id TEXT PRIMARY KEY, transition type TEXT NOT NULL, before state id TEXT, after state id TEXT NOT NULL, timestamp TEXT NOT NULL, attributes TEXT NOT NULL """ Relational edges cursor.execute """ CREATE TABLE IF NOT EXISTS edges source id TEXT NOT NULL, edge type TEXT NOT NULL, target id TEXT NOT NULL, attributes TEXT NOT NULL, PRIMARY KEY source id, edge type, target id """ Performance indexes cursor.execute "CREATE INDEX IF NOT EXISTS idx edges source ON edges source id " cursor.execute "CREATE INDEX IF NOT EXISTS idx edges target ON edges target id " cursor.execute "CREATE INDEX IF NOT EXISTS idx edges type ON edges edge type " cursor.execute "CREATE INDEX IF NOT EXISTS idx nodes after state ON nodes after state id " conn.commit def add node self, node: TransitionNode - bool: try: with sqlite3.connect str self.db path as conn: conn.execute """ INSERT OR REPLACE INTO nodes node id, transition type, before state id, after state id, timestamp, attributes VALUES ?, ?, ?, ?, ?, ? """, node.node id, node.transition type.value, node.before state id, node.after state id, node.timestamp.isoformat , json.dumps node.attributes conn.commit return True except Exception: return False def add edge self, edge: TransitionEdge - bool: try: with sqlite3.connect str self.db path as conn: conn.execute """ INSERT OR REPLACE INTO edges source id, edge type, target id, attributes VALUES ?, ?, ?, ? """, edge.source id, edge.edge type.value, edge.target id, json.dumps edge.attributes conn.commit return True except Exception: return False def get node self, node id: str - Optional TransitionNode : with sqlite3.connect str self.db path as conn: cursor = conn.cursor cursor.execute """ SELECT node id, transition type, before state id, after state id, timestamp, attributes FROM nodes WHERE node id = ? """, node id, row = cursor.fetchone if not row: return None return TransitionNode node id=row 0 , transition type=TransitionType row 1 , before state id=row 2 , after state id=row 3 , timestamp=datetime.fromisoformat row 4 , attributes=json.loads row 5 def get edges self, node id: str, direction: str = "forward", edge types: Optional List EdgeType = None - List TransitionEdge : query = "SELECT source id, edge type, target id, attributes FROM edges WHERE " query += "source id = ?" if direction == "forward" else "target id = ?" with sqlite3.connect str self.db path as conn: cursor = conn.cursor cursor.execute query, node id, rows = cursor.fetchall edges = for row in rows: et = EdgeType row 1 if edge types is None or et in edge types: edges.append TransitionEdge source id=row 0 , edge type=et, target id=row 2 , attributes=json.loads row 3 return edges def traverse self, start id: str, edge types: List EdgeType , direction: str = "backward", depth: int = 10, time filter: Optional str = None - List str : """Traverses the transition graph.""" visited = set result = current level = start id parsed time = datetime.fromisoformat time filter.replace "Z", "+00:00" if time filter else None for in range depth : if not current level: break next level = for node id in current level: if node id in visited: continue if parsed time: curr node = self.get node node id if not curr node or curr node.timestamp parsed time: continue visited.add node id result.append node id edges = self.get edges node id, direction, edge types for edge in edges: target = edge.target id if direction == "forward" else edge.source id if target not in visited: next level.append target current level = next level return result def get state lineage self, state id: str - List str : """Finds all transition IDs that contributed to a state lineage.""" with sqlite3.connect str self.db path as conn: cursor = conn.cursor Find the transition that created this state cursor.execute "SELECT node id FROM nodes WHERE after state id = ?", state id, rows = cursor.fetchall if not rows: return start transition id = rows 0 0 return self.traverse start id=start transition id, edge types= EdgeType.DERIVED FROM, EdgeType.ANCESTOR , direction="backward", depth=100 src/k501/graph/traverse.py Python """ Graph Traversal Utilities: High-level operations on the transition graph. """ from typing import List from .store import ProvenanceGraph def lineage traversal graph: ProvenanceGraph, state id: str - List str : """Gets the full causal transition lineage for a given StateObject ID.""" return graph.get state lineage state id def ancestry depth graph: ProvenanceGraph, state id: str - int: """Calculates the length of the ancestry chain.""" return len graph.get state lineage state id def common ancestors graph: ProvenanceGraph, state id 1: str, state id 2: str - List str : """Finds intersection of transition histories between two states.""" lineage1 = set graph.get state lineage state id 1 lineage2 = set graph.get state lineage state id 2 return list lineage1.intersection lineage2 src/k501/graph/query.py Python """ Graph Query: Specialized queries for searching transition chains. """ from typing import Any, Dict, List from .models import EdgeType from .store import ProvenanceGraph def query lineage graph: ProvenanceGraph, transition id: str, max depth: int = 100 - List str : """Queries derivation history starting from a specific transition.""" return graph.traverse start id=transition id, edge types= EdgeType.DERIVED FROM, EdgeType.ANCESTOR , direction="backward", depth=max depth def query provenance audits graph: ProvenanceGraph, transition id: str, max depth: int = 10 - Dict str, List str : """Queries the validation and audit history for a specific transition.""" proofs = graph.traverse start id=transition id, edge types= EdgeType.VALIDATED BY , direction="backward", depth=max depth audits = graph.traverse start id=transition id, edge types= EdgeType.AUDITED BY , direction="backward", depth=max depth return { "proofs": proofs, "audits": audits } def query before time graph: ProvenanceGraph, transition id: str, time anchor: str, max depth: int = 100 - List str : """Queries all ancestor transitions restricted by a temporal boundary.""" return graph.traverse start id=transition id, edge types= EdgeType.DERIVED FROM, EdgeType.ANCESTOR , direction="backward", depth=max depth, time filter=time anchor Here is Part 3 of 3. This final section contains the causal validation engine, the deterministic replay verifier, the package initialization, and the core integration test. src/k501/graph/validator.py Python """ Transition-Centric Graph Validator: Enforces causality and DAG invariance. """ from typing import Any, Dict, List, Set from .models import TransitionEdge, TransitionNode, EdgeType, TransitionType class GraphValidationError: def init self, error type: str, message: str, details: Dict str, Any = None : self.error type = error type self.message = message self.details = details or {} def to dict self - Dict str, Any : return {"error type": self.error type, "message": self.message, "details": self.details} class TransitionGraphValidator: def init self, nodes: Dict str, TransitionNode , edges: List TransitionEdge : self.nodes = nodes self.edges = edges self. build adjacency def build adjacency self - None: self.forward edges: Dict str, List TransitionEdge = {} self.backward edges: Dict str, List TransitionEdge = {} for edge in self.edges: self.forward edges.setdefault edge.source id, .append edge self.backward edges.setdefault edge.target id, .append edge def validate self - List GraphValidationError : errors = errors.extend self.validate transition chain errors.extend self.validate dag return errors def validate transition chain self - List GraphValidationError : errors = state transitions: Dict str, List TransitionNode = {} for node in self.nodes.values : state transitions.setdefault node.after state id, .append node for state id, transitions in state transitions.items : Rule 1: Every state needs exactly one origin transition CREATE has create = any t.transition type == TransitionType.CREATE for t in transitions if not has create: errors.append GraphValidationError "MISSING CREATE", f"State {state id} has no CREATE transition." continue Rule 2: Enforce chronological integrity of the lifecycle create ts = next t.timestamp for t in transitions if t.transition type == TransitionType.CREATE , None freeze ts = next t.timestamp for t in transitions if t.transition type == TransitionType.FREEZE , None archive ts = next t.timestamp for t in transitions if t.transition type == TransitionType.ARCHIVE , None if freeze ts and create ts and freeze ts < create ts: errors.append GraphValidationError "CHRONOLOGY VIOLATION", f"State {state id}: FREEZE occurs before CREATE." if archive ts and freeze ts and archive ts < freeze ts: errors.append GraphValidationError "CHRONOLOGY VIOLATION", f"State {state id}: ARCHIVE occurs before FREEZE." return errors def validate dag self - List GraphValidationError : errors = derive adj: Dict str, List str = {} for edge in self.edges: if edge.edge type == EdgeType.DERIVED FROM: derive adj.setdefault edge.source id, .append edge.target id visited: Set str = set rec stack: Set str = set def has cycle node: str - bool: visited.add node rec stack.add node for neighbor in derive adj.get node, : if neighbor not in visited: if has cycle neighbor : return True elif neighbor in rec stack: return True rec stack.remove node return False for node id in derive adj: if node id not in visited: if has cycle node id : errors.append GraphValidationError "CYCLE DETECTED", "Cyclic dependency detected in DERIVED FROM edges.", {"node id": node id} return errors src/k501/graph/replay validator.py Python """ Replay Validator: Verifies absolute replay determinism of K501-Aionarc. """ from typing import Dict, List from .models import TransitionEdge, TransitionNode from .validator import GraphValidationError class ReplayValidator: def init self, original nodes: Dict str, TransitionNode , original edges: List TransitionEdge , replayed nodes: Dict str, TransitionNode , replayed edges: List TransitionEdge , : self.original nodes = original nodes self.original edges = original edges self.replayed nodes = replayed nodes self.replayed edges = replayed edges def validate replay self - List GraphValidationError : errors = 1. Check structural identity of nodes orig node ids = set self.original nodes.keys repl node ids = set self.replayed nodes.keys if orig node ids = repl node ids: errors.append GraphValidationError "REPLAY NODE MISMATCH", "Node sets of original and replay do not match.", { "missing in replay": list orig node ids - repl node ids , "extra in replay": list repl node ids - orig node ids } 2. Check structural identity of edges orig edge signatures = { e.source id, e.edge type.value, e.target id for e in self.original edges} repl edge signatures = { e.source id, e.edge type.value, e.target id for e in self.replayed edges} if orig edge signatures = repl edge signatures: errors.append GraphValidationError "REPLAY EDGE MISMATCH", "Edge topology differs after replay.", { "missing edges": list orig edge signatures - repl edge signatures , "extra edges": list repl edge signatures - orig edge signatures } 3. Deep-Comparison of state attributes Invariance check for node id in orig node ids.intersection repl node ids : if self.original nodes node id .attributes = self.replayed nodes node id .attributes: errors.append GraphValidationError "REPLAY ATTRIBUTE MUTATION", f"Attribute mutation detected in Node {node id}.", { "original": self.original nodes node id .attributes, "replayed": self.replayed nodes node id .attributes } return errors def is identical self - bool: return len self.validate replay == 0 src/k501/graph/ init .py Python python """ Graph Module Export Hub """ from .models import TransitionNode, TransitionEdge, TransitionType, EdgeType from .store import ProvenanceGraph from .traverse import lineage traversal, ancestry depth, common ancestors from .query import query lineage, query provenance audits, query before time from .validator import TransitionGraphValidator, GraphValidationError from .replay validator import ReplayValidator all = "TransitionNode", "TransitionEdge", "TransitionType", "EdgeType", "ProvenanceGraph", "lineage traversal", "ancestry depth", "common ancestors", "query lineage", "query provenance audits", "query before time", "TransitionGraphValidator", "GraphValidationError", "ReplayValidator", tests/unit/test transition consensus.py Python """ Integrated Verification Test: Validates RFC 8785 and Causal Topology constraints. """ import pytest from datetime import datetime, timezone, timedelta from k501.utils.canonical serialization import canonical serialize, canonical hash from k501.utils.identity import generate transition id, generate state id from transition from k501.graph.models import TransitionNode, TransitionEdge, TransitionType, EdgeType from k501.graph.validator import TransitionGraphValidator def test rfc8785 float and dict determinism : """Different memory layouts and data types must yield identical byte strings.""" dict a = {"alpha": 1.0, "beta": 1, 2 } dict b = {"beta": 1, 2 , "alpha": 1} assert canonical serialize dict a == canonical serialize dict b def test causal lifecycle chronology : """Constructs a corrupt timeline FREEZE before CREATE and validates rejection.""" base time = datetime.now timezone.utc t1 content = {"action": "initial creation"} t1 id = generate transition id t1 content s id = generate state id from transition t1 id, "frame", "axiom:0", "sys" Create temporally invalid nodes node create = TransitionNode t1 id, TransitionType.CREATE, None, s id, base time node freeze = TransitionNode "transition:fake freeze", TransitionType.FREEZE, s id, s id, base time - timedelta seconds=10 Time travel anomaly nodes = {node create.node id: node create, node freeze.node id: node freeze} validator = TransitionGraphValidator nodes, errors = validator.validate assert any err.error type == "CHRONOLOGY VIOLATION" for err in errors Patrick R. Miller Iinkognit0 — K501 / AIONARC Core Architecture ORCID: https://orcid.org/0009-0005-5125-9711 Website: https://iinkognit0.de/ GitHub: https://github.com/Iinkognit0 GitHub: https://github.com/k501-Information-Space/eArc Publications: https://dev.to/k501is Mastodon: https://mastodon.social/@K501 Email: contact.k501@proton.me Unix Epoch 1781737019