catalyst-brain 1.6.0

Catalyst Brain: persistent agent memory, compact recall, replayable evidence, and token-efficient AI infrastructure.

Catalyst Brain SDK

Persistent agent memory, compact recall, replayable evidence, and token-efficient AI infrastructure

---

catalyst-brain is a closed-source/freemium Rust+PyO3 SDK for persistent agent memory, compact recall, replayable evidence, and token-efficient AI infrastructure. It gives AI systems a local memory ledger, compact prompt-pack generation, HDC primitives, cache adapters, and hosted edge/API paths without making users sign up before they can evaluate the SDK.

Catalyst Brain is a closed-source/freemium PyPI SDK. Local workflows run without signup, while hosted edge/API paths are backed by Cloudflare-hosted services and production usage is governed by quota, entitlement, and commercial terms.

Install locally from PyPI:

pip install --upgrade catalyst-brain

Quick Start

Run a local activation and import smoke:

python -m catalyst_brain

Check the hosted edge path when you need it:

python -m catalyst_brain --edge --json

Use the native HDC primitives directly:

import catalyst_hdc as hdc

a = hdc.rand_bipolar(4096)
b = hdc.rand_bipolar(4096)
score = hdc.resonance(a, b)     # around 0.5 for unrelated vectors
bound = hdc.hdc_bind(a, b)      # self-inverse binding
bundled = hdc.hdc_bundle(a, b)  # majority-vote superposition
shifted = hdc.hdc_permute(a, 3) # circular shift

Agent Memory CLI

Use the operational memory CLI for bounded agent recall, handoff continuity, and evidence capture:

catalyst-brain brain --ledger ~/.codex/memories/catalyst-agent-brain.jsonl \
  --agent-id codex-catalyst --json \
  pulse --role user --domain cli \
  --book "Durable fact or policy" \
  "Short event summary"

catalyst-brain brain --ledger ~/.codex/memories/catalyst-agent-brain.jsonl \
  --agent-id codex-catalyst --json \
  inject --limit 3 --max-chars 1200 "task query"

catalyst-brain brain --ledger ~/.codex/memories/catalyst-agent-brain.jsonl \
  --agent-id codex-catalyst --json \
  doctor

catalyst-brain brain --ledger ~/.codex/memories/catalyst-agent-brain.jsonl \
  --agent-id codex-catalyst --json \
  doctor --edge

catalyst-brain brain --ledger ~/.codex/memories/catalyst-agent-brain.jsonl \
  --agent-id codex-catalyst --json \
  repair --backup ~/.codex/memories/catalyst-agent-brain.jsonl.bak

catalyst-brain brain --ledger ~/.codex/memories/catalyst-agent-brain.jsonl \
  --agent-id codex-catalyst --json \
  compact --recent 5 --max-chars 1200

---

Production Boundaries

catalyst-brain is a product SDK, not an open-source project.

Tier	Intended use	Notes
Free / evaluation	Research, learning, benchmarks, prototypes, early integration	Available from PyPI with no signup or API key required
Production	Production systems, SaaS, hosted APIs, revenue-generating workflows	Contact [e-mail] for the production path, higher quotas, and support
Enterprise	Source access, private support, custom terms, deployment assistance	Contact [e-mail] for pilots and private deployment

The public PyPI package is the supported distribution path. Most users can prototype freely before they need an account, API key, or higher quota path. Redistribution, production deployment, hosted API use, and derivative implementations of the patented methods require written permission.

Support: [e-mail]
Pilots and commercial access: [e-mail]

---

Trust And Support

• Security policy: report vulnerabilities privately; do not put secrets, API keys, payloads, or customer data in public issues.
• Privacy policy: anonymous telemetry is opt-out and does not collect prompts, labels, vectors, model outputs, or API keys.
• Support guide: free evaluation path, production/API-key path, and enterprise/pilot contact.

---

User Commitment

The SDK is already being used by a growing base of early adopters. We treat free-tier users as real users, not throwaway traffic.

• Keep pip install catalyst-brain as the primary onboarding path.
• Preserve documented import paths and method signatures across patch releases.
• Document breaking changes in CHANGELOG.md before users hit them.
• Ship Python type stubs with the wheel so IDEs and CI can catch mistakes early.
• Keep telemetry anonymous, opt-out, and free of user data, vectors, labels, or model outputs.
• Make the free-tier path clear so teams can prototype freely and reach out before production, hosted API use, or higher-volume deployment.

---

SDK Reference

Core HDC Primitives

Raw hypervector algebra. All other SDK classes are built on these.

