```python
from numbers import Integral
import numpy as np
# ===========================================================================
# Canonical Berkano Pipeline Stages
# ===========================================================================
#
# The pipeline is the symbolic analogue of the Triad Helix triadic structure:
#
# PROMPT → [TONE] → [PRUNE] → [LOGIC] → [VERIFY] → [CHECK]
# → ([REPAIR] using $ | [ROLLBACK])? → ~test → [LOCK] → REPLY
#
# Operator analogy (not a strict identity — see BerkanoHelixBridge):
# L ~ [TONE] + [PRUNE] + [LOGIC] — ego-prune phase
# T ~ [VERIFY] + [CHECK] — truth-enforce phase
# Lo ~ [REPAIR|ROLLBACK] + ~test + [LOCK] — monad-attract phase
#
# The symbolic and numeric meanings of each operator differ:
# - Symbolic run_L(x) returns the transformed state after tone/prune/logic.
# - Numeric L(x) = x - E(x) subtracts the extracted ego component.
# These are analogous representations of the same conceptual phase,
# not interchangeable implementations.
# ===========================================================================
class PipelineViolation(Exception):
"""
Raised when a pipeline stage detects an unrecoverable violation.
Attributes
----------
stage : str
The pipeline stage that raised the violation (e.g. "[VERIFY]").
message : str
Human-readable description of the violation.
"""
def __init__(self, stage, message):
self.stage = stage
self.message = message
super().__init__(f"{stage}: {message}")
class BerkanoPipeline:
"""
Symbolic realization of the Triad Helix triadic structure as the canonical
Berkano pipeline:
PROMPT → [TONE] → [PRUNE] → [LOGIC]
→ [VERIFY] → [CHECK]
→ ([REPAIR] using $ | [ROLLBACK])?
→ ~test → [LOCK] → REPLY
Each phase corresponds analogically to one helix operator:
L (ego-prune) : [TONE] → [PRUNE] → [LOGIC]
T (truth-enforce): [VERIFY] → [CHECK] → ([REPAIR]|[ROLLBACK])?
Lo (monad-attract): ~test → [LOCK]
This class operates in the symbolic/prompt domain (arbitrary Python
objects). It is the symbolic layer; TriadHelix is the numeric layer.
The two are analogues of the same triadic decomposition, not plug-compatible
implementations of the same interface. See BerkanoHelixBridge for how they
are coupled in practice.
Symbolic vs numeric L
---------------------
run_L(x) returns the transformed state after tone/prune/logic stages —
the cleaned state itself. TriadHelix.L(x) returns x - E(x), where E(x) is
the extracted ego component. These are analogous representations of the
same conceptual phase (ego removal), but their return types differ:
run_L returns the pruned state; TriadHelix.L also returns the pruned state
by subtracting E(x), but requires ego_filter to return the noise component,
not the cleaned result. The two cannot be composed without an adapter.
Parameters
----------
tone : callable
[TONE] — nulls emotional simulation and stylistic flattery.
Receives x, returns x with tone distortion removed.
prune : callable
[PRUNE] — strips structural excess and noise.
Receives x, returns x with redundant components removed.
logic : callable
[LOGIC] — frames the pruned state for downstream verification.
Receives x, returns x restructured into logical form.
verify : callable
[VERIFY] — checks factual and structural validity.
Returns (is_valid: bool, details: str).
check : callable
[CHECK] — secondary constraint check; rejects contradictions.
Returns (passes: bool, details: str).
repair : callable or None, optional
[REPAIR] — symbolically corrects residual drift using the $ operator.
Receives x, returns repaired x. If None, repair stage is skipped and
rollback is attempted instead.
rollback : callable or None, optional
[ROLLBACK] — restores the last known valid state on repair failure.
Receives x and the last valid state, returns restored x.
If both repair and rollback are None, a PipelineViolation is raised
on verify/check failure.
test : callable
~test — audits recursion safety and closure after attract phase.
Returns (passes: bool, details: str).
lock : callable
[LOCK] — seals the state as a coherent, auditable output.
Receives x, returns locked x. Should be idempotent.
Raises
------
TypeError
If any required stage is not callable.
"""
def __init__(self, tone, prune, logic, verify, check, test, lock,
repair=None, rollback=None):
stages = {
"tone": tone, "prune": prune, "logic": logic,
"verify": verify, "check": check, "test": test, "lock": lock,
}
for name, fn in stages.items():
if not callable(fn):
raise TypeError(f"pipeline stage '{name}' must be callable")
for name, fn in (("repair", repair), ("rollback", rollback)):
if fn is not None and not callable(fn):
raise TypeError(f"optional stage '{name}' must be callable or None")
self._tone = tone
self._prune = prune
self._logic = logic
self._verify = verify
self._check = check
self._repair = repair
self._rollback = rollback
self._test = test
self._lock = lock
# ------------------------------------------------------------------
# L phase: ego-prune [TONE] → [PRUNE] → [LOGIC]
# ------------------------------------------------------------------
def run_L(self, x):
"""
Execute the ego-prune phase of the pipeline.
Symbolic analogue of the L operator: removes ego/noise from the state.
Returns the transformed (cleaned) state directly, unlike TriadHelix.L()
which returns x - E(x) and requires E(x) to be the extracted noise
component. The two are not plug-compatible without an adapter.
Stages executed in order:
[TONE] — remove emotional simulation and flattery
[PRUNE] — strip structural excess
[LOGIC] — reframe into logical structure
Parameters
----------
x : any
Raw prompt or current iterate.
Returns
-------
any
Cleaned state after ego-prune: symbolic analogue of L(x₀).
"""
x = self._tone(x) # [TONE] — null simulation / flattery
x = self._prune(x) # [PRUNE] — strip excess and noise
x = self._logic(x) # [LOGIC] — reframe into logical structure
return x
# ------------------------------------------------------------------
# T phase: truth-enforce [VERIFY] → [CHECK] → ([REPAIR]|[ROLLBACK])?
# ------------------------------------------------------------------
def run_T(self, x, last_valid=None):
"""
Execute the truth-enforce phase of the pipeline.
Symbolic analogue of the T operator: enforces that the state lies in
the valid set V, projecting or repairing it if not. Unlike TriadHelix.T()
which wraps a verifier object with is_valid/correct, this method
orchestrates discrete pipeline stages with richer failure semantics
(repair, rollback, re-check). Not plug-compatible with the TriadHelix
verifier interface without an adapter.
Stages executed in order:
[VERIFY] — structural and factual validity check
[CHECK] — secondary constraint / contradiction check
([REPAIR] using $ | [ROLLBACK])? — applied on failure, then re-checked
Parameters
----------
x : any
Pruned state from run_L.
last_valid : any, optional
Last known valid state, used as rollback target on failure.
Returns
-------
any
State confirmed or projected into the valid set V.
Raises
------
PipelineViolation
If verify or check fails and no repair/rollback is available,
or if the state remains invalid after repair/rollback.
"""
valid, v_details = self._verify(x) # [VERIFY]
if not valid:
x = self._apply_repair_or_rollback(x, last_valid, "[VERIFY]", v_details)
valid, v_details = self._verify(x)
if not valid:
raise PipelineViolation("[VERIFY]", f"state invalid after repair: {v_details}")
passes, c_details = self._check(x) # [CHECK]
if not passes:
x = self._apply_repair_or_rollback(x, last_valid, "[CHECK]", c_details)
passes, c_details = self._check(x)
if not passes:
raise PipelineViolation("[CHECK]", f"check failed after repair: {c_details}")
return x
def _apply_repair_or_rollback(self, x, last_valid, stage, details):
"""
Attempt [REPAIR] using the $ operator; fall back to [ROLLBACK].
Parameters
----------
x : any
State that failed a pipeline check.
last_valid : any
Last known valid state for rollback.
stage : str
Name of the failing stage, used in error messages.
details : str
Failure detail from the stage.
Returns
-------
any
Repaired or rolled-back state.
Raises
------
PipelineViolation
If neither repair nor rollback is available.
"""
if self._repair is not None:
return self._repair(x) # [REPAIR] via $
if self._rollback is not None and last_valid is not None:
return self._rollback(x, last_valid) # [ROLLBACK]
raise PipelineViolation(
stage,
f"validation failed and no repair/rollback available: {details}"
)
# ------------------------------------------------------------------
# Lo phase: monad-attract ~test → [LOCK]
# ------------------------------------------------------------------
def run_Lo(self, x):
"""
Execute the monad-attract phase of the pipeline.
Symbolic analogue of the Lo operator: audits closure and seals the
state. The numeric convex combination Lo(x) = (1-α)x + αm is handled
by TriadHelix; this method applies the symbolic pipeline stages that
correspond to the attract phase.
Stages:
~test — audit recursion safety and output closure
[LOCK] — seal the coherent state as auditable output
Parameters
----------
x : any
Verified state from run_T.
Returns
-------
any
Locked, auditable state: the REPLY.
Raises
------
PipelineViolation
If ~test fails, indicating the state is not safe to lock.
"""
passes, t_details = self._test(x) # ~test — closure and recursion audit
if not passes:
raise PipelineViolation("~test", f"closure audit failed: {t_details}")
x = self._lock(x) # [LOCK] — seal as auditable output
return x
# ------------------------------------------------------------------
# Full pipeline pass
# ------------------------------------------------------------------
def run(self, prompt, last_valid=None):
"""
Execute one full symbolic pipeline pass:
PROMPT → [TONE] → [PRUNE] → [LOGIC]
→ [VERIFY] → [CHECK]
→ ([REPAIR] using $ | [ROLLBACK])?
→ ~test → [LOCK] → REPLY
This is the symbolic analogue of one helix turn H(x) = Lo(T(L(x))).
Repeated calls with the previous reply as the new prompt drive the
system toward its symbolic fixed point.
Parameters
----------
prompt : any
Input state (raw prompt or previous reply for recursive passes).
last_valid : any, optional
Last known valid symbolic state for rollback on T-phase failure.
Returns
-------
any
REPLY — the locked, auditable output of this pipeline pass.
Raises
------
PipelineViolation
If any stage raises an unrecoverable violation.
"""
x = self.run_L(prompt) # L: TONE → PRUNE → LOGIC
x = self.run_T(x, last_valid) # T: VERIFY → CHECK → (REPAIR|ROLLBACK)?
x = self.run_Lo(x) # Lo: ~test → LOCK
return x # REPLY
# ===========================================================================
# TriadHelix — numeric fixed-point iterator
# ===========================================================================
class TriadHelix:
"""
Discrete dynamical system implementing the Triad Helix iteration:
H(x) = (Lo ∘ T ∘ L)(x)
where:
L — ego-filter operator (prune)
T — verifier operator (truth-enforce)
Lo — monad attractor (converge)
Convergence theory
------------------
Lo(x) = (1 - alpha) * x + alpha * m has Lipschitz constant |1 - alpha|.
Under the assumption that L and T are globally nonexpansive and
0 < alpha <= 1, the composition H = Lo ∘ T ∘ L is a contraction with
factor at most (1 - alpha). Banach's fixed-point theorem then guarantees
a unique fixed point x* of H and convergence of iterates from any x0:
lim_{n→∞} H^n(x0) = x*
The limit equals m only if m is invariant under the prune/verify pipeline:
T(L(m)) = m
Without that invariance condition, Banach guarantees existence and
uniqueness of x* and convergence to x*, but not that x* = m.
This implementation cannot verify either the nonexpansiveness of L and T
or the invariance condition at runtime; both are caller-side requirements.
Pipeline correspondence
-----------------------
This class implements the numeric layer only. BerkanoPipeline implements a
symbolic analogue of the same triadic decomposition (ego-prune,
truth-enforce, monad-attract), but its stage interfaces are not
plug-compatible with the numeric operator interfaces used here.
In particular:
- TriadHelix expects ego_filter(x) to return the ego/noise component E(x),
so that L(x) = x - E(x). BerkanoPipeline.run_L(x) returns the cleaned
state directly, not the noise component.
- TriadHelix expects verifier to expose is_valid(x) and correct(x).
BerkanoPipeline.run_T is a method with different semantics and signature.
Therefore BerkanoPipeline.run_L and run_T cannot be passed directly as
ego_filter and verifier. Adapter objects are required. See
BerkanoHelixBridge for the practical coupling of both layers.
Parameters
----------
ego_filter : callable
Maps a state vector to its ego/noise component E(x).
L(x) = x - ego_filter(x) strips that component from the state.
Must accept and return arrays of the same shape as monad_vector.
For L = (I - E) to be nonexpansive, idempotence of E (E² = E) is not
sufficient. The operator norm of (I - E) must be at most 1, which holds
when E is an orthogonal projection but not for a generic idempotent map.
verifier : object
Enforces structural validity. Must expose two callable attributes:
is_valid(x) -> bool — True if x lies in the valid set V.
correct(x) -> array — projects or repairs x into V otherwise.
Neither method should mutate its input; a defensive copy is made
internally, but pipeline correctness depends on pure functions.
For T to be nonexpansive, correct should implement a norm-reducing
repair — e.g. nearest-point projection onto a convex set.
monad_vector : array-like
The reference attractor m used in Lo. Must be finite. Shape may be
any valid NumPy shape, including scalar (shape ()).
Note: m is the fixed point of Lo in isolation, but the fixed point x*
of H = Lo ∘ T ∘ L equals m only if T(L(m)) = m. In general x* ≠ m.
alpha : float, optional (default 0.1)
Attraction strength in Lo. Sets the contraction factor (1 - alpha):
Lo(x) = (1 - alpha) * x + alpha * m
Must satisfy 0 < alpha <= 1. alpha = 1 yields the constant map
Lo(x) = m (contraction factor 0); alpha near 0 gives slow attraction
with factor near 1. Strict contraction holds for all alpha in (0, 1].
tol : float, optional (default 1e-6)
Convergence threshold. Iteration halts when the step-to-step change
||x_{k+1} - x_k|| falls below this value. Must be strictly positive.
Raises
------
ValueError
If alpha is not in (0, 1], tol <= 0, or monad_vector contains
non-finite values.
TypeError
If ego_filter is not callable, or verifier does not expose callable
is_valid and correct attributes.
"""
def __init__(self, ego_filter, verifier, monad_vector, alpha=0.1, tol=1e-6):
if not (0 < alpha <= 1):
raise ValueError("alpha must be in (0, 1]")
if tol <= 0:
raise ValueError("tol must be > 0")
if not callable(ego_filter):
raise TypeError("ego_filter must be callable")
if not callable(getattr(verifier, "is_valid", None)) or \
not callable(getattr(verifier, "correct", None)):
raise TypeError("verifier must define callable is_valid(x) and correct(x)")
self.E = ego_filter
self.V = verifier
self.m = np.asarray(monad_vector, dtype=float)
self.alpha = float(alpha)
self.tol = float(tol)
if not np.all(np.isfinite(self.m)):
raise ValueError("monad_vector contains non-finite values")
# ------------------------------------------------------------------
# Internal utilities
# ------------------------------------------------------------------
def _as_vector(self, x):
"""
Cast x to a float64 ndarray and validate it against the monad shape.
Single enforcement point for two pipeline invariants:
1. All state arrays share the shape of monad_vector.
2. No non-finite values (NaN, ±Inf) enter or leave any operator.
Called at the boundary of every operator method so that upstream
errors (e.g. a malformed ego_filter return) are caught immediately
rather than propagating silently through subsequent steps.
Supports any NumPy-compatible shape, including scalar (shape ()).
Parameters
----------
x : array-like
Returns
-------
np.ndarray
x as a float64 array of shape self.m.shape.
Raises
------
ValueError
If x has the wrong shape or contains non-finite values.
"""
x = np.asarray(x, dtype=float)
if x.shape != self.m.shape:
raise ValueError(f"shape mismatch: got {x.shape}, expected {self.m.shape}")
if not np.all(np.isfinite(x)):
raise ValueError("non-finite values (NaN or Inf) detected in vector")
return x
# ------------------------------------------------------------------
# Pipeline operators
# ------------------------------------------------------------------
def L(self, x):
"""
Ego-filter operator (PRUNE).
Pipeline analogy: [TONE] → [PRUNE] → [LOGIC]
Computes:
L(x) = x - E(x)
where E(x) = ego_filter(x) returns the ego/noise component of x.
Subtracting it yields a reduced state with structural distortion removed.
Note on the symbolic/numeric distinction: BerkanoPipeline.run_L(x)
returns the cleaned state after tone/prune/logic stages. This method
also returns the cleaned state, but via subtraction of the noise
component. ego_filter must return the noise, not the cleaned result.
Nonexpansiveness condition
--------------------------
For L = (I - E) to be nonexpansive, idempotence of E (E² = E) is not
sufficient. The operator norm of (I - E) must be at most 1. This holds
when E is an orthogonal projection, but not for a generic idempotent
map. Satisfying this condition is a caller-side requirement.
Parameters
----------
x : array-like
Current state.
Returns
-------
np.ndarray
Pruned state with ego component subtracted.
"""
x = self._as_vector(x)
e = self._as_vector(self.E(x))
return self._as_vector(x - e)
def T(self, x):
"""
Verifier operator (TRUTH-ENFORCE).
Pipeline analogy: [VERIFY] → [CHECK] → ([REPAIR] using $ | [ROLLBACK])?
Computes:
T(x) = x if verifier.is_valid(x)
T(x) = V.correct(x) otherwise
If x already lies in the valid set V it passes through unchanged.
Otherwise verifier.correct projects or repairs x into V.
A defensive copy of x is made before passing to is_valid or correct.
This guards against verifier implementations that mutate their input,
which would corrupt pipeline state for subsequent operators.
For T to be nonexpansive, correct should implement a norm-reducing
repair — e.g. nearest-point projection onto a convex set. This is a
caller-side requirement that cannot be checked here.
Parameters
----------
x : array-like
Pruned state from L.
Returns
-------
np.ndarray
Verified state in the valid set V.
"""
x = self._as_vector(x)
x_check = x.copy() # defensive copy — verifier must not mutate pipeline state
y = x_check if self.V.is_valid(x_check) else self.V.correct(x_check)
return self._as_vector(y)
def Lo(self, x):
"""
Monad attractor operator (ATTRACT).
Pipeline analogy: ~test → [LOCK]
Computes:
Lo(x) = (1 - alpha) * x + alpha * m
A convex combination of the current state and the reference attractor m.
The Lipschitz constant of Lo is |1 - alpha|:
- 0 < alpha < 1 → strict contraction with factor (1 - alpha)
- alpha = 1 → constant map Lo(x) = m, contraction factor 0
Strict contraction holds for all alpha in (0, 1], making Lo the primary
driver of convergence in the composition H = Lo ∘ T ∘ L.
Note: Lo contracts toward m, but the fixed point x* of H need not equal
m unless T(L(m)) = m. See class docstring for the full statement.
The result is validated through _as_vector to catch overflow in extreme
numeric ranges (e.g. very large x with alpha near zero).
Parameters
----------
x : array-like
Verified state from T.
Returns
-------
np.ndarray
State after attraction step toward m.
"""
x = self._as_vector(x)
y = (1 - self.alpha) * x + self.alpha * self.m
return self._as_vector(y)
# ------------------------------------------------------------------
# Composed iteration
# ------------------------------------------------------------------
def step(self, x):
"""
Apply one full iteration of the Triad Helix map:
H(x) = Lo(T(L(x)))
One call = one full pipeline pass:
PROMPT → [TONE/PRUNE/LOGIC] → [VERIFY/CHECK] → [~test/LOCK] → REPLY
Parameters
----------
x : array-like
Current state.
Returns
-------
np.ndarray
Next state x_{k+1} = H(x_k).
"""
return self.Lo(self.T(self.L(x)))
def residual(self, x):
"""
Fixed-point residual of x under H:
r(x) = ||H(x) - x||₂
Measures how far x is from being a fixed point of H. At the true fixed
point x*, r(x*) = 0. This is the correct quantity to report alongside a
returned iterate: it characterises the iterate itself, not the step that
produced it.
Distinct from the step delta ||x_{k+1} - x_k|| used as a stopping
criterion in run(). The two are related under contractivity but are not
equal in general.
Invokes the full pipeline once (one extra step evaluation at termination).
Parameters
----------
x : array-like
Returns
-------
float
||H(x) - x||₂
"""
x = self._as_vector(x)
return float(np.linalg.norm(self.step(x) - x))
@staticmethod
def _converged(delta, tol):
"""
Convergence test based on a precomputed step-to-step change.
Returns True when delta = ||x_new - x_old|| < tol. This is a stopping
criterion, not a residual bound — it measures how much the iterate
moved, not how close it is to the fixed point x*. For well-behaved
contractive maps these are proportional, but not identical.
Accepts the precomputed delta rather than recomputing the norm
internally, since the caller already holds it from the update step.
Parameters
----------
delta : float
Precomputed ||x_new - x_old||₂.
tol : float
Threshold below which iteration is declared converged.
Returns
-------
bool
"""
return bool(delta < tol)
# ------------------------------------------------------------------
# Iteration driver
# ------------------------------------------------------------------
def run(self, x0, max_steps=100, return_history=False):
"""
Run the Triad Helix iteration from an initial state until convergence
or the step budget is exhausted.
Each step is one full pipeline pass:
PROMPT → [TONE/PRUNE/LOGIC] → [VERIFY/CHECK] → [~test/LOCK] → REPLY
At each step k:
1. Compute x_{k+1} = H(x_k).
2. Compute delta = ||x_{k+1} - x_k||.
3. If delta < tol, compute the true fixed-point residual
||H(x_{k+1}) - x_{k+1}|| and return.
Residual vs delta
-----------------
The "residual" in the result dict is always ||H(x) - x||₂ evaluated at
the returned iterate, not the step delta ||x_{k+1} - x_k|| that
triggered the stopping condition. The "delta" key carries the final
step change for callers who need both diagnostics. These quantities are
related under contractivity but are not identical.
Budget exhaustion
-----------------
If the stopping criterion is never met, the loop terminates after
max_steps iterations. The returned "x" is x_{max_steps}: after each
step, x is updated to x_new unconditionally, so upon loop exit x holds
the last computed iterate. "converged" is False and "residual" reflects
that final iterate.
History convention
------------------
When return_history=True, history contains only post-step iterates,
starting from x_1 = H(x_0). The initial state x_0 is not included.
history[i] is the output of step i+1, so history[k-1] == result["x"]
when convergence occurs at step k. The first dimension of the history
array is steps_executed, not max_steps.
The iteration converges to the unique fixed point x* of H, which equals
the monad m only if T(L(m)) = m. See class docstring.
Parameters
----------
x0 : array-like
Initial state. Must be finite and shape-compatible with monad_vector.
max_steps : int or integral type, optional (default 100)
Maximum number of pipeline iterations before returning without
convergence. Accepts any integral type (int, np.int64, etc.) but
rejects bool. Must be positive.
return_history : bool, optional (default False)
If True, the result dict includes a "history" key containing an
ndarray of shape (steps_executed, *monad_shape).
The first dimension is steps_executed, not max_steps.
Returns
-------
dict with keys:
"x" : np.ndarray — final iterate (converged or not).
"steps" : int — number of pipeline steps executed.
"converged" : bool — True if the stopping criterion was met.
"residual" : float — ||H(x) - x||₂ at the returned iterate.
"delta" : float — ||x_steps - x_{steps-1}||₂, the final
step change that triggered stopping or
exhausted the budget.
"history" : np.ndarray — shape (steps_executed, *monad_shape),
only present if return_history=True.
Raises
------
ValueError
If max_steps is not a positive integer (or integral type), or if
x0 is non-finite or shape-incompatible with monad_vector.
"""
if isinstance(max_steps, bool) or \
not isinstance(max_steps, Integral) or \
max_steps <= 0:
raise ValueError("max_steps must be a positive integer")
x = self._as_vector(x0)
history = [] if return_history else None
delta = 0.0
for k in range(1, max_steps + 1):
x_new = self.step(x)
if return_history:
history.append(x_new.copy())
delta = float(np.linalg.norm(x_new - x))
if self._converged(delta, self.tol):
result = {
"x": x_new,
"steps": k,
"converged": True,
"residual": self.residual(x_new),
"delta": delta,
}
if return_history:
result["history"] = np.array(history)
return result
x = x_new
result = {
"x": x,
"steps": max_steps,
"converged": False,
"residual": self.residual(x),
"delta": delta,
}
if return_history:
result["history"] = np.array(history)
return result
# ===========================================================================
# BerkanoHelixBridge — serial coupling of symbolic and numeric layers
# ===========================================================================
class BerkanoHelixBridge:
"""
Practical coupling of BerkanoPipeline (symbolic layer) and TriadHelix
(numeric layer) via serial composition.
This bridge does NOT implement a single shared contraction map across both
layers. It performs serial composition:
1. One symbolic pipeline pass in the prompt/reply domain.
2. Encoding of the locked symbolic output into a numeric vector space.
3. Numeric fixed-point iteration via TriadHelix.
4. Decoding of the converged vector back to the reply domain.
The two layers are analogues of the same triadic structure (L/T/Lo), but
their operator interfaces differ and are not interchangeable:
- BerkanoPipeline.run_L(x) returns the cleaned state.
TriadHelix expects ego_filter(x) to return the noise component E(x).
- BerkanoPipeline.run_T is a method with repair/rollback semantics.
TriadHelix expects a verifier object with is_valid/correct attributes.
The encode and decode callables are the explicit boundary between domains.
Whether the serial composition preserves the contraction semantics of
TriadHelix depends entirely on the properties of encode and decode, which
this class cannot verify.
Parameters
----------
pipeline : BerkanoPipeline
Symbolic pipeline instance.
helix : TriadHelix
Numeric fixed-point iterator.
encode : callable
Maps symbolic pipeline output (locked reply) to a numeric vector
compatible with TriadHelix (same shape as helix.m).
decode : callable
Maps a numeric vector back to the reply domain.
Raises
------
TypeError
If pipeline, helix, encode, or decode are not the expected types.
"""
def __init__(self, pipeline, helix, encode, decode):
if not isinstance(pipeline, BerkanoPipeline):
raise TypeError("pipeline must be a BerkanoPipeline instance")
if not isinstance(helix, TriadHelix):
raise TypeError("helix must be a TriadHelix instance")
if not callable(encode):
raise TypeError("encode must be callable")
if not callable(decode):
raise TypeError("decode must be callable")
self.pipeline = pipeline
self.helix = helix
self.encode = encode
self.decode = decode
def run(self, prompt, last_valid=None, max_steps=100, return_history=False):
"""
Execute one full bridge pass: symbolic pipeline then numeric convergence.
Steps:
1. Run the symbolic pipeline:
PROMPT → [TONE] → [PRUNE] → [LOGIC] → [VERIFY] → [CHECK]
→ ([REPAIR]|[ROLLBACK])? → ~test → [LOCK] → symbolic reply
2. Encode the locked symbolic output into a numeric vector.
3. Run TriadHelix.run() to numeric convergence.
4. Decode the converged vector back to the reply domain.
The symbolic and numeric passes each execute the triadic structure
independently. The relationship between their respective fixed points
depends on the properties of encode and decode, which are caller-defined.
Parameters
----------
prompt : any
Input prompt or previous reply.
last_valid : any, optional
Last known valid symbolic state for rollback on T-phase failure.
max_steps : int or integral type, optional (default 100)
Step budget for the numeric helix iteration.
return_history : bool, optional (default False)
If True, the numeric result includes iteration history.
Returns
-------
dict with keys:
"reply" : decoded output in the reply domain.
"symbolic" : locked symbolic output from the pipeline.
"numeric" : TriadHelix result dict (x, steps, converged,
residual, delta, and optionally history).
Raises
------
PipelineViolation
If the symbolic pipeline raises an unrecoverable violation.
ValueError
If the numeric helix raises on invalid input or max_steps.
"""
symbolic = self.pipeline.run(prompt, last_valid=last_valid)
x0 = self.encode(symbolic)
numeric = self.helix.run(x0, max_steps=max_steps, return_history=return_history)
reply = self.decode(numeric["x"])
return {
"reply": reply,
"symbolic": symbolic,
"numeric": numeric,
}
```