Prismatic Mycelial Nx

Overview

Prismatic Mycelial Nx provides the platform's computational backbone for distributed data processing, dependency graph analysis, computational caching, and pipeline orchestration. Drawing its name from mycelial networks in biology -- decentralized, resilient systems that propagate information and resources across distributed nodes -- the application implements a graph-based computation model where data flows through interconnected processing stages with automatic backpressure, caching, and fault recovery. Rather than focusing narrowly on tensor operations, Mycelial Nx provides the infrastructure for building composable, observable, and self-healing data pipelines that underpin the platform's intelligence processing capabilities.

The module implements a directed acyclic graph (DAG) execution model where computational stages are connected through typed edges with propagation semantics. Each node in the computation graph represents a processing stage that can cache intermediate results, detect affected downstream consumers when inputs change, and participate in distributed consensus protocols for cross-node coordination. This architecture enables the platform to efficiently re-process only the portions of an intelligence pipeline affected by new data, rather than recomputing entire workflows from scratch.

The evolution subsystem provides a self-improving capability where pipeline configurations undergo genetic optimization, with safety gates and quality validators ensuring that evolutionary changes maintain or improve system behavior. This connects to the broader platform theme of autonomous self-improvement governed by rigorous verification gates.

Architecture

PrismaticMycelialNx.Application
+-- Pipeline Subsystem
|   +-- Pipeline.Producer (GenStage)
|   +-- Pipeline.Processor (GenStage)
|   +-- Pipeline.Consumer (GenStage)
|   +-- Pipeline.CircuitBreaker
|   +-- Pipeline.Backpressure
|   +-- Pipeline.Event (structured event types)
|
+-- Graph Subsystem
|   +-- Graph.DAG (directed acyclic graph engine)
|   +-- Graph.Node (computation node abstraction)
|   +-- Graph.Edge (typed edge with propagation)
|
+-- Cache Subsystem
|   +-- Cache.ComputationalCache (content-addressed)
|   +-- Cache.Store (pluggable backends)
|   +-- Cache.Backend.ETS
|   +-- Cache.Hasher (deterministic hashing)
|   +-- Cache.Invalidator (dependency-aware)
|   +-- Cache.Compressor (result compression)
|
+-- DependencyGraph Subsystem
|   +-- DependencyGraph.Analyzer
|   +-- DependencyGraph.AffectedEngine
|   +-- DependencyGraph.GitIntegration
|   +-- DependencyGraph.FlowIntegration
|
+-- Sync Subsystem
|   +-- Sync.Coordinator (distributed coordination)
|   +-- Sync.VectorClock (causal ordering)
|   +-- Sync.CRDT (conflict-free replicated data)
|   +-- Sync.Consensus (agreement protocols)
|   +-- Sync.KnowledgeState
|   +-- Sync.StreamProducer / StreamConsumer
|
+-- Evolution Subsystem
    +-- Evolution.Orchestrator
    +-- Evolution.Strategy
    +-- Evolution.SafetyGate
    +-- Evolution.QualityValidator
    +-- Evolution.Rollback
    +-- Evolution.Metrics
    +-- Evolution.DSL
    +-- Evolution.BroadwayPipeline

Input Data --> DAG Scheduler --> Stage Execution --> Cache Check --> Compute/Retrieve
     |              |                |                 |                |
  Typed Events  Topological      GenStage           Content-Hash    Deterministic
  with Schema   Sort + Priority  Backpressure       Lookup          Execution
     |              |                |                 |                |
     +-- Affected Engine --+-- Circuit Breaker --+-- Invalidation --+
                           |
               Evolution Optimizer --> Safety Gate --> Quality Validator

The architecture separates concerns into five distinct subsystems. The Pipeline subsystem handles streaming data processing with GenStage backpressure. The Graph subsystem provides the DAG execution engine. The Cache subsystem implements content-addressed caching with dependency-aware invalidation. The Sync subsystem enables distributed coordination across cluster nodes. The Evolution subsystem drives autonomous optimization of pipeline configurations.