Function	Signature	Description
rand_bipolar	(dim: int) → list[float]	Random {−1, +1} hypervector
resonance	(a, b) → float	Cosine similarity normalized to [0, 1]
hdc_bind	(a, b) → list[float]	XOR-like binding (self-inverse: bind(bind(a,b),b) == a)
hdc_bundle	(a, b) → list[float]	Majority-vote superposition
hdc_permute	(v, n) → list[float]	Circular shift by n positions
normalise_bipolar	(v) → list[float]	Normalize to bipolar range

a = hdc.rand_bipolar(4096)
b = hdc.rand_bipolar(4096)

# Self-inverse binding (XOR)
assert hdc.resonance(a, b) > 0.4          # quasi-orthogonal
bound = hdc.hdc_bind(a, b)
recovered = hdc.hdc_bind(bound, b)
assert hdc.resonance(a, recovered) > 0.99  # a ⊕ (a ⊕ b) = b (bit-exact)

# Bundle N vectors using reduce
from functools import reduce
vectors = [hdc.rand_bipolar(4096) for _ in range(4)]
superposition = reduce(hdc.hdc_bundle, vectors)

---

HoloCPU SDK — Cognitive Compute Engine

O(1) semantic memory with Grover-amplified attention routing.

from catalyst_hdc import PyHoloCPUScheduler
import catalyst_hdc as hdc

cpu = PyHoloCPUScheduler(dim=4096, quantum_capacity=8)

Memory

# Store and recall — O(1) regardless of how many memories exist
cpu.store_memory("user_preference_dark_mode")
cpu.store_memory("last_query")

assert cpu.recall("user_preference_dark_mode") == True
assert cpu.recall("nonexistent_key") == False

# Export entire cognitive state as a single 4096-float hypervector (16 KB constant)
state = cpu.export_holographic_state()
assert len(state) == 4096  # always 16 KB

Outcome Feedback

# Signal quality of an inference result (0.0 = bad, 1.0 = perfect)
cpu.process_feedback(0.95)   # positive outcome
print(cpu.feedback_strength())  # → 0.95 (elevated from baseline 0.5)

cpu.process_feedback(0.1)    # negative outcome
print(cpu.feedback_strength())  # → drops toward baseline

Grover-Amplified Attention

# quantum_grover_search takes a hypervector query + lists of key/value hypervectors
query = hdc.rand_bipolar(4096)
keys  = [hdc.rand_bipolar(4096) for _ in range(8)]
values = [hdc.rand_bipolar(4096) for _ in range(8)]

output = cpu.quantum_grover_search(query, keys, values)
# Returns a 4096-dim output vector from Grover-amplified routing
assert len(output) == 4096

Role Vectors

# Generate orthogonal role hypervectors for structured binding
agent   = cpu.generate_role("agent")
user    = cpu.generate_role("user")
system  = cpu.generate_role("system")

# Use for structured message encoding: message = bind(content, agent_role)

API Reference

Method	Signature	Description
dimension()	→ int	Hypervector dimensionality
quantum_capacity()	→ int	Qubit depth
store_memory(key)	(str) → None	Encode and store a semantic key
recall(key)	(str) → bool	O(1) key existence check
export_holographic_state()	→ list[float]	Full state as 4096 floats (16 KB)
process_feedback(signal)	(float) → None	Outcome feedback signal (0.0–1.0)
feedback_strength()	→ float	Current feedback strength
quantum_grover_search(query, keys, values)	(Vec, list[Vec], list[Vec]) → list[float]	Grover attention
run_audit_integrity_check()	→ bool	System health check
generate_role(label)	(str) → list[float]	Orthogonal role vector

---

HoloGen SDK — Geometric Hypervector Engine

Encode 3D geometry, materials, and photon states directly into hypervector space.

from catalyst_hdc import PyHoloGenEngine

engine = PyHoloGenEngine(dim=10_000)

Pixel Geometry

# Map screen coordinates to hypervector addresses
pixel_hv = engine.generate_pixel_geometry(64, 64)
# → list[int8], quasi-orthogonal per unique (x, y) pair

pixel_a = engine.generate_pixel_geometry(100, 200)
pixel_b = engine.generate_pixel_geometry(100, 201)  # adjacent pixel
# pixel_a and pixel_b are quasi-orthogonal — no hash collisions

Surface Materials

# A metallic surface at position (10, 0, 5) facing upward
surface_hv = engine.generate_material_mapping(
    position=[10.0, 0.0, 5.0],  # [f32; 3]
    normal=[0.0, 1.0, 0.0],     # surface normal [f32; 3]
    material_id=42
)

