diff --git a/docs/migration/import-ledger.md b/docs/migration/import-ledger.md index ff4e866..3db5773 100644 --- a/docs/migration/import-ledger.md +++ b/docs/migration/import-ledger.md @@ -52,7 +52,7 @@ Empty placeholder directories (no tracked source): `packages/ts-*`, top-level `e | Repository | Verified source SHA in this clone? | Destination claimed historically | Status | Notes | |------------|-------------------------------------|----------------------------------|--------|-------| | TS-Core | `3ef48ad00efef8659ad0981d71de509a9827f584` | `src/thinking_system/core/typed_tension/` | **PARTIALLY_IMPORTED** | Domain-neutral typed tension kernel imported with its focused test and example; legacy Python application layers, Rust acceleration, CLI, Z3, Grok, UI, and Kernel Wave 12 surfaces remain unimported | -| cig-ts-engine | Not verified here | `packages/ts-graph` | **INSPECTED** / **PLANNED** | Graph lives in `core/graph` | +| cig-ts-engine | `71d91a6a0cdc9e7a5439df972130596f0bae5d2f` | `src/thinking_system/graph/cig/` | **PARTIALLY_IMPORTED** | Core graph models, deterministic propagation, edge tension, derivative observer, representational radius, context-split proposal, YAML IO, seed graph and focused tests imported; CLI, plotting, reports, Obsidian export and generated artifacts remain outside | | TS-Benchmarks | Not verified here | `benchmarks/` | **PLANNED** | `benchmarks/suites` empty | | ts-chat-language | Not verified here | `packages/ts-language` | **INSPECTED** / **PLANNED** | TSLC-like code in `core/language/tslc.py` is in-tree monorepo code; full satellite port not evidenced | | TensionForge | Not verified here | `engines/tension-forge` | **PLANNED** (satellite); in-tree `inference/tension_forge/` is separate | Empty `engines/tension-forge/` | diff --git a/docs/migration/sources/cig-ts-engine.md b/docs/migration/sources/cig-ts-engine.md new file mode 100644 index 0000000..a2d9b0f --- /dev/null +++ b/docs/migration/sources/cig-ts-engine.md @@ -0,0 +1,68 @@ +# cig-ts-engine consolidation record + +## Source + +- Repository: `BoggersTheFish/cig-ts-engine` +- Source branch: `master` +- Source commit: `71d91a6a0cdc9e7a5439df972130596f0bae5d2f` +- Licence: MIT + +## Imported surface + +The deterministic, domain-neutral CIG mechanics were imported into: + +`src/thinking_system/graph/cig/` + +Imported functionality: + +- validated node and edge models +- mutable deterministic CIG graph state +- deterministic activation propagation and relaxation +- edge-local and total tension measurement +- finite-difference derivative-meaning observer +- representational-radius measurement +- overloaded-node localisation +- non-mutating context-split proposals +- tension-reduction versus complexity-cost acceptance rule +- context-split application +- YAML graph loading and saving +- the hand-authored demonstration seed graph +- focused tests for the imported surface + +## Adaptation + +Imports were changed from `cig` to +`thinking_system.graph.cig`. + +The demonstration graph was renamed from: + +`examples/ts_core.yaml` + +to: + +`examples/cig_ts_core.yaml` + +The implementation remains an experimental graph family alongside the existing +Thinking System universal graph. It has not replaced that graph implementation. + +## Deliberately not imported + +- Typer CLI +- matplotlib visualisation +- proof-report generation +- Obsidian export +- generated PNG files +- generated proof-bank documents +- generated Obsidian vault contents +- symbolic or biographical demonstration scripts + +Those surfaces are application, presentation, or generated-output layers rather +than required graph mechanics. + +## Current status + +**PARTIALLY_IMPORTED** + +The core experimental graph mechanics are present and tested. The satellite +must remain unarchived until its excluded surfaces are explicitly classified as +obsolete, historical, independently maintained, or worth importing. diff --git a/examples/cig_ts_core.yaml b/examples/cig_ts_core.yaml new file mode 100644 index 0000000..7457e3a --- /dev/null +++ b/examples/cig_ts_core.yaml @@ -0,0 +1,429 @@ +nodes: + - id: meaning + label: Meaning + activation: 0.0 + stability: 1.0 + metadata: + type: concept + - id: life + label: Life + activation: 0.0 + stability: 1.0 + metadata: + type: concept + - id: number_47 + label: Number 47 + activation: 0.0 + stability: 1.0 + metadata: + type: symbol + - id: arbitrary_symbol + label: Arbitrary Symbol + activation: 0.0 + stability: 1.0 + metadata: + type: symbol + - id: stable_attractor + label: Stable Attractor + activation: 0.0 + stability: 1.0 + metadata: + type: state + - id: coherence + label: Coherence + activation: 0.0 + stability: 1.0 + metadata: + type: state + - id: humour + label: Humour + activation: 0.0 + stability: 1.0 + metadata: + type: concept + - id: absurdity + label: Absurdity + activation: 0.0 + stability: 1.0 + metadata: + type: concept + - id: religion + label: Religion + activation: 0.0 + stability: 1.0 + metadata: + type: concept + - id: ritual + label: Ritual + activation: 0.0 + stability: 1.0 + metadata: + type: memory + - id: comfort + label: Comfort + activation: 0.0 + stability: 1.0 + metadata: + type: state + - id: trauma + label: Trauma + activation: 0.0 + stability: 1.0 + metadata: + type: memory + - id: control + label: Control + activation: 0.0 + stability: 1.0 + metadata: + type: concept + - id: community + label: Community + activation: 0.0 + stability: 1.0 + metadata: + type: concept + - id: language + label: Language + activation: 0.0 + stability: 1.0 + metadata: + type: concept + - id: symbol + label: Symbol + activation: 0.0 + stability: 1.0 + metadata: + type: symbol + - id: signal + label: Signal + activation: 0.0 + stability: 1.0 + metadata: + type: state + - id: wave + label: Wave + activation: 0.0 + stability: 1.0 + metadata: + type: concept + - id: propagation + label: Propagation + activation: 0.0 + stability: 1.0 + metadata: + type: transformation + - id: graph + label: Graph + activation: 0.0 + stability: 1.0 + metadata: + type: structure + - id: edge_strength + label: Edge Strength + activation: 0.0 + stability: 1.0 + metadata: + type: constraint + - id: derivative_response + label: Derivative Response + activation: 0.0 + stability: 1.0 + metadata: + type: state + - id: tension + label: Tension + activation: 0.0 + stability: 1.0 + metadata: + type: state + - id: relaxation + label: Relaxation + activation: 0.0 + stability: 1.0 + metadata: + type: transformation + - id: compression + label: Compression + activation: 0.0 + stability: 1.0 + metadata: + type: concept + - id: R_radius + label: R Radius + activation: 0.0 + stability: 1.0 + metadata: + type: metric + - id: minimum_R_compressor + label: Minimum-R Compressor + activation: 0.0 + stability: 1.0 + metadata: + type: objective + +edges: + - source: number_47 + target: arbitrary_symbol + relation: exemplifies + weight: 0.95 + polarity: 1.0 + expected_ratio: 0.9 + metadata: + note: "47 can act as a deliberately arbitrary symbol." + - source: arbitrary_symbol + target: stable_attractor + relation: can_settle_into + weight: 0.45 + polarity: 1.0 + expected_ratio: 0.5 + metadata: + note: "Repeated arbitrary symbols can become stabilizing attractors." + - source: arbitrary_symbol + target: absurdity + relation: evokes + weight: 0.8 + polarity: 1.0 + expected_ratio: 0.7 + metadata: + note: "Arbitrary symbols often activate absurdity." + - source: absurdity + target: humour + relation: evokes + weight: 0.75 + polarity: 1.0 + expected_ratio: 0.8 + metadata: + note: "Absurdity can become humorous when tension remains low." + - source: stable_attractor + target: coherence + relation: stabilizes + weight: 0.9 + polarity: 1.0 + expected_ratio: 0.85 + metadata: + note: "Stable attractors reduce drift and support coherence." + - source: meaning + target: stable_attractor + relation: seeks + weight: 0.9 + polarity: 1.0 + expected_ratio: 0.8 + metadata: + note: "Meaning often seeks stabilizing attractors." + - source: life + target: meaning + relation: raises_question_of + weight: 0.85 + polarity: 1.0 + expected_ratio: 0.75 + metadata: + note: "Life commonly activates meaning-seeking." + - source: religion + target: ritual + relation: organizes + weight: 0.85 + polarity: 1.0 + expected_ratio: 0.8 + metadata: + note: "Religion can organize repeated ritual forms." + - source: ritual + target: comfort + relation: supports + weight: 0.7 + polarity: 1.0 + expected_ratio: 0.65 + metadata: + note: "Ritual repetition may support comfort." + - source: religion + target: comfort + relation: can_support + weight: 0.6 + polarity: 1.0 + expected_ratio: 0.55 + metadata: + note: "Religion can directly support comfort for some contexts." + - source: religion + target: control + relation: can_enable + weight: 0.85 + polarity: 1.0 + expected_ratio: 0.4 + metadata: + note: "Religion can also enable social control." + - source: religion + target: trauma + relation: can_trigger + weight: 0.05 + polarity: 1.0 + expected_ratio: 0.35 + metadata: + note: "Religion can activate trauma associations in harm contexts." + - source: religion + target: tension + relation: can_raise + weight: 0.05 + polarity: 1.0 + expected_ratio: 0.45 + metadata: + note: "Religion can raise tension in harm contexts." + - source: trauma + target: tension + relation: increases + weight: 0.1 + polarity: 1.0 + expected_ratio: 0.9 + metadata: + note: "Trauma tends to increase unresolved tension." + - source: control + target: tension + relation: increases + weight: 0.85 + polarity: 1.0 + expected_ratio: 0.6 + metadata: + note: "Control can increase tension when constraints conflict." + - source: language + target: symbol + relation: uses + weight: 0.9 + polarity: 1.0 + expected_ratio: 0.85 + metadata: + note: "Language uses symbols as signal carriers." + - source: symbol + target: signal + relation: carries + weight: 0.85 + polarity: 1.0 + expected_ratio: 0.8 + metadata: + note: "Symbols carry signals through interpretation." + - source: signal + target: wave + relation: can_be + weight: 0.65 + polarity: 1.0 + expected_ratio: 0.6 + metadata: + note: "Signals may be represented as waves." + - source: wave + target: propagation + relation: enables + weight: 0.7 + polarity: 1.0 + expected_ratio: 0.65 + metadata: + note: "Waves provide a concrete propagation metaphor." + - source: graph + target: edge_strength + relation: parameterizes + weight: 0.8 + polarity: 1.0 + expected_ratio: 0.75 + metadata: + note: "Graphs expose edge strength as a reasoning parameter." + - source: edge_strength + target: derivative_response + relation: shapes + weight: 0.75 + polarity: 1.0 + expected_ratio: 0.7 + metadata: + note: "Edge strength shapes derivative response." + - source: compression + target: R_radius + relation: minimizes + weight: 0.8 + polarity: -1.0 + expected_ratio: 0.65 + metadata: + note: "Better compression should reduce representational radius." + - source: R_radius + target: minimum_R_compressor + relation: defines_objective_for + weight: 0.9 + polarity: -1.0 + expected_ratio: 0.8 + metadata: + note: "Lower R supports the minimum-R compressor objective." + - source: religion + target: community + relation: can_build + weight: 0.75 + polarity: 1.0 + expected_ratio: 0.65 + metadata: + note: "Religion can build community in comfort-oriented contexts." + - source: comfort + target: stable_attractor + relation: stabilizes + weight: 0.55 + polarity: 1.0 + expected_ratio: 0.6 + metadata: + note: "Comfort can help stabilize attractor states." + - source: comfort + target: relaxation + relation: supports + weight: 0.55 + polarity: 1.0 + expected_ratio: 0.55 + metadata: + note: "Comfort supports relaxation." + - source: relaxation + target: coherence + relation: supports + weight: 0.5 + polarity: 1.0 + expected_ratio: 0.55 + metadata: + note: "Relaxation can support low-tension coherence." + - source: community + target: comfort + relation: supports + weight: 0.35 + polarity: 1.0 + expected_ratio: 0.45 + metadata: + note: "Community can reinforce comfort." + - source: comfort + target: control + relation: softens + weight: 1.0 + polarity: -1.0 + expected_ratio: 0.3 + metadata: + note: "Comfort context softens the control path." + - source: comfort + target: tension + relation: relaxes + weight: 0.8 + polarity: -1.0 + expected_ratio: 0.3 + metadata: + note: "Comfort context relaxes tension." + - source: trauma + target: comfort + relation: disrupts + weight: 1.3 + polarity: -1.0 + expected_ratio: 0.3 + metadata: + note: "Trauma context disrupts comfort response." + - source: trauma + target: ritual + relation: disrupts + weight: 1.0 + polarity: -1.0 + expected_ratio: 0.3 + metadata: + note: "Trauma context disrupts ritual response." + - source: trauma + target: community + relation: isolates + weight: 1.0 + polarity: -1.0 + expected_ratio: 0.3 + metadata: + note: "Trauma context can isolate community response." diff --git a/pyproject.toml b/pyproject.toml index 3f7daa7..506b85a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,6 +32,7 @@ dependencies = [ "pyyaml", "jsonschema", "numpy", + "pydantic>=2", ] [project.optional-dependencies] diff --git a/src/thinking_system/graph/cig/__init__.py b/src/thinking_system/graph/cig/__init__.py new file mode 100644 index 0000000..585751b --- /dev/null +++ b/src/thinking_system/graph/cig/__init__.py @@ -0,0 +1,8 @@ +"""Experimental deterministic Concept/Constraint Intelligence Graph engine.""" + +from thinking_system.graph.cig.edge import Edge +from thinking_system.graph.cig.engine import CIGEngine, ThinkingSystemEngine +from thinking_system.graph.cig.graph import CIGraph, Graph +from thinking_system.graph.cig.node import Node + +__all__ = ["CIGEngine", "CIGraph", "Edge", "Graph", "Node", "ThinkingSystemEngine"] diff --git a/src/thinking_system/graph/cig/compression.py b/src/thinking_system/graph/cig/compression.py new file mode 100644 index 0000000..75dc6bd --- /dev/null +++ b/src/thinking_system/graph/cig/compression.py @@ -0,0 +1,19 @@ +from __future__ import annotations + +from thinking_system.graph.cig.graph import Graph + + +def representational_radius(graph: Graph, beta: float = 0.01) -> float: + """Compute first-pass representational radius R. + + R = sum primitive_cost for primitive nodes + beta * edge_count. + """ + if beta < 0.0: + raise ValueError("beta must be non-negative") + + primitive_cost = 0.0 + for node in graph.nodes.values(): + metadata = node.metadata + if metadata.get("primitive") is True or "primitive_cost" in metadata: + primitive_cost += float(metadata.get("primitive_cost", 0.0)) + return float(primitive_cost + beta * len(graph.edges)) diff --git a/src/thinking_system/graph/cig/edge.py b/src/thinking_system/graph/cig/edge.py new file mode 100644 index 0000000..284cd49 --- /dev/null +++ b/src/thinking_system/graph/cig/edge.py @@ -0,0 +1,17 @@ +from __future__ import annotations + +from pydantic import BaseModel, ConfigDict, Field + + +class Edge(BaseModel): + """A weighted relation or constraint between two CIG nodes.""" + + model_config = ConfigDict(extra="allow") + + source: str + target: str + relation: str + weight: float = Field(default=1.0, ge=0.0) + polarity: float = Field(default=1.0, ge=-1.0, le=1.0) + expected_ratio: float = 1.0 + metadata: dict = Field(default_factory=dict) diff --git a/src/thinking_system/graph/cig/engine.py b/src/thinking_system/graph/cig/engine.py new file mode 100644 index 0000000..9fa9072 --- /dev/null +++ b/src/thinking_system/graph/cig/engine.py @@ -0,0 +1,166 @@ +from __future__ import annotations + +import numpy as np + +from thinking_system.graph.cig.graph import Graph +from thinking_system.graph.cig.tension import ( + TensionReport, + detect_tension, + tension_report, +) + + +class CIGEngine: + """Deterministic TS cycle engine over an explicit CIGraph.""" + + def __init__( + self, + graph: Graph, + decay: float = 0.85, + activation_rate: float = 0.35, + clamp: bool = True, + ) -> None: + self.graph = graph + self.decay = decay + self.activation_rate = activation_rate + self.clamp = clamp + self._source_amounts: dict[str, float] = {} + + def activate_inputs(self, input_node_ids: list[str], amount: float = 1.0) -> None: + for node_id in input_node_ids: + self.graph.set_activation(node_id, self._clamp(amount)) + + def propagate_step(self) -> None: + """Run one deterministic propagation update. + + a_next = decay * a + activation_rate * incoming + """ + incoming = {node_id: 0.0 for node_id in self.graph.nodes} + for edge in self.graph.edges: + source_activation = self._source_activation(edge.source) + incoming[edge.target] += ( + source_activation * edge.weight * edge.polarity * self.activation_rate + ) + + for node_id, node in self.graph.nodes.items(): + next_activation = self.decay * node.activation + incoming[node_id] + node.activation = self._clamp(next_activation) + + def relax_step(self) -> None: + """Decay all activations deterministically.""" + for node in self.graph.nodes.values(): + node.activation = self._clamp(node.activation * self.decay) + + def run_cycle( + self, + input_node_ids: list[str], + steps: int = 5, + input_amounts: dict[str, float] | None = None, + hold_inputs: bool = True, + ) -> dict: + if steps < 0: + raise ValueError("steps must be non-negative") + + initial_activations = self._activation_snapshot() + self._set_cycle_inputs(input_node_ids, input_amounts) + tension_before_report = tension_report(self.graph) + + for _ in range(steps): + self.propagate_step() + if hold_inputs: + self._set_cycle_inputs(input_node_ids, input_amounts) + else: + self._source_amounts = {} + + final_activations = self._activation_snapshot() + tension_after_report = tension_report(self.graph) + self._source_amounts = {} + return { + "input_nodes": list(input_node_ids), + "initial_activations": initial_activations, + "final_activations": final_activations, + "top_activated_nodes": self._top_activated_nodes(), + "tension_before": tension_before_report["total"], + "tension_after": tension_after_report["total"], + "top_tension_edges_before": tension_before_report["top_edges"], + "top_tension_edges_after": tension_after_report["top_edges"], + "steps": steps, + } + + def propagate(self) -> None: + """Compatibility alias for older placeholder engine API.""" + self.propagate_step() + + def relax(self) -> None: + """Compatibility alias for older placeholder engine API.""" + self.relax_step() + + def detect_tension(self) -> TensionReport: + return detect_tension(self.graph) + + def evolve(self, report: TensionReport) -> None: + """Placeholder for Break/Evolve restructuring.""" + # TODO: Split nodes, add constraints, or create new abstractions when + # unresolved tension justifies added graph complexity. + _ = report + + def step(self) -> TensionReport: + self.propagate_step() + report = self.detect_tension() + self.evolve(report) + return report + + def _activation_snapshot(self) -> dict[str, float]: + return {node_id: node.activation for node_id, node in self.graph.nodes.items()} + + def _top_activated_nodes(self) -> list[dict[str, float | str]]: + return [ + { + "id": node_id, + "label": node.label, + "activation": node.activation, + } + for node_id, node in sorted( + self.graph.nodes.items(), + key=lambda item: (-item[1].activation, item[0]), + ) + if node.activation > 0.0 + ] + + def _clamp(self, value: float) -> float: + if not self.clamp: + return float(value) + return float(np.clip(value, 0.0, 1.0)) + + def _set_cycle_inputs( + self, + input_node_ids: list[str], + input_amounts: dict[str, float] | None, + ) -> None: + self._source_amounts = {} + for node_id in input_node_ids: + amount = 1.0 if input_amounts is None else input_amounts.get(node_id, 1.0) + self.graph.set_activation(node_id, self._clamp(amount)) + self._source_amounts[node_id] = float(amount) + + def _source_activation(self, node_id: str) -> float: + return self._source_amounts.get( + node_id, self.graph.get_node(node_id).activation + ) + + +class ThinkingSystemEngine(CIGEngine): + """Compatibility wrapper for the initial skeleton API.""" + + def __init__( + self, + graph: Graph, + propagation_rate: float = 1.0, + relaxation_rate: float = 0.05, + ) -> None: + super().__init__( + graph=graph, + decay=1.0 - relaxation_rate, + activation_rate=propagation_rate, + clamp=True, + ) diff --git a/src/thinking_system/graph/cig/evolve.py b/src/thinking_system/graph/cig/evolve.py new file mode 100644 index 0000000..19ab4ef --- /dev/null +++ b/src/thinking_system/graph/cig/evolve.py @@ -0,0 +1,281 @@ +from __future__ import annotations + +from collections import defaultdict +from copy import deepcopy + +from thinking_system.graph.cig.compression import representational_radius +from thinking_system.graph.cig.edge import Edge +from thinking_system.graph.cig.graph import CIGraph, Graph +from thinking_system.graph.cig.node import Node +from thinking_system.graph.cig.tension import TensionReport, edge_tension, total_tension + +DEFAULT_SPLIT_PRIMITIVE_COST = 0.5 + + +def find_overloaded_nodes(graph: Graph, top_k: int = 5) -> list[dict]: + """Find nodes whose outgoing constraints suggest multiple contexts. + + This is a deterministic heuristic, not an automatic graph edit. A node is + treated as overloaded when it has multiple outgoing edges and either those + edges create tension, vary in tension, or point into divergent target + metadata groups. + """ + if top_k < 0: + raise ValueError("top_k must be non-negative") + + reports: list[dict] = [] + for node_id, node in graph.nodes.items(): + outgoing = graph.outgoing_edges(node_id) + if len(outgoing) < 2: + continue + + edge_reports = [_edge_report(graph, edge) for edge in outgoing] + tensions = [item["tension"] for item in edge_reports] + target_groups = sorted( + {_semantic_group(graph.get_node(edge.target)) for edge in outgoing} + ) + total = sum(tensions) + variation = max(tensions) - min(tensions) if tensions else 0.0 + divergent = len(target_groups) > 1 + overloaded = total > 0.0 or variation > 0.0 or divergent + if not overloaded: + continue + + score = total + variation + max(0, len(target_groups) - 1) * 0.1 + reports.append( + { + "node_id": node_id, + "label": node.label, + "outgoing_count": len(outgoing), + "outgoing_tension": float(total), + "tension_variation": float(variation), + "target_groups": target_groups, + "score": float(score), + "edges": sorted( + edge_reports, + key=lambda item: (-item["tension"], item["target"]), + ), + "explanation": ( + f"{node_id} has {len(outgoing)} outgoing edges across " + f"{len(target_groups)} target groups." + ), + } + ) + + reports.sort(key=lambda item: (-item["score"], item["node_id"])) + return reports[:top_k] + + +def suggest_context_split( + graph: Graph, + node_id: str, + split_names: list[str], + alpha: float = 1.0, + beta: float = 0.01, +) -> dict: + """Return a non-mutating context split proposal for an overloaded node.""" + if node_id not in graph.nodes: + raise KeyError(node_id) + if not split_names: + raise ValueError("split_names must not be empty") + + original = graph.get_node(node_id) + new_nodes = [ + { + "id": split_name, + "label": _split_label(original.label, split_name), + "metadata": { + **original.metadata, + "primitive": True, + "primitive_cost": DEFAULT_SPLIT_PRIMITIVE_COST, + "split_from": node_id, + "split_role": split_name, + }, + } + for split_name in split_names + ] + + outgoing = graph.outgoing_edges(node_id) + incoming = graph.incoming_edges(node_id) + edges_to_redirect = [ + { + "source": edge.source, + "target": edge.target, + "relation": edge.relation, + "suggested_new_source": _choose_split_name( + graph.get_node(edge.target), split_names + ), + "tension": edge_tension(graph, edge), + } + for edge in outgoing + ] + edges_to_copy = [ + { + "source": edge.source, + "target": edge.target, + "relation": edge.relation, + "copy_to_targets": list(split_names), + } + for edge in incoming + ] + + mapping = { + item["target"]: item["suggested_new_source"] for item in edges_to_redirect + } + proposed = apply_context_split(graph, node_id, mapping) + delta_r = representational_radius(proposed, beta=beta) - representational_radius( + graph, beta=beta + ) + tension_reduction = total_tension(graph) - total_tension(proposed) + + return { + "original_node": { + "id": original.id, + "label": original.label, + "metadata": deepcopy(original.metadata), + }, + "new_nodes": new_nodes, + "edges_to_redirect": edges_to_redirect, + "edges_to_copy": edges_to_copy, + "delta_R": float(delta_r), + "expected_complexity_increase_delta_R": float(delta_r), + "tension_reduction": float(tension_reduction), + "alpha": float(alpha), + "accepted": bool(tension_reduction > alpha * delta_r), + "explanation": ( + f"Split {node_id} into {', '.join(split_names)} so outgoing " + "constraints can be assigned to context-specific source states." + ), + } + + +def apply_context_split( + graph: Graph, + node_id: str, + mapping: dict[str, str], +) -> CIGraph: + """Return a copied graph with selected outgoing edges redirected. + + `mapping` maps target node id -> new source node id. The original graph is + not mutated. + """ + if node_id not in graph.nodes: + raise KeyError(node_id) + if not mapping: + raise ValueError("mapping must not be empty") + + result = graph.copy() + original = result.get_node(node_id) + new_source_ids = sorted(set(mapping.values())) + + for new_source_id in new_source_ids: + if new_source_id not in result.nodes: + result.add_node( + Node( + id=new_source_id, + label=_split_label(original.label, new_source_id), + activation=original.activation, + stability=original.stability, + metadata={ + **deepcopy(original.metadata), + "primitive": True, + "primitive_cost": DEFAULT_SPLIT_PRIMITIVE_COST, + "split_from": node_id, + }, + ) + ) + + result.edges = [ + ( + _redirect_edge(edge, mapping[edge.target]) + if edge.source == node_id and edge.target in mapping + else edge + ) + for edge in result.edges + ] + _fit_split_source_activations(result, mapping) + return result + + +def break_or_evolve(graph: Graph, report: TensionReport) -> Graph: + """Placeholder graph restructuring hook.""" + _ = report + return graph + + +def _edge_report(graph: Graph, edge: Edge) -> dict: + return { + "source": edge.source, + "target": edge.target, + "relation": edge.relation, + "target_group": _semantic_group(graph.get_node(edge.target)), + "tension": edge_tension(graph, edge), + } + + +def _semantic_group(node: Node) -> str: + for key in ("context", "group", "type"): + value = node.metadata.get(key) + if value: + return str(value) + return "default" + + +def _choose_split_name(target: Node, split_names: list[str]) -> str: + group = _semantic_group(target).lower() + target_text = f"{target.id} {target.label} {group}".lower() + for split_name in split_names: + role = split_name.lower() + role_tail = role.rsplit("_", maxsplit=1)[-1] + if role in target_text or role_tail in target_text: + return split_name + + harm_terms = {"control", "trauma", "tension", "harm"} + comfort_terms = {"comfort", "ritual", "community", "support"} + tokens = set(target_text.replace("_", " ").split()) + if tokens & harm_terms: + return _find_split(split_names, "harm") or split_names[-1] + if tokens & comfort_terms: + return _find_split(split_names, "comfort") or split_names[0] + return split_names[0] + + +def _find_split(split_names: list[str], token: str) -> str | None: + for split_name in split_names: + if token in split_name.lower(): + return split_name + return None + + +def _split_label(original_label: str, split_name: str) -> str: + role = split_name.replace("_", " ").title() + if original_label.lower() in role.lower(): + return role + return f"{original_label} ({role})" + + +def _redirect_edge(edge: Edge, new_source_id: str) -> Edge: + payload = edge.model_dump(mode="json") + payload["source"] = new_source_id + payload["metadata"] = { + **payload.get("metadata", {}), + "redirected_from": edge.source, + } + return Edge(**payload) + + +def _fit_split_source_activations(graph: CIGraph, mapping: dict[str, str]) -> None: + grouped_edges: dict[str, list[Edge]] = defaultdict(list) + for edge in graph.edges: + if edge.target in mapping and edge.source == mapping[edge.target]: + grouped_edges[edge.source].append(edge) + + for source_id, edges in grouped_edges.items(): + numerator = 0.0 + denominator = 0.0 + for edge in edges: + target_activation = graph.get_node(edge.target).activation + numerator += edge.weight * edge.expected_ratio * target_activation + denominator += edge.weight * edge.expected_ratio**2 + if denominator > 0.0: + graph.set_activation(source_id, numerator / denominator) diff --git a/src/thinking_system/graph/cig/graph.py b/src/thinking_system/graph/cig/graph.py new file mode 100644 index 0000000..272e471 --- /dev/null +++ b/src/thinking_system/graph/cig/graph.py @@ -0,0 +1,125 @@ +from __future__ import annotations + +from copy import deepcopy +from dataclasses import dataclass, field + +import numpy as np + +from thinking_system.graph.cig.edge import Edge +from thinking_system.graph.cig.node import Node + + +@dataclass +class CIGraph: + """Mutable deterministic Concept/Constraint Intelligence Graph state.""" + + nodes: dict[str, Node] = field(default_factory=dict) + edges: list[Edge] = field(default_factory=list) + + def add_node(self, node: Node) -> None: + if node.id in self.nodes: + raise ValueError(f"Duplicate node id: {node.id}") + self.nodes[node.id] = node + + def add_edge(self, edge: Edge) -> None: + if edge.source not in self.nodes: + raise KeyError(f"Edge source does not exist: {edge.source}") + if edge.target not in self.nodes: + raise KeyError(f"Edge target does not exist: {edge.target}") + self.edges.append(edge) + + def get_node(self, node_id: str) -> Node: + return self.nodes[node_id] + + def node(self, node_id: str) -> Node: + """Compatibility alias for get_node.""" + return self.get_node(node_id) + + def edge(self, edge_key: int | str) -> Edge: + """Return an edge by list index or legacy extra `id` value.""" + if isinstance(edge_key, int): + return self.edges[edge_key] + for edge in self.edges: + if getattr(edge, "id", None) == edge_key: + return edge + raise KeyError(edge_key) + + def reset_activations(self, value: float = 0.0) -> None: + for node in self.nodes.values(): + node.activation = value + + def activation_vector(self) -> np.ndarray: + return np.array([node.activation for node in self.nodes.values()], dtype=float) + + def set_activation(self, node_id: str, activation: float) -> None: + self.get_node(node_id).activation = activation + + def copy(self) -> CIGraph: + return deepcopy(self) + + def clone(self) -> CIGraph: + return self.copy() + + def incoming_edges(self, node_id: str) -> list[Edge]: + return [edge for edge in self.edges if edge.target == node_id] + + def outgoing_edges(self, node_id: str) -> list[Edge]: + return [edge for edge in self.edges if edge.source == node_id] + + def outgoing(self, node_id: str) -> list[Edge]: + """Compatibility alias for outgoing_edges.""" + return self.outgoing_edges(node_id) + + def split_by_context(self) -> dict[str, CIGraph]: + """Return subgraphs grouped by node context metadata. + + TODO: Replace this with a proper context-boundary algorithm that can + preserve selected cross-context constraints. + """ + grouped: dict[str, CIGraph] = {} + for node in self.nodes.values(): + context = _node_context(node) + grouped.setdefault(context, CIGraph()).add_node(deepcopy(node)) + + for edge in self.edges: + source_context = _node_context(self.nodes[edge.source]) + target_context = _node_context(self.nodes[edge.target]) + if source_context == target_context: + grouped[source_context].add_edge(deepcopy(edge)) + + return grouped + + @classmethod + def from_records( + cls, + nodes: list[dict[str, object]], + edges: list[dict[str, object]], + ) -> CIGraph: + graph = cls() + for node_record in nodes: + graph.add_node(Node(**_normalize_node_record(node_record))) + for edge_record in edges: + graph.add_edge(Edge(**_normalize_edge_record(edge_record))) + return graph + + +def _normalize_node_record(record: dict[str, object]) -> dict[str, object]: + normalized = dict(record) + normalized.setdefault("label", normalized["id"]) + return normalized + + +def _normalize_edge_record(record: dict[str, object]) -> dict[str, object]: + normalized = dict(record) + normalized.setdefault("relation", "related") + return normalized + + +def _node_context(node: Node) -> str: + context = node.metadata.get("context") + if context is None: + context = getattr(node, "context", None) + return str(context or "default") + + +Graph = CIGraph diff --git a/src/thinking_system/graph/cig/io.py b/src/thinking_system/graph/cig/io.py new file mode 100644 index 0000000..cd2d6da --- /dev/null +++ b/src/thinking_system/graph/cig/io.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import yaml + +from thinking_system.graph.cig.graph import CIGraph + + +def load_graph(path: str | Path) -> CIGraph: + """Load a CIGraph from a YAML file.""" + with Path(path).open("r", encoding="utf-8") as handle: + payload: dict[str, Any] = yaml.safe_load(handle) or {} + return graph_from_mapping(payload) + + +def save_graph(graph: CIGraph, path: str | Path) -> None: + """Save a CIGraph to a YAML file.""" + payload = graph_to_mapping(graph) + with Path(path).open("w", encoding="utf-8") as handle: + yaml.safe_dump(payload, handle, sort_keys=False) + + +def load_graph_yaml(path: str | Path) -> CIGraph: + """Compatibility alias for load_graph.""" + return load_graph(path) + + +def graph_from_mapping(payload: dict[str, Any]) -> CIGraph: + nodes = payload.get("nodes", []) + edges = payload.get("edges", []) + if not isinstance(nodes, list): + raise TypeError("nodes must be a list") + if not isinstance(edges, list): + raise TypeError("edges must be a list") + return CIGraph.from_records(nodes=nodes, edges=edges) + + +def graph_to_mapping(graph: CIGraph) -> dict[str, list[dict[str, Any]]]: + return { + "nodes": [node.model_dump(mode="json") for node in graph.nodes.values()], + "edges": [edge.model_dump(mode="json") for edge in graph.edges], + } diff --git a/src/thinking_system/graph/cig/meaning.py b/src/thinking_system/graph/cig/meaning.py new file mode 100644 index 0000000..27db667 --- /dev/null +++ b/src/thinking_system/graph/cig/meaning.py @@ -0,0 +1,101 @@ +from __future__ import annotations + +from collections.abc import Iterable + +from thinking_system.graph.cig.engine import CIGEngine +from thinking_system.graph.cig.graph import Graph + + +def derivative_meaning( + graph: Graph, + input_node_id: str, + context_inputs: Iterable[str] | None = None, + epsilon: float = 0.05, + steps: int = 6, +) -> dict: + """Estimate meaning as d(final graph state) / d(input amount). + + M(u) ~= (final_state(input amount 1 + epsilon) - final_state(input amount 1)) / epsilon. + """ + if epsilon <= 0: + raise ValueError("epsilon must be positive") + if steps < 0: + raise ValueError("steps must be non-negative") + + context_node_ids = _normalize_context_inputs(context_inputs) + input_node_ids = [*context_node_ids, input_node_id] + + baseline = graph.copy() + perturbed = graph.copy() + + baseline_amounts = {node_id: 1.0 for node_id in input_node_ids} + perturbed_amounts = dict(baseline_amounts) + perturbed_amounts[input_node_id] = 1.0 + epsilon + + baseline_report = CIGEngine(baseline).run_cycle( + input_node_ids, + steps=steps, + input_amounts=baseline_amounts, + hold_inputs=False, + ) + perturbed_report = CIGEngine(perturbed).run_cycle( + input_node_ids, + steps=steps, + input_amounts=perturbed_amounts, + hold_inputs=False, + ) + + baseline_state = baseline_report["final_activations"] + perturbed_state = perturbed_report["final_activations"] + derivative_vector = { + node_id: (perturbed_state[node_id] - baseline_state[node_id]) / epsilon + for node_id in graph.nodes + } + + top_derivative_nodes = [ + { + "id": node_id, + "label": graph.get_node(node_id).label, + "derivative": value, + "abs_derivative": abs(value), + } + for node_id, value in sorted( + derivative_vector.items(), + key=lambda item: (-abs(item[1]), item[0]), + ) + if value != 0.0 + ] + + return { + "input_node": input_node_id, + "context_inputs": context_node_ids, + "epsilon": epsilon, + "steps": steps, + "derivative_vector": derivative_vector, + "top_derivative_nodes": top_derivative_nodes, + "baseline_final_activations": baseline_state, + "perturbed_final_activations": perturbed_state, + } + + +def meaning_derivative( + graph: Graph, + input_node_id: str, + perturbation: float = 0.01, +) -> dict[str, float]: + """Compatibility wrapper returning only the derivative vector.""" + report = derivative_meaning( + graph, + input_node_id=input_node_id, + epsilon=perturbation, + steps=1, + ) + return report["derivative_vector"] + + +def _normalize_context_inputs(context_inputs: Iterable[str] | None) -> list[str]: + if context_inputs is None: + return [] + if isinstance(context_inputs, str): + return [context_inputs] + return list(context_inputs) diff --git a/src/thinking_system/graph/cig/node.py b/src/thinking_system/graph/cig/node.py new file mode 100644 index 0000000..73bca79 --- /dev/null +++ b/src/thinking_system/graph/cig/node.py @@ -0,0 +1,24 @@ +from __future__ import annotations + +from pydantic import BaseModel, ConfigDict, Field, field_validator + + +class Node(BaseModel): + """A concept, claim, memory, symbol, or state in a CIG graph.""" + + model_config = ConfigDict(extra="allow", validate_assignment=True) + + id: str + label: str + activation: float = 0.0 + stability: float = Field(default=1.0, gt=0.0) + metadata: dict = Field(default_factory=dict) + + @field_validator("activation") + @classmethod + def clamp_activation(cls, value: float) -> float: + return max(0.0, min(1.0, float(value))) + + @property + def display_label(self) -> str: + return self.label diff --git a/src/thinking_system/graph/cig/tension.py b/src/thinking_system/graph/cig/tension.py new file mode 100644 index 0000000..49f14a6 --- /dev/null +++ b/src/thinking_system/graph/cig/tension.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +from pydantic import BaseModel, Field + +from thinking_system.graph.cig.edge import Edge +from thinking_system.graph.cig.graph import Graph + + +class EdgeTension(BaseModel): + edge_index: int + source: str + target: str + relation: str + value: float = Field(ge=0.0) + + +class TensionReport(BaseModel): + edge_tensions: list[EdgeTension] = Field(default_factory=list) + total: float = Field(default=0.0, ge=0.0) + + @property + def coherence(self) -> float: + """Simple coherence score in [0, 1] derived from total tension.""" + return 1.0 / (1.0 + self.total) + + +def edge_tension(graph: Graph, edge: Edge) -> float: + """Return tau_ij = weight * (a_j - expected_ratio * a_i)^2.""" + source_activation = graph.get_node(edge.source).activation + target_activation = graph.get_node(edge.target).activation + expected = edge.expected_ratio * source_activation + return float(edge.weight * (target_activation - expected) ** 2) + + +def total_tension(graph: Graph) -> float: + return float(sum(edge_tension(graph, edge) for edge in graph.edges)) + + +def tension_report(graph: Graph, top_k: int = 10) -> dict: + if top_k < 0: + raise ValueError("top_k must be non-negative") + + edge_reports = [] + for index, edge in enumerate(graph.edges): + source_activation = graph.get_node(edge.source).activation + target_activation = graph.get_node(edge.target).activation + expected = edge.expected_ratio * source_activation + edge_reports.append( + { + "edge_index": index, + "source": edge.source, + "target": edge.target, + "relation": edge.relation, + "weight": edge.weight, + "expected_ratio": edge.expected_ratio, + "source_activation": source_activation, + "target_activation": target_activation, + "expected_target_activation": expected, + "tension": edge_tension(graph, edge), + } + ) + + edge_reports.sort(key=lambda item: (-item["tension"], item["edge_index"])) + return { + "total": total_tension(graph), + "top_edges": edge_reports[:top_k], + } + + +def detect_tension(graph: Graph) -> TensionReport: + """Compatibility wrapper returning the older Pydantic report shape.""" + edge_tensions = [ + EdgeTension( + edge_index=index, + source=edge.source, + target=edge.target, + relation=edge.relation, + value=edge_tension(graph, edge), + ) + for index, edge in enumerate(graph.edges) + ] + return TensionReport(edge_tensions=edge_tensions, total=total_tension(graph)) diff --git a/tests/test_cig_compression_radius.py b/tests/test_cig_compression_radius.py new file mode 100644 index 0000000..772425e --- /dev/null +++ b/tests/test_cig_compression_radius.py @@ -0,0 +1,92 @@ +from pytest import approx + +from thinking_system.graph.cig.compression import representational_radius +from thinking_system.graph.cig.edge import Edge +from thinking_system.graph.cig.evolve import apply_context_split, suggest_context_split +from thinking_system.graph.cig.graph import CIGraph +from thinking_system.graph.cig.node import Node + + +def test_adding_primitive_node_increases_representational_radius() -> None: + graph = CIGraph() + graph.add_node(Node(id="base", label="Base")) + before = representational_radius(graph) + + graph.add_node( + Node( + id="axiom", + label="Axiom", + metadata={"primitive": True, "primitive_cost": 2.5}, + ) + ) + + assert representational_radius(graph) == approx(before + 2.5) + + +def test_adding_many_edges_increases_representational_radius_slightly() -> None: + graph = CIGraph() + graph.add_node(Node(id="source", label="Source")) + for index in range(5): + graph.add_node(Node(id=f"target_{index}", label=f"Target {index}")) + + before = representational_radius(graph, beta=0.01) + for index in range(5): + graph.add_edge( + Edge( + source="source", + target=f"target_{index}", + relation="connects", + ) + ) + + assert representational_radius(graph, beta=0.01) == approx(before + 0.05) + + +def test_context_split_has_nonzero_delta_r() -> None: + graph = make_split_graph() + + split = apply_context_split( + graph, + "religion", + { + "comfort": "religion_comfort", + "control": "religion_harm", + }, + ) + delta_r = representational_radius(split) - representational_radius(graph) + suggestion = suggest_context_split( + graph, + "religion", + ["religion_comfort", "religion_harm"], + ) + + assert delta_r > 0.0 + assert suggestion["delta_R"] == approx(delta_r) + assert suggestion["accepted"] is True + assert suggestion["tension_reduction"] > suggestion["alpha"] * suggestion["delta_R"] + + +def make_split_graph() -> CIGraph: + graph = CIGraph() + graph.add_node(Node(id="religion", label="Religion", activation=1.0)) + graph.add_node(Node(id="comfort", label="Comfort", activation=0.9)) + graph.add_node(Node(id="control", label="Control", activation=0.1)) + graph.add_edge( + Edge( + source="religion", + target="comfort", + relation="supports", + weight=1.0, + expected_ratio=0.9, + ) + ) + graph.add_edge( + Edge( + source="religion", + target="control", + relation="enables", + weight=2.0, + expected_ratio=0.9, + ) + ) + return graph diff --git a/tests/test_cig_context_split.py b/tests/test_cig_context_split.py new file mode 100644 index 0000000..812f9c7 --- /dev/null +++ b/tests/test_cig_context_split.py @@ -0,0 +1,132 @@ +from thinking_system.graph.cig.edge import Edge +from thinking_system.graph.cig.evolve import ( + apply_context_split, + find_overloaded_nodes, + suggest_context_split, +) +from thinking_system.graph.cig.graph import CIGraph +from thinking_system.graph.cig.node import Node +from thinking_system.graph.cig.tension import total_tension + + +def test_context_split_groups_nodes_and_keeps_internal_edges() -> None: + graph = CIGraph() + graph.add_node(Node(id="math_a", label="Math A", metadata={"context": "math"})) + graph.add_node(Node(id="math_b", label="Math B", metadata={"context": "math"})) + graph.add_node( + Node(id="culture_a", label="Culture A", metadata={"context": "culture"}) + ) + graph.add_edge(Edge(source="math_a", target="math_b", relation="related")) + graph.add_edge(Edge(source="math_a", target="culture_a", relation="related")) + + contexts = graph.split_by_context() + + assert set(contexts) == {"math", "culture"} + assert set(contexts["math"].nodes) == {"math_a", "math_b"} + assert set(contexts["culture"].nodes) == {"culture_a"} + assert len(contexts["math"].edges) == 1 + + +def test_find_overloaded_religion_node_and_suggest_split() -> None: + graph = make_overloaded_religion_graph() + + reports = find_overloaded_nodes(graph) + religion_report = reports[0] + suggestion = suggest_context_split( + graph, + "religion", + ["religion_comfort", "religion_harm"], + ) + + assert religion_report["node_id"] == "religion" + assert religion_report["outgoing_count"] == 6 + assert set(religion_report["target_groups"]) == {"comfort", "harm"} + assert suggestion["original_node"]["id"] == "religion" + assert [node["id"] for node in suggestion["new_nodes"]] == [ + "religion_comfort", + "religion_harm", + ] + assert suggestion["delta_R"] > 0.0 + assert suggestion["expected_complexity_increase_delta_R"] == suggestion["delta_R"] + assert "tension_reduction" in suggestion + assert "accepted" in suggestion + redirected = { + item["target"]: item["suggested_new_source"] + for item in suggestion["edges_to_redirect"] + } + assert redirected["comfort"] == "religion_comfort" + assert redirected["ritual"] == "religion_comfort" + assert redirected["community"] == "religion_comfort" + assert redirected["control"] == "religion_harm" + assert redirected["trauma"] == "religion_harm" + assert redirected["tension"] == "religion_harm" + + +def test_apply_context_split_returns_copy_and_can_reduce_tension() -> None: + graph = make_overloaded_religion_graph() + before = total_tension(graph) + + split = apply_context_split( + graph, + "religion", + { + "comfort": "religion_comfort", + "ritual": "religion_comfort", + "community": "religion_comfort", + "control": "religion_harm", + "trauma": "religion_harm", + "tension": "religion_harm", + }, + ) + after = total_tension(split) + + assert "religion_comfort" not in graph.nodes + assert "religion_harm" not in graph.nodes + assert "religion_comfort" in split.nodes + assert "religion_harm" in split.nodes + assert after < before + assert { + edge.source + for edge in split.edges + if edge.target in {"comfort", "ritual", "community"} + } == {"religion_comfort"} + assert { + edge.source + for edge in split.edges + if edge.target in {"control", "trauma", "tension"} + } == {"religion_harm"} + + +def make_overloaded_religion_graph() -> CIGraph: + graph = CIGraph() + graph.add_node(Node(id="religion", label="Religion", activation=1.0)) + for node_id in ["comfort", "ritual", "community"]: + graph.add_node( + Node( + id=node_id, + label=node_id.title(), + activation=0.9, + metadata={"context": "comfort"}, + ) + ) + for node_id in ["control", "trauma", "tension"]: + graph.add_node( + Node( + id=node_id, + label=node_id.title(), + activation=0.1, + metadata={"context": "harm"}, + ) + ) + + for target in ["comfort", "ritual", "community", "control", "trauma", "tension"]: + graph.add_edge( + Edge( + source="religion", + target=target, + relation="activates", + weight=1.0, + expected_ratio=0.9, + ) + ) + return graph diff --git a/tests/test_cig_derivative_meaning.py b/tests/test_cig_derivative_meaning.py new file mode 100644 index 0000000..fd36565 --- /dev/null +++ b/tests/test_cig_derivative_meaning.py @@ -0,0 +1,120 @@ +from pytest import approx + +from thinking_system.graph.cig.edge import Edge +from thinking_system.graph.cig.graph import CIGraph +from thinking_system.graph.cig.meaning import derivative_meaning, meaning_derivative +from thinking_system.graph.cig.node import Node + + +def test_meaning_derivative_compatibility_wrapper_returns_vector() -> None: + graph = CIGraph() + graph.add_node(Node(id="idea", label="Idea", activation=0.2)) + graph.add_node(Node(id="effect", label="Effect", activation=0.0)) + graph.add_edge( + Edge(source="idea", target="effect", relation="supports", weight=0.5) + ) + + derivative = meaning_derivative(graph, input_node_id="idea", perturbation=0.01) + + assert derivative["effect"] == approx(0.175) + + +def test_derivative_meaning_religion_with_comfort_context_has_expected_top_responses() -> ( + None +): + graph = CIGraph() + for node_id in ["religion", "ritual", "comfort", "coherence", "other"]: + graph.add_node(Node(id=node_id, label=node_id.title())) + graph.add_edge( + Edge(source="religion", target="ritual", relation="organizes", weight=0.5) + ) + graph.add_edge( + Edge(source="ritual", target="comfort", relation="supports", weight=0.5) + ) + graph.add_edge( + Edge(source="religion", target="comfort", relation="supports", weight=0.25) + ) + graph.add_edge( + Edge(source="comfort", target="coherence", relation="stabilizes", weight=0.5) + ) + + report = derivative_meaning( + graph, + input_node_id="religion", + context_inputs=["comfort"], + epsilon=0.05, + steps=6, + ) + top_ids = [node["id"] for node in report["top_derivative_nodes"][:3]] + + assert top_ids == ["comfort", "ritual", "coherence"] + assert report["derivative_vector"]["comfort"] > 0.0 + assert report["derivative_vector"]["ritual"] > 0.0 + assert report["derivative_vector"]["coherence"] > 0.0 + + +def test_derivative_vector_is_deterministic() -> None: + first = make_symbol_context_graph() + second = make_symbol_context_graph() + + first_report = derivative_meaning( + first, + input_node_id="number_47", + context_inputs=["humour"], + epsilon=0.05, + steps=6, + ) + second_report = derivative_meaning( + second, + input_node_id="number_47", + context_inputs=["humour"], + epsilon=0.05, + steps=6, + ) + + assert first_report["derivative_vector"] == second_report["derivative_vector"] + assert first_report["top_derivative_nodes"] == second_report["top_derivative_nodes"] + + +def test_same_symbol_with_different_context_produces_different_derivative_vectors() -> ( + None +): + symbol_context = derivative_meaning( + make_symbol_context_graph(), + input_node_id="number_47", + context_inputs=["arbitrary_symbol"], + epsilon=0.05, + steps=6, + ) + humour_context = derivative_meaning( + make_symbol_context_graph(), + input_node_id="number_47", + context_inputs=["humour"], + epsilon=0.05, + steps=6, + ) + + assert symbol_context["derivative_vector"] != humour_context["derivative_vector"] + + +def make_symbol_context_graph() -> CIGraph: + graph = CIGraph() + for node_id in ["number_47", "arbitrary_symbol", "humour", "coherence"]: + graph.add_node(Node(id=node_id, label=node_id.replace("_", " ").title())) + graph.add_edge( + Edge( + source="number_47", target="arbitrary_symbol", relation="evokes", weight=0.7 + ) + ) + graph.add_edge( + Edge(source="arbitrary_symbol", target="humour", relation="evokes", weight=0.7) + ) + graph.add_edge( + Edge( + source="arbitrary_symbol", + target="coherence", + relation="stabilizes", + weight=0.7, + ) + ) + return graph diff --git a/tests/test_cig_graph_load.py b/tests/test_cig_graph_load.py new file mode 100644 index 0000000..9021964 --- /dev/null +++ b/tests/test_cig_graph_load.py @@ -0,0 +1,36 @@ +from pathlib import Path + +from thinking_system.graph.cig.io import load_graph, save_graph + + +def test_load_graph_from_yaml() -> None: + graph = load_graph(Path("examples/cig_ts_core.yaml")) + + assert len(graph.nodes) == 27 + assert len(graph.edges) == 33 + assert graph.get_node("meaning").label == "Meaning" + assert graph.get_node("meaning").metadata["type"] == "concept" + + +def test_load_graph_contains_key_edges() -> None: + graph = load_graph(Path("examples/cig_ts_core.yaml")) + edge_keys = {(edge.source, edge.target, edge.relation) for edge in graph.edges} + + assert ("number_47", "arbitrary_symbol", "exemplifies") in edge_keys + assert ("meaning", "stable_attractor", "seeks") in edge_keys + assert ("religion", "ritual", "organizes") in edge_keys + assert ("compression", "R_radius", "minimizes") in edge_keys + assert ("R_radius", "minimum_R_compressor", "defines_objective_for") in edge_keys + + +def test_save_graph_round_trip(tmp_path: Path) -> None: + graph = load_graph(Path("examples/cig_ts_core.yaml")) + output_path = tmp_path / "round_trip.yaml" + + save_graph(graph, output_path) + reloaded = load_graph(output_path) + + assert list(reloaded.nodes) == list(graph.nodes) + assert len(reloaded.edges) == len(graph.edges) + assert reloaded.edges[0].source == graph.edges[0].source + assert reloaded.edges[0].metadata == graph.edges[0].metadata diff --git a/tests/test_cig_models.py b/tests/test_cig_models.py new file mode 100644 index 0000000..ef736da --- /dev/null +++ b/tests/test_cig_models.py @@ -0,0 +1,89 @@ +import pytest +from pydantic import ValidationError + +from thinking_system.graph.cig.edge import Edge +from thinking_system.graph.cig.graph import CIGraph +from thinking_system.graph.cig.node import Node + + +def test_create_node_defaults_and_clamps_activation() -> None: + node = Node(id="n1", label="Node 1", activation=1.5) + + assert node.id == "n1" + assert node.label == "Node 1" + assert node.activation == 1.0 + assert node.stability == 1.0 + assert node.metadata == {} + + +def test_node_stability_must_be_positive() -> None: + with pytest.raises(ValidationError): + Node(id="n1", label="Node 1", stability=0.0) + + +def test_create_edge_defaults() -> None: + edge = Edge(source="a", target="b", relation="supports") + + assert edge.source == "a" + assert edge.target == "b" + assert edge.relation == "supports" + assert edge.weight == 1.0 + assert edge.polarity == 1.0 + assert edge.expected_ratio == 1.0 + assert edge.metadata == {} + + +def test_edge_weight_must_be_non_negative() -> None: + with pytest.raises(ValidationError): + Edge(source="a", target="b", relation="supports", weight=-0.1) + + +def test_edge_polarity_is_bounded() -> None: + with pytest.raises(ValidationError): + Edge(source="a", target="b", relation="supports", polarity=2.0) + + +def test_add_nodes_edges_and_query_edges() -> None: + graph = CIGraph() + graph.add_node(Node(id="a", label="A")) + graph.add_node(Node(id="b", label="B")) + edge = Edge(source="a", target="b", relation="supports") + graph.add_edge(edge) + + assert graph.get_node("a").label == "A" + assert graph.incoming_edges("b") == [edge] + assert graph.outgoing_edges("a") == [edge] + + +def test_add_edge_requires_existing_nodes() -> None: + graph = CIGraph() + graph.add_node(Node(id="a", label="A")) + + with pytest.raises(KeyError): + graph.add_edge(Edge(source="a", target="missing", relation="supports")) + + +def test_set_and_reset_activations() -> None: + graph = CIGraph() + graph.add_node(Node(id="a", label="A")) + graph.add_node(Node(id="b", label="B", activation=0.4)) + + graph.set_activation("a", 0.7) + assert graph.get_node("a").activation == 0.7 + + graph.set_activation("b", -1.0) + assert graph.get_node("b").activation == 0.0 + + graph.reset_activations() + assert graph.activation_vector().tolist() == [0.0, 0.0] + + +def test_copy_is_deep() -> None: + graph = CIGraph() + graph.add_node(Node(id="a", label="A", activation=0.3)) + + copied = graph.copy() + copied.set_activation("a", 0.9) + + assert graph.get_node("a").activation == 0.3 + assert copied.get_node("a").activation == 0.9 diff --git a/tests/test_cig_propagation.py b/tests/test_cig_propagation.py new file mode 100644 index 0000000..be0c7aa --- /dev/null +++ b/tests/test_cig_propagation.py @@ -0,0 +1,59 @@ +from thinking_system.graph.cig.edge import Edge +from thinking_system.graph.cig.engine import CIGEngine, ThinkingSystemEngine +from thinking_system.graph.cig.graph import CIGraph +from thinking_system.graph.cig.io import load_graph +from thinking_system.graph.cig.node import Node + + +def test_propagation_moves_activation_along_weighted_edge() -> None: + graph = CIGraph() + graph.add_node(Node(id="source", label="Source", activation=1.0)) + graph.add_node(Node(id="target", label="Target", activation=0.0)) + graph.add_edge( + Edge(source="source", target="target", relation="supports", weight=0.5) + ) + + ThinkingSystemEngine(graph, relaxation_rate=0.0).propagate() + + assert graph.get_node("target").activation == 0.5 + + +def test_cycle_activates_seed_meaning_and_47_paths() -> None: + graph = load_graph("examples/cig_ts_core.yaml") + report = CIGEngine(graph).run_cycle( + ["meaning", "life", "number_47"], + steps=5, + ) + final = report["final_activations"] + + assert final["arbitrary_symbol"] > 0.9 + assert final["stable_attractor"] > 0.9 + assert final["coherence"] > 0.7 + assert final["absurdity"] > 0.6 + assert final["humour"] > 0.15 + assert report["input_nodes"] == ["meaning", "life", "number_47"] + assert report["steps"] == 5 + + +def test_cycle_activations_remain_in_unit_interval() -> None: + graph = load_graph("examples/cig_ts_core.yaml") + report = CIGEngine(graph).run_cycle( + ["meaning", "life", "number_47"], + steps=8, + ) + + assert all( + 0.0 <= activation <= 1.0 for activation in report["final_activations"].values() + ) + + +def test_cycle_output_is_deterministic_for_same_input() -> None: + first_graph = load_graph("examples/cig_ts_core.yaml") + second_graph = load_graph("examples/cig_ts_core.yaml") + + first = CIGEngine(first_graph).run_cycle(["meaning", "life", "number_47"], steps=5) + second = CIGEngine(second_graph).run_cycle( + ["meaning", "life", "number_47"], steps=5 + ) + + assert first == second diff --git a/tests/test_cig_tension.py b/tests/test_cig_tension.py new file mode 100644 index 0000000..258853a --- /dev/null +++ b/tests/test_cig_tension.py @@ -0,0 +1,85 @@ +from pytest import approx + +from thinking_system.graph.cig.edge import Edge +from thinking_system.graph.cig.engine import CIGEngine +from thinking_system.graph.cig.graph import CIGraph +from thinking_system.graph.cig.node import Node +from thinking_system.graph.cig.tension import ( + edge_tension, + tension_report, + total_tension, +) + + +def make_two_edge_graph() -> CIGraph: + graph = CIGraph() + graph.add_node(Node(id="source", label="Source", activation=1.0)) + graph.add_node(Node(id="matched", label="Matched", activation=0.5)) + graph.add_node(Node(id="far", label="Far", activation=0.0)) + graph.add_edge( + Edge( + source="source", + target="matched", + relation="matches", + weight=2.0, + expected_ratio=0.5, + ) + ) + graph.add_edge( + Edge( + source="source", + target="far", + relation="misses", + weight=2.0, + expected_ratio=0.5, + ) + ) + return graph + + +def test_edge_tension_is_near_zero_when_target_matches_expected_ratio() -> None: + graph = make_two_edge_graph() + + assert edge_tension(graph, graph.edges[0]) == approx(0.0) + + +def test_edge_tension_is_higher_when_target_is_far_from_expected() -> None: + graph = make_two_edge_graph() + + matched_tension = edge_tension(graph, graph.edges[0]) + far_tension = edge_tension(graph, graph.edges[1]) + + assert far_tension > matched_tension + assert far_tension == approx(0.5) + + +def test_total_tension_equals_sum_of_edge_tensions() -> None: + graph = make_two_edge_graph() + + expected = sum(edge_tension(graph, edge) for edge in graph.edges) + + assert total_tension(graph) == approx(expected) + + +def test_tension_report_sorts_edges_by_highest_tension() -> None: + graph = make_two_edge_graph() + + report = tension_report(graph, top_k=2) + + assert report["total"] == approx(0.5) + assert report["top_edges"][0]["target"] == "far" + assert report["top_edges"][0]["tension"] == approx(0.5) + assert report["top_edges"][1]["target"] == "matched" + + +def test_run_cycle_returns_tension_before_and_after() -> None: + graph = make_two_edge_graph() + + report = CIGEngine(graph).run_cycle(["source"], steps=1) + + assert "tension_before" in report + assert "tension_after" in report + assert "top_tension_edges_before" in report + assert "top_tension_edges_after" in report + assert isinstance(report["tension_before"], float) + assert isinstance(report["tension_after"], float)