Key Modules

Core Protocols and Behaviours

The application defines a rich set of protocols and behaviours that enable composable, type-safe pipeline construction.

Protocols

Behaviours

Configuration

config :prismatic_mycelial_nx,
  # Pipeline configuration
  default_concurrency: System.schedulers_online(),
  max_demand: 100,
  min_demand: 50,

  # Cache configuration
  cache_backend: PrismaticMycelialNx.Cache.Backend.ETS,
  cache_ttl: :timer.hours(1),
  cache_max_size_mb: 512,
  compression_enabled: true,

  # Circuit breaker
  circuit_breaker_threshold: 5,
  circuit_breaker_reset_ms: 30_000,

  # Evolution
  evolution_enabled: false,
  evolution_strategy: :genetic,
  safety_gate_enabled: true,

  # Sync
  vector_clock_enabled: true,
  crdt_backend: :g_counter,

  # Telemetry
  telemetry_prefix: [:prismatic_mycelial_nx]

API Reference

# Build and execute a computation DAG
{:ok, dag} = PrismaticMycelialNx.build_dag([
  {:fetch_data, fn -> fetch_osint_data() end},
  {:normalize, fn data -> normalize(data) end, depends_on: :fetch_data},
  {:enrich, fn data -> enrich_with_context(data) end, depends_on: :normalize},
  {:analyze, fn data -> run_analysis(data) end, depends_on: :enrich}
])
{:ok, results} = PrismaticMycelialNx.execute(dag)

# Detect affected stages after input change
{:ok, affected} = PrismaticMycelialNx.affected_stages(dag, changed: [:fetch_data])
# => [:normalize, :enrich, :analyze]

# Content-addressed caching
{:ok, cached} = PrismaticMycelialNx.Cache.ComputationalCache.get_or_compute(
  cache_key, fn -> expensive_computation() end)

# Dependency graph analysis
{:ok, deps} = PrismaticMycelialNx.DependencyGraph.analyze("apps/prismatic_signals")
# => %{modules: 12, dependencies: 34, cycles: 0}

# Evolutionary pipeline optimization
{:ok, optimized} = PrismaticMycelialNx.Evolution.optimize(pipeline_config,
  strategy: :genetic,
  generations: 50,
  fitness: &measure_throughput/1)

Testing

# Run all Mycelial Nx tests
cd apps/prismatic_mycelial_nx && mix test

# Run with coverage
mix test --cover

# Run pipeline tests
mix test test/prismatic_mycelial_nx/pipeline

# Run cache tests
mix test test/prismatic_mycelial_nx/cache

# Run evolution tests
mix test test/prismatic_mycelial_nx/evolution

# Run benchmarks
mix mycelial_nx.test --benchmark

Testing includes unit tests for each protocol and behaviour implementation, integration tests for end-to-end DAG execution with caching, property-based tests for CRDT convergence and vector clock ordering, circuit breaker state transition tests, and benchmark suites measuring pipeline throughput and cache hit rates. The evolution subsystem has dedicated safety gate tests verifying that optimized configurations cannot degrade system behavior below established baselines.

Integration Points

NABLA Compliance

Performance

Related Resources

• Prismatic Agents -- Agent execution leveraging Mycelial Nx pipeline infrastructure
• Prismatic Storage ETS -- ETS-backed computational cache backend
• Prismatic Telemetry -- Pipeline and cache performance metrics
• Multi-Paradigm Solving -- Combines graph-based, stream-based, and evolutionary approaches
• NABLA Axioms -- Pipeline provenance tracking and cached result time decay
• Telemetry Integration -- Pipeline execution observability and performance monitoring

Connect & Contribute

Created by Tomáš Korcak (korczis) | Open Source under GHL

• GitHub | GitLab | LinkedIn | Contact
• Developer Portal | Architecture | Meet the Creator