Photon State

# Form 1: encode photon color as a semantic hypervector
photon_hv = engine.generate_photon("blue")
# Supported: "violet"/"purple", "blue", "cyan", "green", "yellow",
# "amber"/"orange", "red", "white".

# Form 2: full geometric form
photon_hv = engine.generate_photon(
    [0.0, 5.0, 0.0],   # position [f32; 3]
    [1.0, 0.0, 0.0],   # direction [f32; 3]
    480.0,             # wavelength (nm)
)

BVH Nodes

# encode_bvh_node(min_bounds, max_bounds, left_hv, right_hv)
# left_hv and right_hv must be bipolar hypervectors as list[int8]
# Convert: [int(x) for x in hdc.rand_bipolar(dim)]

left_hv  = [int(x) for x in hdc.rand_bipolar(4096)]
right_hv = [int(x) for x in hdc.rand_bipolar(4096)]

bvh_node = engine.encode_bvh_node(
    [0.0, -10.0, 0.0],   # min_bounds [f32; 3]
    [10.0, 10.0, 10.0],  # max_bounds [f32; 3]
    left_hv,
    right_hv,
)

Counterfactual Physics

# Ask "what if this photon took a different path?"
# Inputs may be either bipolar i8 hypervectors or any object that stringifies
# (the latter is hashed deterministically into a hypervector of dim D).
actual_state = "jump→reward"
intervention = "crouch→reward"

alt_reality = engine.simulate_counterfactual(actual_state, intervention)
# Returns hypervector encoding hypothetical deviation (list[int8] of length D)

API Reference

Method	Signature	Description
structural_dimension()	→ int	Hypervector dimensionality
generate_pixel_geometry(x, y, frame_id=None)	(u32, u32, Optional[u64]) → list[int8]	Pixel coords → HDC address. frame_id defaults to 0.
generate_material_mapping(position, normal, material_id)	([f32;3], [f32;3], u32) → list[int8]	Surface → HDC
generate_photon(color)	(str) → list[int8]	Color string → HDC. Also accepts (position, direction, wavelength) as the geometric form.
encode_bvh_node(min_bounds, max_bounds, left_hv, right_hv)	([f32;3], [f32;3], Vec<i8>, Vec<i8>) → list[int8]	BVH node
simulate_counterfactual(state, intervention)	(Any, Any) → list[int8]	Counterfactual physics. Args are either bipolar i8 vectors or any stringifiable object (hashed to a vector).

---

Metacognition & Self-Audit

Self-improvement loop: observe → recommend → apply → audit.

from catalyst_hdc import PyMetacognition, PyOptimizer, PySelfAudit
import catalyst_hdc as hdc

meta = PyMetacognition(dim=4096)

Record Observations

# Record inference outcomes with resonance, coherence, accuracy
hv = hdc.rand_bipolar(4096)
meta.record(res=0.85, coh=0.90, acc=0.75, context=hv, hash=12345)
meta.record(res=0.92, coh=0.88, acc=0.81, context=hv, hash=12346)
meta.record(res=0.61, coh=0.72, acc=0.55, context=hv, hash=12347)

Query State

print(f"success_rate:  {meta.success_rate():.3f}")   # ratio of high-resonance successes
print(f"avg_resonance: {meta.avg_resonance():.3f}")  # mean resonance score
recs = meta.recommend()
# → [("momentum_increase", 0.05, "success rate > 80%, reinforce"), ...]

Apply Recommendations

opt = PyOptimizer()
opt.apply("momentum_increase", 0.05, "success rate above 80%")
params = opt.get_params()
# → {"learning_rate": 0.6, "momentum": 0.5, "attention_weight": 0.55, "identity_lr": 0.01}
opt.rollback()  # revert last parameter change

Audit Integrity

audit = PySelfAudit(dim=4096)
hv = hdc.rand_bipolar(4096)
score, passed, issues = audit.full_audit(hv)
# → score=1.0, passed=True, issues=[]

API Reference

Class	Method	Signature	Description
PyMetacognition	record(res, coh, acc, context, hash)	(float, float, float, Vec, u64)	Log observation
PyMetacognition	success_rate()	→ float	Ratio of high-res successes
PyMetacognition	avg_resonance()	→ float	Mean resonance
PyMetacognition	recommend()	→ list[tuple]	Parameter recommendations
PyOptimizer	apply(action, delta, reason)	(str, float, str)	Apply parameter delta
PyOptimizer	get_params()	→ dict	Current parameters
PyOptimizer	rollback()	→ None	Revert last change
PySelfAudit	full_audit(hv)	(Vec) → (float, bool, list)	Integrity check

