build(agent): jabba#56a767 iteration
This commit is contained in:
parent
6d39e5f8e0
commit
d8765a77a4
|
|
@ -39,3 +39,25 @@ class ToyBioreactorAdapter:
|
||||||
next_biomass = max(0.0, biomass + dt * (0.35 * substrate * biomass - self.decay * biomass - 0.03 * toxin + 0.02 * purge))
|
next_biomass = max(0.0, biomass + dt * (0.35 * substrate * biomass - self.decay * biomass - 0.03 * toxin + 0.02 * purge))
|
||||||
next_toxin = max(0.0, toxin + dt * (0.05 * biomass - self.toxin_relaxation * toxin - 0.06 * purge))
|
next_toxin = max(0.0, toxin + dt * (0.05 * biomass - self.toxin_relaxation * toxin - 0.06 * purge))
|
||||||
return (next_substrate, next_biomass, next_toxin)
|
return (next_substrate, next_biomass, next_toxin)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class PlantAdapterShim:
|
||||||
|
"""Lightweight shim that adapts existing step-style adapters to a
|
||||||
|
named PlantAdapter interface. This makes it easier to retarget compiled
|
||||||
|
attractors to different plant models without changing the compiler.
|
||||||
|
|
||||||
|
The shim simply forwards the step call to the underlying adapter but
|
||||||
|
provides a stable `name` attribute for logging/indexing.
|
||||||
|
"""
|
||||||
|
|
||||||
|
underlying: object
|
||||||
|
name: str | None = None
|
||||||
|
|
||||||
|
def step(self, state: State, control: Control, dt: float) -> State:
|
||||||
|
# forward to underlying object which is expected to implement
|
||||||
|
# step(state, control, dt)
|
||||||
|
return self.underlying.step(state, control, dt)
|
||||||
|
|
||||||
|
def __repr__(self) -> str: # pragma: no cover - trivial
|
||||||
|
return f"PlantAdapterShim(name={self.name or type(self.underlying).__name__})"
|
||||||
|
|
|
||||||
|
|
@ -77,6 +77,33 @@ def _fingerprint(spec: AttractorSpec, gain: tuple[float, ...], controller_hash:
|
||||||
return _hash_text(payload)
|
return _hash_text(payload)
|
||||||
|
|
||||||
|
|
||||||
|
def generate_attractor_fingerprint(
|
||||||
|
spec: AttractorSpec,
|
||||||
|
gain: tuple[float, ...],
|
||||||
|
controller_hash: str,
|
||||||
|
certificate_hash: str | None = None,
|
||||||
|
sample_call_len: int | None = None,
|
||||||
|
) -> str:
|
||||||
|
"""Public fingerprint generator for compact DTN admission.
|
||||||
|
|
||||||
|
The fingerprint is a deterministic SHA-256 hex of a small JSON payload
|
||||||
|
containing a few stable feature bits. The certificate_hash and a small
|
||||||
|
sample_call_len may be included when available to increase collision
|
||||||
|
resistance while keeping the output compact (64 hex chars).
|
||||||
|
"""
|
||||||
|
payload = {
|
||||||
|
"spec_id": spec.spec_id,
|
||||||
|
"kind": spec.kind,
|
||||||
|
"gain_len": len(gain),
|
||||||
|
"ctrl_hash": controller_hash,
|
||||||
|
}
|
||||||
|
if certificate_hash:
|
||||||
|
payload["cert"] = certificate_hash[:16]
|
||||||
|
if sample_call_len is not None:
|
||||||
|
payload["calls"] = int(sample_call_len)
|
||||||
|
return _hash_text(_json(payload))
|
||||||
|
|
||||||
|
|
||||||
class AttractorCompiler:
|
class AttractorCompiler:
|
||||||
def compile(self, spec: AttractorSpec) -> CompilationResult:
|
def compile(self, spec: AttractorSpec) -> CompilationResult:
|
||||||
spec = spec.normalized()
|
spec = spec.normalized()
|
||||||
|
|
@ -90,7 +117,13 @@ class AttractorCompiler:
|
||||||
estimated_basin_radius=spec.basin_radius,
|
estimated_basin_radius=spec.basin_radius,
|
||||||
convergence_rate_bound=spec.convergence_rate,
|
convergence_rate_bound=spec.convergence_rate,
|
||||||
controller_hash=_hash_text(_json({"gain": gain, "target": spec.target_state, "kind": spec.kind})),
|
controller_hash=_hash_text(_json({"gain": gain, "target": spec.target_state, "kind": spec.kind})),
|
||||||
attractor_fingerprint=_fingerprint(spec, gain, _hash_text(_json({"gain": gain, "target": spec.target_state, "kind": spec.kind}))),
|
attractor_fingerprint=generate_attractor_fingerprint(
|
||||||
|
spec,
|
||||||
|
gain,
|
||||||
|
_hash_text(_json({"gain": gain, "target": spec.target_state, "kind": spec.kind})),
|
||||||
|
certificate_hash=plan_cert.certificate_hash,
|
||||||
|
sample_call_len=len(plan_delta.steps),
|
||||||
|
),
|
||||||
dt=spec.dt,
|
dt=spec.dt,
|
||||||
)
|
)
|
||||||
return CompilationResult(controller=controller, plan_cert=plan_cert, plan_delta=plan_delta, basin_sketch=basin_sketch)
|
return CompilationResult(controller=controller, plan_cert=plan_cert, plan_delta=plan_delta, basin_sketch=basin_sketch)
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,38 @@
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import Tuple
|
||||||
|
|
||||||
|
from .models import PlanCert
|
||||||
|
|
||||||
|
|
||||||
|
def verify(cert: PlanCert, seed_id: str) -> Tuple[bool, float]:
|
||||||
|
"""Prototype PlanCert verifier.
|
||||||
|
|
||||||
|
This is a tiny, deterministic verifier intended as a lightweight
|
||||||
|
on-device check. It is NOT a formal SMT/Lyapunov verifier; rather it
|
||||||
|
performs sanity checks on stable fields and returns a simple damage
|
||||||
|
score in [0,1] (0==low risk, 1==high risk) derived from the
|
||||||
|
robustness_margin and worst_case_cost.
|
||||||
|
|
||||||
|
Returns: (ok, damage_score)
|
||||||
|
"""
|
||||||
|
# Basic structural checks
|
||||||
|
ok = True
|
||||||
|
if not isinstance(cert.certificate_hash, str) or len(cert.certificate_hash) != 64:
|
||||||
|
ok = False
|
||||||
|
if not isinstance(cert.witness_hash, str) or len(cert.witness_hash) != 64:
|
||||||
|
ok = False
|
||||||
|
if not isinstance(cert.merkle_root, str) or len(cert.merkle_root) == 0:
|
||||||
|
ok = False
|
||||||
|
|
||||||
|
# Damage heuristic: higher robustness margin relative to worst-case cost
|
||||||
|
# yields lower damage_score. Clamp to [0,1]. Add small epsilon to avoid div0.
|
||||||
|
try:
|
||||||
|
ratio = cert.robustness_margin / (cert.worst_case_cost + 1e-12)
|
||||||
|
damage = max(0.0, 1.0 - float(ratio))
|
||||||
|
except Exception:
|
||||||
|
damage = 1.0
|
||||||
|
|
||||||
|
# Ensure damage in [0,1]
|
||||||
|
damage_score = max(0.0, min(1.0, damage))
|
||||||
|
return ok, damage_score
|
||||||
|
|
@ -0,0 +1,41 @@
|
||||||
|
from idea194_attractorforge_verified_attractor import AttractorCompiler
|
||||||
|
from idea194_attractorforge_verified_attractor.adapters import ToyOrbitalAdapter, PlantAdapterShim
|
||||||
|
from idea194_attractorforge_verified_attractor.verifier import verify
|
||||||
|
|
||||||
|
|
||||||
|
def test_plant_adapter_shim_forwards_step() -> None:
|
||||||
|
adapter = ToyOrbitalAdapter()
|
||||||
|
shim = PlantAdapterShim(underlying=adapter, name="orbital-shim")
|
||||||
|
state = (0.5, -0.5)
|
||||||
|
control = (0.1, -0.1)
|
||||||
|
next1 = adapter.step(state, control, 0.1)
|
||||||
|
next2 = shim.step(state, control, 0.1)
|
||||||
|
assert next1 == next2
|
||||||
|
|
||||||
|
|
||||||
|
def test_verifier_returns_ok_and_damage_score() -> None:
|
||||||
|
spec = AttractorCompiler().compile(
|
||||||
|
AttractorCompiler().compile.__defaults__[0]
|
||||||
|
) if False else None
|
||||||
|
|
||||||
|
# Instead of complex construction, compile a simple spec via known pattern
|
||||||
|
from idea194_attractorforge_verified_attractor import AttractorSpec
|
||||||
|
|
||||||
|
spec = AttractorSpec(
|
||||||
|
spec_id="v-demo",
|
||||||
|
kind="motion_manifold",
|
||||||
|
state_labels=("r", "t"),
|
||||||
|
target_state=(0.0, 0.0),
|
||||||
|
control_limit=0.5,
|
||||||
|
dt=0.1,
|
||||||
|
horizon=8,
|
||||||
|
basin_radius=0.5,
|
||||||
|
convergence_rate=0.2,
|
||||||
|
safety_margin=0.05,
|
||||||
|
objective="verify-demo",
|
||||||
|
)
|
||||||
|
|
||||||
|
result = AttractorCompiler().compile(spec)
|
||||||
|
ok, damage = verify(result.plan_cert, result.plan_delta.spec_id)
|
||||||
|
assert isinstance(ok, bool)
|
||||||
|
assert 0.0 <= damage <= 1.0
|
||||||
Loading…
Reference in New Issue