#!/usr/bin/env python3 """Deterministic, candidate-only Formation Mind kernel for one canonical source. This is a deliberately bounded bridge between an immutable source chapter and the governed Identity Field. It does not infer meaning, approve semantic relations, promote identities, or claim that display geometry proves a fact. The first configured source is ``canonical/16_ENGINE_FORMATION.md``. A later orchestrator may run the same one-source contract for additional files, but one kernel run accepts exactly one canonical Markdown file. Formation rules are intentionally inspectable: * source occurrences have content-addressed byte/token addresses; * every occurrence enters as a triadic PatternThread/MindCell; * repeated surface patterns and ordered contacts build candidate basins; * attraction is fixed-step integer state derived from current structure; * branch, attach, deform, and retire decisions are append-only events; * relation basins are children of pattern basins, making recursion explicit; * a relational-position next-basin prediction and two declared baselines are issued before their target is observed and scored afterward; * all semantic authority remains outside this kernel. No LLM, embedding, ontology category, moral rank, or hidden clock is used. """ from __future__ import annotations import argparse import hashlib import json import re import unicodedata from dataclasses import dataclass, field from pathlib import Path, PurePosixPath from typing import Any, Iterable, Mapping, Sequence ROOT = Path(__file__).resolve().parents[1] DEFAULT_SOURCE_RELPATH = "canonical/16_ENGINE_FORMATION.md" KERNEL_VERSION = "FMK-2026.09.01.2" TOKENIZER_VERSION = "unicode-word-symbol-v2-markdown-body" GENESIS_HASH = "0" * 64 MASK_64 = (1 << 64) - 1 # A word (with an optional internal apostrophe/hyphen) or one non-space symbol. # This preserves grammar-bearing punctuation as addressable source occurrences. TOKEN_RE = re.compile(r"\w+(?:[’'-]\w+)*|[^\w\s]", re.UNICODE) ROLE_NAMES = ("inside", "outside", "relation") CANDIDATE_GUARDRAIL = ( "Computational formation candidate only; no referent identity, semantic " "relation, truth, proof, observerhood, or consciousness is established." ) def canonical_bytes(value: Any) -> bytes: """Encode a value for stable hashing.""" return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") def sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def stable_digest(value: Any) -> str: return sha256_bytes(canonical_bytes(value)) def _assert_integer_tree(value: Any, path: str = "root") -> None: """Reject runtime floats while allowing strings, nulls, and booleans.""" if isinstance(value, float): raise TypeError(f"Floating-point state is forbidden at {path}") if isinstance(value, Mapping): for key, item in value.items(): _assert_integer_tree(item, f"{path}.{key}") elif isinstance(value, (list, tuple)): for index, item in enumerate(value): _assert_integer_tree(item, f"{path}[{index}]") def _rotate_left_64(value: int, count: int) -> int: count %= 64 return ((value << count) | (value >> (64 - count))) & MASK_64 def _digest_seed(hex_digest: str, offset: int = 0) -> int: return int(hex_digest[offset : offset + 16], 16) def _seeded_position(hex_digest: str, coordinate_bound: int) -> list[int]: """Place a new candidate reproducibly without claiming semantic geometry.""" span = coordinate_bound * 2 + 1 return [ int(hex_digest[index : index + 16], 16) % span - coordinate_bound for index in (0, 16, 32) ] def _sign(value: int) -> int: return (value > 0) - (value < 0) def _surface_bit_compatibility(left_pattern: str, right_pattern: str) -> int: """Return a 0..256 score from the addressed surface bytes themselves. Content digests identify state; their pseudo-random bit proximity must not masquerade as structural compatibility. This bounded score instead uses aligned normalized UTF-8 bytes, so its complete native mechanism is visible. Missing bytes contribute no agreement. """ left = left_pattern.encode("utf-8") right = right_pattern.encode("utf-8") maximum_bits = max(len(left), len(right)) * 8 if maximum_bits == 0: return 0 agreement = sum(8 - (a ^ b).bit_count() for a, b in zip(left, right)) return agreement * 256 // maximum_bits def _normalized_pattern(label: str) -> str: return unicodedata.normalize("NFC", label).casefold() @dataclass(frozen=True) class FormationConfig: """Explicit one-source and fixed-step formation contract.""" source_relpath: str = DEFAULT_SOURCE_RELPATH context_window: int = 8 coordinate_bound: int = 32_768 velocity_bound: int = 256 phase_modulus: int = 12 force_divisor: int = 64 maximum_pull: int = 8 minimum_persistence_occurrences: int = 2 maximum_tokens: int | None = None def __post_init__(self) -> None: relative = PurePosixPath(self.source_relpath.replace("\\", "/")) if relative.is_absolute() or ".." in relative.parts: raise ValueError("source_relpath must be a safe relative path") if not relative.parts or relative.parts[0] != "canonical" or relative.suffix.lower() != ".md": raise ValueError("one Formation Mind run requires one canonical Markdown source") positive = { "context_window": self.context_window, "coordinate_bound": self.coordinate_bound, "velocity_bound": self.velocity_bound, "phase_modulus": self.phase_modulus, "force_divisor": self.force_divisor, "maximum_pull": self.maximum_pull, "minimum_persistence_occurrences": self.minimum_persistence_occurrences, } invalid = [name for name, value in positive.items() if not isinstance(value, int) or value < 1] if invalid: raise ValueError(f"positive integer Formation Mind settings required: {invalid}") if self.maximum_tokens is not None and ( not isinstance(self.maximum_tokens, int) or self.maximum_tokens < 1 ): raise ValueError("maximum_tokens must be null or a positive integer") def to_dict(self) -> dict[str, Any]: return { "source_relpath": self.source_relpath.replace("\\", "/"), "context_window": self.context_window, "coordinate_bound": self.coordinate_bound, "velocity_bound": self.velocity_bound, "phase_modulus": self.phase_modulus, "force_divisor": self.force_divisor, "maximum_pull": self.maximum_pull, "minimum_persistence_occurrences": self.minimum_persistence_occurrences, "maximum_tokens": self.maximum_tokens, } @dataclass(frozen=True) class SourceOccurrence: """One immutable, content-addressed token occurrence.""" ordinal: int byte_start: int byte_end: int char_start: int char_end: int label: str normalized_pattern: str token_sha256: str source_sha256: str address: str def to_dict(self) -> dict[str, Any]: return { "ordinal": self.ordinal, "byte_start": self.byte_start, "byte_end": self.byte_end, "char_start": self.char_start, "char_end": self.char_end, "label": self.label, "normalized_pattern": self.normalized_pattern, "token_sha256": self.token_sha256, "source_sha256": self.source_sha256, "address": self.address, } def formation_content_span(source_bytes: bytes) -> tuple[int, int]: """Return the Markdown body span, excluding an opening YAML front matter. The whole-file hash remains the immutable source snapshot identity. The formation operator consumes the body so release metadata cannot masquerade as theory structure. Returned offsets always address the original bytes. """ lines = source_bytes.splitlines(keepends=True) if not lines or lines[0].rstrip(b"\r\n") != b"---": return 0, len(source_bytes) offset = len(lines[0]) for line in lines[1:]: offset += len(line) if line.rstrip(b"\r\n") == b"---": return offset, len(source_bytes) raise ValueError("opening YAML front matter has no closing delimiter") def tokenize_source( source_bytes: bytes, *, source_sha256: str | None = None, byte_offset: int = 0, char_offset: int = 0, ) -> tuple[SourceOccurrence, ...]: """Tokenize UTF-8 bytes while retaining whole-source byte/character spans.""" text = source_bytes.decode("utf-8") source_sha = source_sha256 or sha256_bytes(source_bytes) char_to_byte = [0] byte_count = 0 for character in text: byte_count += len(character.encode("utf-8")) char_to_byte.append(byte_count) occurrences: list[SourceOccurrence] = [] for ordinal, match in enumerate(TOKEN_RE.finditer(text)): label = match.group(0) local_byte_start = char_to_byte[match.start()] local_byte_end = char_to_byte[match.end()] byte_start = byte_offset + local_byte_start byte_end = byte_offset + local_byte_end token_bytes = source_bytes[local_byte_start:local_byte_end] token_sha = sha256_bytes(token_bytes) address = ( f"ytocc:sha256:{source_sha}" f"#bytes={byte_start}:{byte_end};token={ordinal};token_sha256={token_sha}" ) occurrences.append( SourceOccurrence( ordinal=ordinal, byte_start=byte_start, byte_end=byte_end, char_start=char_offset + match.start(), char_end=char_offset + match.end(), label=label, normalized_pattern=_normalized_pattern(label), token_sha256=token_sha, source_sha256=source_sha, address=address, ) ) return tuple(occurrences) def tokenize_formation_source(source_bytes: bytes) -> tuple[SourceOccurrence, ...]: """Tokenize only the authorized Markdown body under the whole-file address.""" byte_start, byte_end = formation_content_span(source_bytes) prefix = source_bytes[:byte_start].decode("utf-8") return tokenize_source( source_bytes[byte_start:byte_end], source_sha256=sha256_bytes(source_bytes), byte_offset=byte_start, char_offset=len(prefix), ) @dataclass(frozen=True) class MindRoleFrame: role: str refs: tuple[str, ...] structural_reading: str def __post_init__(self) -> None: if self.role not in ROLE_NAMES: raise ValueError(f"unsupported role: {self.role}") if not self.refs or len(self.refs) != len(set(self.refs)): raise ValueError(f"{self.role} role requires unique non-empty references") if not self.structural_reading.strip(): raise ValueError(f"{self.role} role requires a structural reading") def to_dict(self) -> dict[str, Any]: return { "role": self.role, "refs": list(self.refs), "structural_reading": self.structural_reading, "semantic_binding": "unresolved_candidate", } @dataclass(frozen=True) class MindCellFrame: identifier: str reference_scope: str inside: MindRoleFrame outside: MindRoleFrame relation: MindRoleFrame capability_boundary: str = CANDIDATE_GUARDRAIL def __post_init__(self) -> None: if (self.inside.role, self.outside.role, self.relation.role) != ROLE_NAMES: raise ValueError("MindCellFrame requires exactly inside, outside, relation") if not self.identifier or not self.reference_scope.strip() or not self.capability_boundary.strip(): raise ValueError("MindCellFrame identifier, scope, and boundary are required") def to_dict(self) -> dict[str, Any]: return { "id": self.identifier, "reference_scope": self.reference_scope, "roles": { "inside": self.inside.to_dict(), "outside": self.outside.to_dict(), "relation": self.relation.to_dict(), }, "authority_state": "computational_candidate_only", "semantic_relation_approved": False, "referent_identity_established": False, "capability_boundary": self.capability_boundary, } @dataclass(frozen=True) class PatternThread: """One source occurrence entering the field through a triadic frame.""" identifier: str occurrence: SourceOccurrence prior_occurrence_address: str | None context_fingerprint: int mind_cell: MindCellFrame def to_dict(self) -> dict[str, Any]: return { "id": self.identifier, "occurrence": self.occurrence.to_dict(), "prior_occurrence_address": self.prior_occurrence_address, "context_fingerprint": self.context_fingerprint, "mind_cell": self.mind_cell.to_dict(), } @dataclass(frozen=True) class FormationEvent: sequence: int fixed_step: int event_type: str payload: Mapping[str, Any] previous_event_hash: str event_hash: str def unhashed_dict(self) -> dict[str, Any]: return { "sequence": self.sequence, "fixed_step": self.fixed_step, "event_type": self.event_type, "payload": dict(self.payload), "previous_event_hash": self.previous_event_hash, } def to_dict(self) -> dict[str, Any]: return {**self.unhashed_dict(), "event_hash": self.event_hash} class HashChainLog: """Single-writer append-only event log exposed as immutable snapshots.""" def __init__(self) -> None: self._events: list[FormationEvent] = [] @property def events(self) -> tuple[FormationEvent, ...]: return tuple(self._events) def append(self, *, fixed_step: int, event_type: str, payload: Mapping[str, Any]) -> FormationEvent: if not isinstance(fixed_step, int) or fixed_step < 0: raise ValueError("fixed_step must be a non-negative integer") if not event_type.strip(): raise ValueError("event_type is required") payload_copy = json.loads(canonical_bytes(payload).decode("utf-8")) _assert_integer_tree(payload_copy, "event.payload") previous_hash = self._events[-1].event_hash if self._events else GENESIS_HASH event = FormationEvent( sequence=len(self._events), fixed_step=fixed_step, event_type=event_type, payload=payload_copy, previous_event_hash=previous_hash, event_hash="", ) event_hash = stable_digest(event.unhashed_dict()) completed = FormationEvent( sequence=event.sequence, fixed_step=event.fixed_step, event_type=event.event_type, payload=event.payload, previous_event_hash=event.previous_event_hash, event_hash=event_hash, ) self._events.append(completed) return completed def verify_event_chain(events: Sequence[FormationEvent | Mapping[str, Any]]) -> bool: previous_hash = GENESIS_HASH for expected_sequence, raw_event in enumerate(events): event = raw_event.to_dict() if isinstance(raw_event, FormationEvent) else dict(raw_event) event_hash = event.pop("event_hash", None) if event.get("sequence") != expected_sequence: return False if event.get("previous_event_hash") != previous_hash: return False if stable_digest(event) != event_hash: return False previous_hash = str(event_hash) return True @dataclass class CandidateBasin: identifier: str basin_kind: str parent_id: str | None depth: int pattern_sha256: str label: str position: list[int] velocity: list[int] phase_tick: int = 0 mass: int = 1 support: int = 1 residue_total: int = 0 context_fingerprint: int = 0 member_refs: list[str] = field(default_factory=list) child_basin_ids: list[str] = field(default_factory=list) relation_refs: list[str] = field(default_factory=list) first_step: int = 0 last_step: int = 0 state: str = "active_candidate" def structural_state(self) -> dict[str, Any]: return { "position": list(self.position), "velocity": list(self.velocity), "phase_tick": self.phase_tick, "mass": self.mass, "support": self.support, "residue_total": self.residue_total, "context_fingerprint": self.context_fingerprint, "member_count": len(self.member_refs), "child_count": len(self.child_basin_ids), "relation_count": len(self.relation_refs), "last_step": self.last_step, "state": self.state, } def mind_cell(self, root_id: str) -> MindCellFrame: if self.basin_kind == "relation_candidate": inside_ref, outside_ref = self.label.split(" -> ", 1) inside_refs = (inside_ref,) outside_refs = (outside_ref,) relation_refs = tuple(self.relation_refs) or (f"candidate_relation:{self.identifier}",) scope = "ordered contact candidate between two pattern basins" elif self.basin_kind == "pattern_candidate": inside_refs = tuple(self.member_refs) or (f"empty_candidate:{self.identifier}",) outside_refs = (self.parent_id or root_id,) relation_refs = tuple(self.child_basin_ids) or (f"unresolved_relations:{self.identifier}",) scope = "surface-pattern candidate within one source formation field" else: inside_refs = tuple(self.child_basin_ids) or (f"empty_field:{self.identifier}",) outside_refs = (f"source_snapshot:{self.pattern_sha256}",) relation_refs = ("kernel:fixed_step_candidate_formation",) scope = "one immutable canonical source formation field" return MindCellFrame( identifier=f"MC::{self.identifier}", reference_scope=scope, inside=MindRoleFrame("inside", inside_refs, "currently retained members or child candidates"), outside=MindRoleFrame("outside", outside_refs, "declared field or counterpart at this reference scope"), relation=MindRoleFrame("relation", relation_refs, "recorded structural contacts; meaning remains unresolved"), ) def to_dict(self, root_id: str) -> dict[str, Any]: return { "id": self.identifier, "basin_kind": self.basin_kind, "parent_id": self.parent_id, "depth": self.depth, "pattern_sha256": self.pattern_sha256, "label": self.label, "member_refs": list(self.member_refs), "child_basin_ids": list(self.child_basin_ids), "relation_refs": list(self.relation_refs), "first_step": self.first_step, "last_step": self.last_step, "state": self.state, "structural_state": self.structural_state(), "mind_cell": self.mind_cell(root_id).to_dict(), "authority": { "authority_state": "computational_candidate_only", "proof_status": "not_proof", "validation_state": "unreviewed_computational_candidate", "canonical_approval": False, "semantic_relation_approved": False, "guardrail": CANDIDATE_GUARDRAIL, }, } @dataclass(frozen=True) class FormationRun: source: Mapping[str, Any] config: Mapping[str, Any] pattern_threads: tuple[PatternThread, ...] basins: tuple[Mapping[str, Any], ...] events: tuple[FormationEvent, ...] summary: Mapping[str, Any] run_digest: str def semantic_dict(self) -> dict[str, Any]: return { "kernel_version": KERNEL_VERSION, "source": dict(self.source), "config": dict(self.config), "pattern_threads": [thread.to_dict() for thread in self.pattern_threads], "basins": [dict(basin) for basin in self.basins], "events": [event.to_dict() for event in self.events], "summary": dict(self.summary), "guardrails": { "semantic_inference_performed": False, "semantic_promotion_performed": False, "identity_referent_established": False, "prediction_is_wisdom": False, "geometry_is_proof": False, "llm_used": False, "embedding_used": False, "category_gravity_used": False, "claim": CANDIDATE_GUARDRAIL, }, } def to_dict(self) -> dict[str, Any]: return {**self.semantic_dict(), "run_digest": self.run_digest} class FormationMindKernel: """Single-source, deterministic candidate-formation runtime.""" def __init__(self, source_bytes: bytes, *, source_relpath: str, config: FormationConfig): if source_relpath.replace("\\", "/") != config.source_relpath.replace("\\", "/"): raise ValueError("runtime source path must match the explicit FormationConfig source_relpath") self.source_bytes = bytes(source_bytes) self.source_sha256 = sha256_bytes(self.source_bytes) self.source_relpath = source_relpath.replace("\\", "/") self.config = config self.formation_byte_start, self.formation_byte_end = formation_content_span(self.source_bytes) self.formation_bytes = self.source_bytes[self.formation_byte_start : self.formation_byte_end] self.formation_sha256 = sha256_bytes(self.formation_bytes) occurrences = tokenize_formation_source(self.source_bytes) if config.maximum_tokens is not None: occurrences = occurrences[: config.maximum_tokens] self.occurrences = occurrences self.log = HashChainLog() self.fixed_step = 0 self.pattern_threads: list[PatternThread] = [] self.basins: dict[str, CandidateBasin] = {} self.pattern_index: dict[str, str] = {} self.relation_index: dict[tuple[str, str], str] = {} self.transition_counts: dict[str, dict[str, int]] = {} self.recent_pattern_basins: list[str] = [] self.root_id = f"BASIN::ROOT::{self.source_sha256[:24]}" self.basins[self.root_id] = CandidateBasin( identifier=self.root_id, basin_kind="formation_root", parent_id=None, depth=0, pattern_sha256=self.source_sha256, label=self.source_relpath, position=[0, 0, 0], velocity=[0, 0, 0], mass=0, support=0, first_step=0, last_step=0, state="structural_root", ) def _pattern_basin_id(self, normalized_pattern: str) -> str: digest = sha256_bytes(normalized_pattern.encode("utf-8")) return f"BASIN::PATTERN::{digest[:24]}" def _relation_basin_id(self, origin_id: str, target_id: str) -> str: digest = stable_digest({"origin": origin_id, "target": target_id}) return f"BASIN::RELATION::{digest[:24]}" def _thread_for(self, occurrence: SourceOccurrence, prior: SourceOccurrence | None) -> PatternThread: prior_address = prior.address if prior else None # Formation dynamics are seeded by the selected Markdown body, not by # release metadata in the excluded YAML front matter. The full-file # hash still remains authoritative in every source address and event # lineage, so a metadata-only release change produces a new snapshot # without silently perturbing the same semantic formation stream. prior_digest = prior.token_sha256 if prior else self.formation_sha256 context_fingerprint = _digest_seed(prior_digest) ^ occurrence.ordinal identifier = f"THREAD::{sha256_bytes(occurrence.address.encode('utf-8'))[:24]}" outside_refs = [f"formation_field:{self.root_id}"] if prior_address: outside_refs.append(prior_address) mind_cell = MindCellFrame( identifier=f"MC::{identifier}", reference_scope=f"source token occurrence {occurrence.ordinal} in {self.source_relpath}", inside=MindRoleFrame( "inside", (occurrence.address,), "the exact addressed source occurrence entering this local interaction", ), outside=MindRoleFrame( "outside", tuple(outside_refs), "the declared formation field and already-observed local predecessor", ), relation=MindRoleFrame( "relation", (f"ordered_contact:{occurrence.ordinal}",), "the deterministic arrival/contact operation; no meaning is assigned", ), ) return PatternThread(identifier, occurrence, prior_address, context_fingerprint, mind_cell) def _issue_prediction(self, origin_basin_id: str, target_ordinal: int) -> FormationEvent: origin = self.basins[origin_basin_id] geometry_candidates = [ basin for basin in self.basins.values() if basin.basin_kind == "pattern_candidate" and basin.identifier != origin_basin_id and basin.state == "active_candidate" ] if geometry_candidates: predicted_id, geometry_distance = min( [ ( basin.identifier, sum(abs(basin.position[axis] - origin.position[axis]) for axis in range(3)), ) for basin in geometry_candidates ], key=lambda item: (item[1], item[0]), ) state = "issued_candidate" else: predicted_id = None geometry_distance = None state = "abstained_insufficient_relational_positions" transition_candidates = self.transition_counts.get(origin_basin_id, {}) if transition_candidates: frequency_id, evidence_count = min( transition_candidates.items(), key=lambda item: (-item[1], item[0]), ) else: frequency_id = None evidence_count = 0 prediction_id = ( f"PRED::{self.source_sha256[:16]}::{target_ordinal:08d}::" f"{sha256_bytes(origin_basin_id.encode('utf-8'))[:12]}" ) return self.log.append( fixed_step=self.fixed_step, event_type="prediction_issued", payload={ "prediction_id": prediction_id, "origin_basin_id": origin_basin_id, "prediction_model_id": "relational_position_nearest_neighbor_v1", "prediction_model_role": "candidate_relational_geometry", "reference_frame": self.root_id, "distance_metric": "manhattan_integer", "predicted_target_basin_id": predicted_id, "predicted_distance": geometry_distance, "baseline_predictions": { "transition_frequency_v1": { "predicted_target_basin_id": frequency_id, "evidence_transition_count": evidence_count, }, "persistence_v1": { "predicted_target_basin_id": origin_basin_id, "evidence_transition_count": 1, }, }, "target_ordinal": target_ordinal, "causal_boundary": "issued_from_prior_state_before_target_observation", "semantic_claim": False, "wisdom_claim": False, "state": state, }, ) def _observe_pattern(self, thread: PatternThread) -> tuple[CandidateBasin, str]: occurrence = thread.occurrence pattern_digest = sha256_bytes(occurrence.normalized_pattern.encode("utf-8")) basin_id = self.pattern_index.get(occurrence.normalized_pattern) if basin_id is None: basin_id = self._pattern_basin_id(occurrence.normalized_pattern) basin = CandidateBasin( identifier=basin_id, basin_kind="pattern_candidate", parent_id=self.root_id, depth=1, pattern_sha256=pattern_digest, label=occurrence.normalized_pattern, position=_seeded_position(pattern_digest, self.config.coordinate_bound), velocity=[0, 0, 0], phase_tick=occurrence.ordinal % self.config.phase_modulus, mass=1, support=1, context_fingerprint=thread.context_fingerprint, member_refs=[occurrence.address], first_step=self.fixed_step, last_step=self.fixed_step, ) self.basins[basin_id] = basin self.pattern_index[occurrence.normalized_pattern] = basin_id self.basins[self.root_id].child_basin_ids.append(basin_id) action = "candidate_branch" self.log.append( fixed_step=self.fixed_step, event_type=action, payload={ "candidate_kind": "pattern_candidate", "basin_id": basin_id, "parent_basin_id": self.root_id, "source_occurrence_address": occurrence.address, "depth": 1, "reason": "first_exact_surface_pattern_occurrence", "candidate_only": True, }, ) else: basin = self.basins[basin_id] before = basin.structural_state() previous_context = basin.context_fingerprint residue = (previous_context ^ thread.context_fingerprint).bit_count() basin.member_refs.append(occurrence.address) basin.support += 1 basin.mass += 1 + residue // 16 basin.residue_total += residue basin.context_fingerprint = _rotate_left_64(previous_context, 1) ^ thread.context_fingerprint basin.phase_tick = (basin.phase_tick + 1 + residue) % self.config.phase_modulus basin.last_step = self.fixed_step action = "candidate_attach" attach_event = self.log.append( fixed_step=self.fixed_step, event_type=action, payload={ "candidate_kind": "pattern_candidate", "basin_id": basin_id, "source_occurrence_address": occurrence.address, "compatibility_rule": "exact_normalized_surface_pattern", "context_residue_bits": residue, "candidate_only": True, }, ) self.log.append( fixed_step=self.fixed_step, event_type="candidate_deform", payload={ "candidate_kind": "pattern_candidate", "basin_id": basin_id, "cause_event_hash": attach_event.event_hash, "before_state_digest": stable_digest(before), "after_state_digest": stable_digest(basin.structural_state()), "changed_integer_fields": [ "support", "mass", "residue_total", "context_fingerprint", "phase_tick", "member_count", "last_step", ], "candidate_only": True, }, ) return basin, action def _apply_fixed_step_attraction(self, basin: CandidateBasin) -> None: active_ids: list[str] = [] for candidate_id in reversed(self.recent_pattern_basins): if candidate_id != basin.identifier and candidate_id not in active_ids: active_ids.append(candidate_id) if len(active_ids) >= self.config.context_window: break before = basin.structural_state() force = [0, 0, 0] components: list[dict[str, Any]] = [] for attractor_id in active_ids: attractor = self.basins[attractor_id] compatibility = _surface_bit_compatibility(basin.label, attractor.label) transition_support = self.transition_counts.get(attractor_id, {}).get(basin.identifier, 0) structural_support = min(attractor.support, 32) raw_attraction = compatibility + structural_support * 4 + min(transition_support, 16) * 8 pull = min(self.config.maximum_pull, 1 + raw_attraction // self.config.force_divisor) axis_force = [] for axis in range(3): delta = attractor.position[axis] - basin.position[axis] contribution = _sign(delta) * pull force[axis] += contribution axis_force.append(contribution) components.append( { "attractor_basin_id": attractor_id, "surface_bit_compatibility_0_256": compatibility, "attractor_support": attractor.support, "prior_transition_support": transition_support, "pull": pull, "axis_force": axis_force, } ) for axis in range(3): accelerated = basin.velocity[axis] + force[axis] accelerated = max(-self.config.velocity_bound, min(self.config.velocity_bound, accelerated)) # Integer damping. Truncation toward zero is written explicitly. damped = _sign(accelerated) * (abs(accelerated) * 7 // 8) basin.velocity[axis] = damped moved = basin.position[axis] + damped basin.position[axis] = max(-self.config.coordinate_bound, min(self.config.coordinate_bound, moved)) basin.last_step = self.fixed_step after = basin.structural_state() self.log.append( fixed_step=self.fixed_step, event_type="candidate_deform", payload={ "candidate_kind": "pattern_candidate", "basin_id": basin.identifier, "deformation_kind": "integer_fixed_step_attraction", "reference_frame": self.root_id, "attractor_components": components, "net_force": force, "before_state_digest": stable_digest(before), "after_state_digest": stable_digest(after), "position_before": before["position"], "position_after": after["position"], "velocity_before": before["velocity"], "velocity_after": after["velocity"], "geometry_claim": False, "candidate_only": True, }, ) self.recent_pattern_basins.append(basin.identifier) if len(self.recent_pattern_basins) > self.config.context_window * 4: del self.recent_pattern_basins[: -self.config.context_window * 4] def _score_prediction( self, prediction_event: FormationEvent, actual_basin: CandidateBasin, occurrence: SourceOccurrence, ) -> None: model_predictions = { str(prediction_event.payload["prediction_model_id"]): prediction_event.payload[ "predicted_target_basin_id" ], **{ str(model_id): baseline["predicted_target_basin_id"] for model_id, baseline in prediction_event.payload["baseline_predictions"].items() }, } model_outcomes: dict[str, dict[str, Any]] = {} for model_id, predicted_id in model_predictions.items(): if predicted_id is None: outcome = "abstain" score_milli = 0 elif predicted_id == actual_basin.identifier: outcome = "hit" score_milli = 1000 else: outcome = "miss" score_milli = 0 model_outcomes[model_id] = { "predicted_target_basin_id": predicted_id, "outcome": outcome, "score_milli": score_milli, } primary = model_outcomes[str(prediction_event.payload["prediction_model_id"])] self.log.append( fixed_step=self.fixed_step, event_type="prediction_scored", payload={ "prediction_id": prediction_event.payload["prediction_id"], "prediction_event_hash": prediction_event.event_hash, "issued_sequence": prediction_event.sequence, "target_observed_at_sequence": next( event.sequence for event in reversed(self.log.events) if event.event_type == "pattern_observed" and event.payload["source_occurrence_address"] == occurrence.address ), "target_occurrence_address": occurrence.address, "prediction_model_id": prediction_event.payload["prediction_model_id"], "prediction_model_role": prediction_event.payload["prediction_model_role"], "predicted_target_basin_id": primary["predicted_target_basin_id"], "actual_target_basin_id": actual_basin.identifier, "outcome": primary["outcome"], "score_milli": primary["score_milli"], "model_outcomes": model_outcomes, "semantic_claim": False, "wisdom_claim": False, }, ) def _record_transition(self, origin: CandidateBasin, target: CandidateBasin, occurrence: SourceOccurrence) -> None: pair = (origin.identifier, target.identifier) transition_map = self.transition_counts.setdefault(origin.identifier, {}) transition_map[target.identifier] = transition_map.get(target.identifier, 0) + 1 relation_id = self.relation_index.get(pair) relation_ref = f"contact:{occurrence.ordinal - 1}:{occurrence.ordinal}" if relation_id is None: relation_id = self._relation_basin_id(*pair) digest = stable_digest({"ordered_pair": list(pair)}) midpoint = [ (origin.position[axis] + target.position[axis]) // 2 for axis in range(3) ] relation = CandidateBasin( identifier=relation_id, basin_kind="relation_candidate", parent_id=origin.identifier, depth=2, pattern_sha256=digest, label=f"{origin.identifier} -> {target.identifier}", position=midpoint, velocity=[0, 0, 0], mass=1, support=1, member_refs=[relation_ref], relation_refs=[relation_ref], first_step=self.fixed_step, last_step=self.fixed_step, ) self.basins[relation_id] = relation self.relation_index[pair] = relation_id origin.child_basin_ids.append(relation_id) self.log.append( fixed_step=self.fixed_step, event_type="candidate_branch", payload={ "candidate_kind": "relation_candidate", "basin_id": relation_id, "parent_basin_id": origin.identifier, "origin_basin_id": origin.identifier, "target_basin_id": target.identifier, "contact_ref": relation_ref, "depth": 2, "reason": "first_observed_ordered_contact_after_prediction_scoring", "semantic_relation_asserted": False, "candidate_only": True, }, ) else: relation = self.basins[relation_id] before = relation.structural_state() relation.member_refs.append(relation_ref) relation.relation_refs.append(relation_ref) relation.support += 1 relation.mass += 1 relation.phase_tick = (relation.phase_tick + 1) % self.config.phase_modulus relation.last_step = self.fixed_step attach_event = self.log.append( fixed_step=self.fixed_step, event_type="candidate_attach", payload={ "candidate_kind": "relation_candidate", "basin_id": relation_id, "origin_basin_id": origin.identifier, "target_basin_id": target.identifier, "contact_ref": relation_ref, "compatibility_rule": "same_ordered_pattern_basin_pair", "semantic_relation_asserted": False, "candidate_only": True, }, ) self.log.append( fixed_step=self.fixed_step, event_type="candidate_deform", payload={ "candidate_kind": "relation_candidate", "basin_id": relation_id, "cause_event_hash": attach_event.event_hash, "before_state_digest": stable_digest(before), "after_state_digest": stable_digest(relation.structural_state()), "changed_integer_fields": ["support", "mass", "phase_tick", "member_count", "relation_count", "last_step"], "semantic_relation_asserted": False, "candidate_only": True, }, ) def _retire_weak_candidates(self) -> None: candidates = sorted( (basin for basin in self.basins.values() if basin.basin_kind != "formation_root"), key=lambda basin: (-basin.depth, basin.identifier), ) for basin in candidates: if len(basin.member_refs) >= self.config.minimum_persistence_occurrences: continue basin.state = "retired_candidate" basin.last_step = self.fixed_step self.log.append( fixed_step=self.fixed_step, event_type="candidate_retire", payload={ "candidate_kind": basin.basin_kind, "basin_id": basin.identifier, "member_count": len(basin.member_refs), "minimum_persistence_occurrences": self.config.minimum_persistence_occurrences, "reason": "insufficient_recurrence_in_bounded_source_run", "deleted": False, "candidate_only": True, }, ) def run(self) -> FormationRun: self.log.append( fixed_step=0, event_type="formation_started", payload={ "kernel_version": KERNEL_VERSION, "source_relpath": self.source_relpath, "source_sha256": self.source_sha256, "source_byte_count": len(self.source_bytes), "formation_byte_start": self.formation_byte_start, "formation_byte_end": self.formation_byte_end, "formation_content_sha256": sha256_bytes(self.formation_bytes), "configured_token_count": len(self.occurrences), "one_source_only": True, "candidate_only": True, }, ) previous_occurrence: SourceOccurrence | None = None previous_basin: CandidateBasin | None = None for occurrence in self.occurrences: self.fixed_step += 1 prediction_event = ( self._issue_prediction(previous_basin.identifier, occurrence.ordinal) if previous_basin is not None else None ) thread = self._thread_for(occurrence, previous_occurrence) self.pattern_threads.append(thread) self.log.append( fixed_step=self.fixed_step, event_type="pattern_observed", payload={ "pattern_thread_id": thread.identifier, "source_occurrence_address": occurrence.address, "token_ordinal": occurrence.ordinal, "mind_cell_id": thread.mind_cell.identifier, }, ) basin, _ = self._observe_pattern(thread) self._apply_fixed_step_attraction(basin) if prediction_event is not None and previous_basin is not None: self._score_prediction(prediction_event, basin, occurrence) # Transition learning occurs only after target observation and score. self._record_transition(previous_basin, basin, occurrence) previous_occurrence = occurrence previous_basin = basin self.fixed_step += 1 self._retire_weak_candidates() prediction_events = [event for event in self.log.events if event.event_type == "prediction_issued"] scored_events = [event for event in self.log.events if event.event_type == "prediction_scored"] outcomes = {"hit": 0, "miss": 0, "abstain": 0} model_outcomes = { "relational_position_nearest_neighbor_v1": {"hit": 0, "miss": 0, "abstain": 0}, "transition_frequency_v1": {"hit": 0, "miss": 0, "abstain": 0}, "persistence_v1": {"hit": 0, "miss": 0, "abstain": 0}, } for event in scored_events: outcomes[str(event.payload["outcome"])] += 1 for model_id, result in event.payload["model_outcomes"].items(): model_outcomes[model_id][str(result["outcome"])] += 1 model_scores = { model_id: { **counts, "issued_score_rate_milli": ( counts["hit"] * 1000 // max(1, counts["hit"] + counts["miss"]) ), "all_target_score_rate_milli": counts["hit"] * 1000 // max(1, len(scored_events)), } for model_id, counts in model_outcomes.items() } ordered_basins = tuple( basin.to_dict(self.root_id) for basin in sorted(self.basins.values(), key=lambda item: (item.depth, item.identifier)) ) summary = { "fixed_steps": self.fixed_step, "pattern_thread_count": len(self.pattern_threads), "basin_count": len(ordered_basins), "pattern_candidate_count": sum(basin["basin_kind"] == "pattern_candidate" for basin in ordered_basins), "relation_candidate_count": sum(basin["basin_kind"] == "relation_candidate" for basin in ordered_basins), "maximum_recursive_depth": max((int(basin["depth"]) for basin in ordered_basins), default=0), "active_candidate_count": sum(basin["state"] == "active_candidate" for basin in ordered_basins), "retired_candidate_count": sum(basin["state"] == "retired_candidate" for basin in ordered_basins), "event_count": len(self.log.events), "event_chain_head": self.log.events[-1].event_hash, "event_chain_verified": verify_event_chain(self.log.events), "predictions_issued": len(prediction_events), "predictions_scored": len(scored_events), "prediction_outcomes": outcomes, "prediction_model_scores": model_scores, "relational_geometry_predictive_lift_evaluated": True, "predictive_wisdom_earned": False, "prediction_protocol": "issue_from_prior_state_then_observe_then_score_then_learn_transition", } source = { "relative_path": self.source_relpath, "content_address": f"sha256:{self.source_sha256}", "sha256": self.source_sha256, "byte_count": len(self.source_bytes), "token_count": len(self.occurrences), "encoding": "utf-8", "tokenizer_version": TOKENIZER_VERSION, "scope": "one_canonical_chapter_only", "formation_content": { "selection_rule": "markdown_body_excluding_opening_yaml_front_matter", "byte_start": self.formation_byte_start, "byte_end": self.formation_byte_end, "byte_count": len(self.formation_bytes), "sha256": self.formation_sha256, }, } partial = FormationRun( source=source, config=self.config.to_dict(), pattern_threads=tuple(self.pattern_threads), basins=ordered_basins, events=self.log.events, summary=summary, run_digest="", ) semantic = partial.semantic_dict() _assert_integer_tree(semantic) return FormationRun( source=source, config=self.config.to_dict(), pattern_threads=tuple(self.pattern_threads), basins=ordered_basins, events=self.log.events, summary=summary, run_digest=stable_digest(semantic), ) def run_configured_source(root: Path = ROOT, config: FormationConfig | None = None) -> FormationRun: """Run exactly the canonical file named in ``config``.""" selected = config or FormationConfig() root_resolved = root.resolve() source_path = (root_resolved / Path(selected.source_relpath)).resolve() canonical_root = (root_resolved / "canonical").resolve() if canonical_root not in source_path.parents: raise ValueError("configured source must remain inside the canonical directory") if not source_path.is_file(): raise FileNotFoundError(source_path) kernel = FormationMindKernel( source_path.read_bytes(), source_relpath=selected.source_relpath, config=selected, ) return kernel.run() def _write_output(path: Path, value: Mapping[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", newline="\n") def main(argv: Iterable[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--source", default=DEFAULT_SOURCE_RELPATH, help="one canonical Markdown path relative to the living-source root", ) parser.add_argument( "--output", type=Path, help="optional JSON output path; omitted runs verification and prints a compact summary", ) parser.add_argument( "--maximum-tokens", type=int, default=None, help="optional deterministic prefix bound for a diagnostic run", ) args = parser.parse_args(list(argv) if argv is not None else None) config = FormationConfig( source_relpath=args.source.replace("\\", "/"), maximum_tokens=args.maximum_tokens, ) run = run_configured_source(ROOT, config) if args.output: _write_output(args.output, run.to_dict()) print( json.dumps( { "kernel_version": KERNEL_VERSION, "source": run.source, "summary": run.summary, "run_digest": run.run_digest, "output": str(args.output.resolve()) if args.output else None, }, ensure_ascii=False, indent=2, ) ) return 0 if __name__ == "__main__": raise SystemExit(main())