---

Quantum Attention Head

Drop-in replacement for standard softmax attention using Grover-amplified routing.

from catalyst_hdc import PyQuantumAttentionHead
import catalyst_hdc as hdc

head = PyQuantumAttentionHead(dim=512, nqubits=40)

query  = hdc.rand_bipolar(512)
keys   = [hdc.rand_bipolar(512) for _ in range(10)]
values = [hdc.rand_bipolar(512) for _ in range(10)]

output = head.compute(query, keys, values)
# Returns 512-dim output vector

Method	Signature	Description
compute(query, keys, values)	(Vec, list[Vec], list[Vec]) → list[float]	Grover attention

Note: amplify() does not exist as a standalone method. Grover amplification for large memory stores is implemented inside PyHoloCPUScheduler.quantum_grover_search(). PyQuantumAttentionHead is for fine-grained per-layer attention.

---

HoloSwarm — Multi-Agent Spectral Synthesis

Superpose an arbitrary number of agents (Role ⊗ Policy ⊗ Skill) into a single hypervector and tune into any one of them at query time via iterative resonance decomposition.

from catalyst_hdc import PyHoloSwarm
import catalyst_hdc as hdc

swarm = PyHoloSwarm(dim=4096)

# Register agents — each compound is permuted before bundling
# to de-correlate overlapping roles.
swarm.add_agent(
    role="planner",   r_hv=hdc.rand_bipolar(4096),
    policy="explore", p_hv=hdc.rand_bipolar(4096),
    skill="search",   s_hv=hdc.rand_bipolar(4096),
)
swarm.add_agent(
    role="executor",  r_hv=hdc.rand_bipolar(4096),
    policy="exploit", p_hv=hdc.rand_bipolar(4096),
    skill="tool_use", s_hv=hdc.rand_bipolar(4096),
)

# Decompose: given a role key, recover policy + skill via iterative unbinding
role, policy, skill, confidence = swarm.resonate(
    role_key="planner",
    p_guess=hdc.rand_bipolar(4096),
    s_guess=hdc.rand_bipolar(4096),
    max_iter=10,
)

# Probe which agents are active in a semantic sector
active = swarm.materialize(probe=hdc.rand_bipolar(4096), threshold=0.6)
# → [("planner", 0.73), ...]

Method	Signature	Description
add_agent(role, r_hv, policy, p_hv, skill, s_hv)	(str, Vec, str, Vec, str, Vec) → None	Superpose Role ⊗ Policy ⊗ Skill into swarm
add_paradox_trap(names, roles, keys)	(list[str], list[Vec], list[Vec]) → None	Recursive causal-loop trap (HoloSec)
resonate(role_key, p_guess, s_guess, max_iter)	→ tuple[str,str,str,float]	Decompose swarm into (role, policy, skill, confidence)
materialize(probe, threshold)	(Vec, float) → list[tuple[str,float]]	Find agents resonating above threshold
get_swarm_vector()	→ list[float]	Raw composite hypervector

---

PyHKVC — Holographic Key-Value Cache

O(1) recency-unbiased KV cache using complex-domain phase accumulation. All entries contribute equal representational weight regardless of insertion order — no recency bias.

from catalyst_hdc import PyHKVC

cache = PyHKVC(dim=1024)

# Store key-value pairs at sequence positions
cache.store("question:capital_france", "Paris", position=0)
cache.store("question:capital_japan",  "Tokyo", position=1)
cache.store("question:capital_uk",     "London", position=2)

# O(1) retrieval: HashMap lookup → phase-domain resonance
value, score = cache.query("question:capital_france")
# → ("Paris", 0.94)

print(cache.count())  # → 3

Method	Signature	Description
store(key, value, position)	(str, str, int) → None	Insert key-value at position
query(query_key)	(str) → tuple[str, float]	Retrieve (value, confidence)
count()	→ int	Number of stored entries
position_score(position)	(int) → float	Recency-bias diagnostic (should be ≈constant)
accumulator_magnitude()	→ list[float]	Raw complex accumulator magnitudes

---

CausalMemory & MultiHopReasoner

Store causal relationships as hypervector triples and query them holographically.

from catalyst_hdc import PyCausalMemory, PyMultiHopReasoner
import catalyst_hdc as hdc

# CausalMemory: cause → effect temporal chains
mem = PyCausalMemory(dim=4096)

t0 = hdc.rand_bipolar(4096)   # time-role HV
cause  = hdc.rand_bipolar(4096)
effect = hdc.rand_bipolar(4096)

mem.store(cause, effect, t0)

recovered_effect = mem.recall_effect(cause)   # → list[float] or None
recovered_causes = mem.recall_cause(effect)   # → list[list[float]]
by_time          = mem.recall_by_time(t0)     # → list[float] or None

# MultiHopReasoner: traverse fact graphs up to N hops
reasoner = PyMultiHopReasoner(dim=4096)

f0 = reasoner.add_fact(hdc.rand_bipolar(4096))   # → index 0
f1 = reasoner.add_fact(hdc.rand_bipolar(4096))   # → index 1
reasoner.add_link(f0, f1)

query = hdc.rand_bipolar(4096)
results = reasoner.reason(query, hops=2)
# → [(fact_index, resonance_score), ...] sorted by resonance descending

Class	Method	Description
PyCausalMemory	store(cause, effect, time)	Record a causal triple
PyCausalMemory	recall_effect(cause)	Retrieve effect for a cause
PyCausalMemory	recall_cause(effect)	Retrieve all causes for an effect
PyCausalMemory	recall_by_time(time)	Retrieve effect at a time
PyMultiHopReasoner	add_fact(hv)	Register a fact, returns index
PyMultiHopReasoner	add_link(a, b)	Undirected association between facts
PyMultiHopReasoner	reason(query, hops)	Multi-hop resonance query

---

Rain Protocol — Stateless Agent State Transfer

Rain v2 is a binary-first wire protocol for transferring HDC agent state between serverless invocations. Instead of a database or JSON tokens, agents exchange compact .rain binaries carrying their holographic world vector, causal edges, and Hebbian weights.

Wire format (48-byte header + zlib payload):

[RAIN 4B][version u16 BE][flags u16 BE][dim u32 BE][n_edges u32 BE][sha256 32B][compressed body]

from catalyst_brain import RainPayload, rain_dumps, rain_loads
from catalyst_brain.rain import merge_digests, RainDigest, to_header, from_header
import catalyst_hdc as hdc

# Serialize agent state to .rain bytes
wv = hdc.rand_bipolar(10_000)
payload = RainPayload(
    agent_id="swarm-lead",
    dim=10_000,
    world_vector=wv,
)
blob = rain_dumps(payload)       # compact binary, SHA-256 verified
print(len(blob))                 # ≪ 100 KB even for 10k-dim vectors

# Round-trip
restored = rain_loads(blob)
assert restored.agent_id == "swarm-lead"
assert len(restored.world_vector) == 10_000

File I/O

from catalyst_brain.rain import dump, load

dump(payload, "checkpoint.rain")
restored = load("checkpoint.rain")

HTTP Header Transfer

Pass agent state between serverless functions without a database:

# Agent A — encode state into request header
header_value = to_header(payload)
# → base64 string, drop into X-Rain-State header

# Agent B — recover state on the other side
incoming = from_header(request.headers["X-Rain-State"])
resume_from(incoming.world_vector)

Rain Evidence Envelope

Attach replay-safe workflow metadata to a .rain payload when you need auditable handoff, idempotency, or capability-scoped state transfer:

from catalyst_brain import (
    RainPayload,
    seal_evidence_envelope,
    verify_evidence_envelope,
)

payload = RainPayload(agent_id="coding-agent", dim=4096, world_vector=wv)
sealed = seal_evidence_envelope(
    payload,
    secret="shared-replay-secret",
    workflow_id="run-2026-06-17",
    step_index=3,
    idempotency_token="run-2026-06-17:step-3",
    capability_scope=("memory.write", "rain.replay"),
    uncertainty=0.08,
)

verified = verify_evidence_envelope(
    sealed,
    secret="shared-replay-secret",
    expected_workflow_id="run-2026-06-17",
    last_step_index=2,
    required_capability="memory.write",
)
assert verified.ok

The envelope stores public-safe metadata under payload.metadata["rain_evidence"]: workflow id, monotonic step, idempotency token, capability scope, state-integrity digest, uncertainty, and HMAC signature. It does not expose private memory internals.

Holographic Digest Merge

Combine knowledge from N specialist agents into one vector without exposing underlying data:

digest_a = RainDigest(agent_id="specialist-A", vector=hdc.rand_bipolar(4096))
digest_b = RainDigest(agent_id="specialist-B", vector=hdc.rand_bipolar(4096))

merged = merge_digests([digest_a, digest_b])
# → RainDigest with bundled (majority-vote) world vector

Function	Signature	Description
rain_dumps(payload)	(RainPayload) → bytes	Serialize to .rain binary
rain_loads(data)	(bytes) → RainPayload	Deserialize from .rain binary
dump(payload, path)	(RainPayload, str|Path) → None	Write .rain file
load(path)	(str|Path) → RainPayload	Read .rain file
to_header(payload)	(RainPayload) → str	Base64 encode for X-Rain-State header
from_header(value)	(str) → RainPayload	Decode X-Rain-State header
merge_digests(digests)	(list[RainDigest]) → RainDigest	Algebraic knowledge merge
seal_evidence_envelope(payload, ...)	(RainPayload, secret, workflow_id, step_index, idempotency_token, ...) → RainPayload	Attach signed replay/evidence metadata
verify_evidence_envelope(payload, ...)	(RainPayload, secret, ...) → RainEvidenceVerification	Verify state integrity, signature, workflow, replay, and capability constraints

---

Memory Update Policy — Coherence Guard

Use MemoryUpdatePolicy to make bounded memory-write decisions from public signals supplied by your application. It is a policy surface for novelty, confidence, saturation, and evidence count; it is not a disclosure of private memory recipes.

from catalyst_brain import MemoryUpdatePolicy

policy = MemoryUpdatePolicy(
    min_novelty=0.20,
    min_confidence=0.65,
    consolidate_saturation=0.75,
    max_saturation=0.95,
)

decision = policy.evaluate(
    novelty=0.72,
    confidence=0.91,
    saturation=0.34,
    evidence_count=3,
)

if decision.action == "accept":
    store_memory_update()
elif decision.action == "consolidate":
    compact_then_store()
else:
    skip_update(decision.reason)

Decision	Meaning
accept	Novel, reliable update can be written directly
consolidate	Useful update should be compacted/consolidated before write
reject	Update failed a confidence, novelty, or saturation guard

---

CatalystTokenKernel — Progressive Tool Discovery

Use CatalystTokenKernel to keep large tool schemas and execution output out of the model context until the agent actually needs them. It is designed as a thin kernel for MCP servers and coding agents that want paginated tool discovery, schema-on-demand expansion, deferred code-execution status records, and Rain state handoff.

from catalyst_brain import CatalystTokenKernel, ToolSpec

kernel = CatalystTokenKernel(dim=4096)
kernel.register_tool(
    ToolSpec(
        name="sandbox.execute_python",
        description="Run Python code safely in a deferred sandbox task.",
        input_schema={
            "type": "object",
            "properties": {"code": {"type": "string"}},
            "required": ["code"],
        },
        tags=("code", "execution", "python", "sandbox"),
    )
)

# Progressive tools/list style page: no full schema in context.
page = kernel.list_tools(limit=10)
print(page.tools[0]["schema_ref"])

# Query-gated discovery: expand the schema only when dispatch is likely.
tool = kernel.discover("run python safely", limit=1, include_schema=True)[0]
print(tool["schema"]["properties"]["code"]["type"])

# Deferred code-execution state: compact status first, full output on fetch.
task = kernel.run_python_task("print('hello')")
result = kernel.fetch_task_result(task["task_id"])

# Rain snapshot for compact agent/session handoff.
snapshot = kernel.export_rain_snapshot(agent_id="coding-agent")
print(snapshot["estimated_reduction_ratio"])

Class / Method	Description
ToolSpec	Verbose tool definition registered once
CatalystTokenKernel.register_tool(spec)	Stores a full descriptor in PyHKVC and returns a compact handle
CatalystTokenKernel.list_tools(limit, cursor)	Cursor-paginated compact tool manifest
CatalystTokenKernel.discover(query, include_schema)	Query-gated ranking with optional schema expansion
CatalystTokenKernel.run_python_task(code)	Constrained local Python execution with compact task status
CatalystTokenKernel.create_code_execution_task(...)	Compact deferred task status, with output stored outside context
CatalystTokenKernel.fetch_task_result(task_id)	Explicitly retrieve full code/stdout/stderr
CatalystTokenKernel.export_rain_snapshot(agent_id)	Export a Rain header for compact agent state transfer

---

Benchmarks

Memory Footprint

Catalyst public state is constant in the SDK memory model; it does not grow with token count. This table is a footprint model, not a claim that the compact state alone is semantically equivalent to every model's internal past_key_values tensors.

Tokens	Standard FP16 KV-Cache	Catalyst public state	Reduction
1,000	655.36 MB	0.016 MB	40,000x
5,000	3,276.80 MB	0.016 MB	200,000x
10,000	6,553.60 MB	0.016 MB	400,000x

Live Hugging Face cache equivalence is proven in tensor-preserving mode through CatalystDynamicKVCache: logits, target perplexity, retrieval score, and greedy generation match baseline DynamicCache on the checked-in TinyLlama run.

For the frictionless replacement path, use CatalystKVAttentionReplacement. It pairs CatalystHolographicKVCache with the SDK's PyQuantumAttentionHead, replacing the unbounded growing HF tensor cache with a bounded local tensor window plus compact Catalyst Rain/HKVC state for evicted context:

from catalyst_brain import CatalystKVAttentionReplacement

attention = CatalystKVAttentionReplacement(max_tokens=128)
output = attention(query, key_states, value_states, layer_idx=layer_idx)

For cache-only interoperability with stock Hugging Face attention, CatalystCompactKVState is an HF-compatible past_key_values object for state-for-stateless handoff. Production long-context reconstruction is the Catalyst API path: the SDK sends the compact state to the Catalyst service, which performs closed-source holographic hydration behind quota and entitlement checks. Use CatalystAPIBackedKVCache when the stock HF cache interface needs tensor-shaped K/V outputs materialized through the Catalyst API on overflow. The gold-proof benchmark for this path requires the live /kv-cache/hydrate endpoint to return replacement key_states and value_states tensors for the requested target sequence shape; structured non-tensor responses are treated as proof failures, not successes.

For validation before compact-only semantic hydration is enabled, the cache also supports lossless_payload=True. That mode sends source K/V payloads to the Catalyst API with a stable hydration session ID, lets the API materialize tensor-shaped responses, and keeps the SDK-side cache bounded after each overflow. Benchmark reports label this as an API transport proof; the stronger compact gold proof remains separate and should only pass without lossless source tensor payloads.

Use lossless_payload_encoding="base64" to send K/V tensors in the base64_tensor_v1 envelope instead of nested JSON arrays during API transport validation:

cache = CatalystAPIBackedKVCache(
    max_tokens=128,
    lossless_payload=True,
    lossless_payload_encoding="base64",
)

For production compact-handle validation, use server-shadow mode. On overflow, the SDK ingests evicted K/V state into the managed Catalyst API once, then later hydrates with no source tensors in the hydration request. The local process retains only a bounded K/V window, Rain metadata, and a fixed-size complex-domain accumulator snapshot:

cache = CatalystAPIBackedKVCache(
    max_tokens=128,
    server_shadow_state=True,
    complex_accumulator_dim=256,
)

The public benchmark reports this separately as a server-backed compact-handle proof. It is distinct from pure fixed-vector reconstruction: source tensors are not sent during hydration, but managed Catalyst infrastructure retains the server-side state required to materialize exact K/V tensors for stock HF causal LMs.

Every hydration request carries Rain protocol metadata, a SHA-256 integrity tag for the compact state, and a deterministic idempotency token. Those fields are part of the production contract for replay-safe retries, quota-safe hydration, and future compact-only tensor materialization.

CatalystAPIBackedKVCache.get_seq_length() reports the logical context length needed by Hugging Face cache-position bookkeeping after eviction. get_local_seq_length() reports the retained tensor-window length used for memory accounting.

For adversarial long-context retrieval checks, use CatalystLongContextSecretMemory: the public benchmark places random 256-bit secrets at logical token 10 and retrieves them at logical token 1,000,000 through this SDK surface. For TinyLlama-style causal LMs, use CatalystTinyLlamaLongContextAdapter to carry far-context facts in Catalyst Brain state and materialize the recalled fact into the active prompt at logical positions up to 2,000,000 tokens.

For the transient-collapse research path, use TransientCollapseAttention. It implements the practical version of the transiency math: historical K/V state is enfolded into fixed phase banks, then queried by a conjugate collapse operator. The banked design explicitly acknowledges the finite-D noise floor from the transiency proof and lowers it by partitioning load plus signed collision cancellation. The checked-in benchmark artifact is docs/transient_collapse_results.json; charts are in docs/charts/.

from catalyst_brain import TransientCollapseAttention

attention = TransientCollapseAttention(key_dim=64, value_dim=64)
attention.enfold(key_vector, value_vector, position=token_position)
value = attention.collapse(query_vector, position=query_position)

In the current synthetic SDK benchmark, transient phase banks retrieve the old target in 100% of 12 trials, while the monolithic phase-field control retrieves 16.67%. This is operator evidence, not yet a model-level perplexity proof.

Bit-Exact Recovery

Bind/unbind is provably lossless — XOR is its own inverse.

Operation	Fidelity	Tested depth
BCV bind/unbind	100.00% bit-exact	1,000 trials
Chained composition (depth 2–100)	100.00% bit-exact	6 depths
.rain serialization	100.00% bit-exact	100 trials

Multi-Item Superposition

Multi-item bundling maintains 98.4% constant bit accuracy regardless of item count (up to ~7,213 items at D=10,000).

Performance Benchmarks

All reproducible public-wheel benchmarks are maintained in the public catalyst-brain-benchmarks repository:

https://github.com/CrewRiz/catalyst-brain-benchmarks

The suite installs catalyst-brain from PyPI and uses only public SDK APIs, so users can verify the published results without source access.

---

Package Distribution

pip install catalyst-brain
python -c "import catalyst_hdc as hdc; print(len(hdc.rand_bipolar(4096)))"

Release wheels are built with the CPython stable ABI (abi3-py39) so one wheel serves Python 3.9+ on the same platform. Each public release should include platform wheels for the supported operating systems, the native catalyst_hdc extension, the pure-Python catalyst_brain companion package, and type stubs. The free-tier PyPI release is wheel-first; source distributions are reserved for licensed source-access customers.

Source builds are not part of the public free tier. Production and enterprise customers can receive source access, private build instructions, or deployment support under separate terms.

---

Free Tier And Production Use

The public PyPI package is suitable for learning, academic experiments, local prototypes, benchmark reproduction, and early integration. It does not require registration, signup, or an API key to get started.

Most users should not hit free-tier limits early. When usage moves toward production, enterprise deployment, hosted APIs, higher quotas, redistribution, resale, revenue workflows, or customer pilots, use the production path.

For production access, enterprise evaluation, higher quotas, private support, or source-access discussions, contact:

[e-mail]

The public SDK never relies on client-side checks for quota or entitlement decisions. Account status, tenant quotas, and API-key authorization are enforced server-side in the managed Catalyst infrastructure. Operational deployment details, payment-provider configuration, and secret management are intentionally not part of the public PyPI documentation.

---

Architecture

catalyst-brain wheel
├── catalyst_hdc        # Native Rust/PyO3 extension
│   ├── Core HDC        # bind/unbind, bundle, permute, resonance
│   ├── HoloCPU         # O(1) scheduler + Grover search
│   ├── HoloGen         # Geometric encoding facade
│   ├── Quantum Heads   # Quantum-inspired attention primitives
│   ├── HKVC            # Holographic key-value cache
│   └── MetaLearning    # Metacognition, SelfAudit, Optimizer, LearningLog
├── catalyst_brain      # Pure-Python companion package
│   ├── Rain Protocol   # Binary state transfer and digest merge
│   ├── Token Kernel    # Progressive tool discovery and compact task state
│   ├── Client          # Edge-worker HTTP wrapper
│   └── Telemetry       # Anonymous, opt-out SDK health events
├── catalyst_hdc.pyi    # Python type stubs (PEP 561)
└── py.typed            # PEP 561 marker

---

Telemetry & Privacy

catalyst-brain collects anonymous usage data to help improve the SDK. No user data, vectors, labels, or model outputs are ever sent.

What is collected (all anonymous):

• SDK version, Python version, OS, CPU architecture
• A one-way hash of your machine's platform info (cannot be reversed)
• Which top-level feature was used and whether an exception occurred

Opt-out at any time:

export CATALYST_NO_TELEMETRY=1

Data is sent to a Cloudflare Worker endpoint over HTTPS in a background daemon thread and never blocks your code.

---

License

Closed-source SDK with a generous free tier — see LICENSE file for the free research and evaluation grant.

Use	Permitted?
Academic research	Free tier
Personal experimentation	Free tier
Benchmarking & evaluation	Free tier
Publishing results with attribution	Free tier
Production deployment	Contact for production path
SaaS / hosted API	Contact for production path
Redistribution or resale	Separate written permission required
Public source-code use	Not included in the free PyPI package

Patent: U.S. Provisional Patent Application CATALYST-2026-001 covers holographic key-value caching, BlockCodeVector binding, resonant superposition memory, and Grover-amplified attention routing.

Support: [e-mail]
Pilots and commercial access: [e-mail]

---

Copyright © 2026 Strategic Innovations AI. Built with Rust 🦀 + PyO3 🐍.