build(agent): c3po#b883b4 iteration
This commit is contained in:
parent
9ef1d07cc0
commit
669f46bb55
|
|
@ -0,0 +1,21 @@
|
||||||
|
node_modules/
|
||||||
|
.npmrc
|
||||||
|
.env
|
||||||
|
.env.*
|
||||||
|
__tests__/
|
||||||
|
coverage/
|
||||||
|
.nyc_output/
|
||||||
|
dist/
|
||||||
|
build/
|
||||||
|
.cache/
|
||||||
|
*.log
|
||||||
|
.DS_Store
|
||||||
|
tmp/
|
||||||
|
.tmp/
|
||||||
|
__pycache__/
|
||||||
|
*.pyc
|
||||||
|
.venv/
|
||||||
|
venv/
|
||||||
|
*.egg-info/
|
||||||
|
.pytest_cache/
|
||||||
|
READY_TO_PUBLISH
|
||||||
|
|
@ -0,0 +1,30 @@
|
||||||
|
# Repository Guide
|
||||||
|
|
||||||
|
## Architecture
|
||||||
|
|
||||||
|
- `src/idea194_attractorforge_verified_attractor/models.py` defines the DSL and artifact objects.
|
||||||
|
- `src/idea194_attractorforge_verified_attractor/adapters.py` contains deterministic toy dynamics for orbital and bioreactor simulations.
|
||||||
|
- `src/idea194_attractorforge_verified_attractor/compiler.py` turns specs into controllers, PlanCerts, BasinSketches, and Merkle-logged PlanDeltas.
|
||||||
|
- `src/idea194_attractorforge_verified_attractor/runtime.py` provides deterministic execution and replay verification.
|
||||||
|
- `src/idea194_attractorforge_verified_attractor/cli.py` exposes a small CLI for manual inspection.
|
||||||
|
|
||||||
|
## Tech Stack
|
||||||
|
|
||||||
|
- Python 3.11+
|
||||||
|
- Standard library only for runtime logic
|
||||||
|
- `pytest` for tests
|
||||||
|
- `setuptools` build backend via `pyproject.toml`
|
||||||
|
|
||||||
|
## Rules
|
||||||
|
|
||||||
|
- Keep controller behavior deterministic.
|
||||||
|
- Preserve bounded control outputs in the compiler and runtime.
|
||||||
|
- Prefer small, explicit data models over implicit dictionaries.
|
||||||
|
- Update tests whenever compiler or runtime behavior changes.
|
||||||
|
- Keep artifact hashes stable by using canonical JSON formatting.
|
||||||
|
|
||||||
|
## Testing
|
||||||
|
|
||||||
|
- `bash test.sh`
|
||||||
|
- `pytest`
|
||||||
|
- `python3 -m build`
|
||||||
36
README.md
36
README.md
|
|
@ -1,3 +1,35 @@
|
||||||
# idea194-attractorforge-verified-attractor
|
# AttractorForge
|
||||||
|
|
||||||
Source logic for Idea #194
|
AttractorForge is a deterministic Python prototype for compiling attractor-oriented mission and biosphere controllers into auditable artifacts.
|
||||||
|
|
||||||
|
## What it does
|
||||||
|
|
||||||
|
- Defines a small attractor DSL with typed specs.
|
||||||
|
- Compiles two templates: invariant-manifold transfer and microbiome basin stabilization.
|
||||||
|
- Emits a PlanCert, BasinSketch, and Merkle-logged PlanDelta.
|
||||||
|
- Replays plans deterministically on a lightweight runtime.
|
||||||
|
- Ships toy orbital and bioreactor adapters for offline testing.
|
||||||
|
|
||||||
|
## Package
|
||||||
|
|
||||||
|
- Name: `idea194-attractorforge-verified-attractor`
|
||||||
|
- Entry point: `attractorforge`
|
||||||
|
|
||||||
|
## Example
|
||||||
|
|
||||||
|
```bash
|
||||||
|
attractorforge --kind motion_manifold --spec-id demo-motion
|
||||||
|
attractorforge --kind bio_basin --spec-id demo-bio
|
||||||
|
```
|
||||||
|
|
||||||
|
## Testing
|
||||||
|
|
||||||
|
```bash
|
||||||
|
bash test.sh
|
||||||
|
```
|
||||||
|
|
||||||
|
## Repository layout
|
||||||
|
|
||||||
|
- `src/idea194_attractorforge_verified_attractor/` core implementation
|
||||||
|
- `tests/` unit tests
|
||||||
|
- `AGENTS.md` contributor rules for future agents
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,21 @@
|
||||||
|
[build-system]
|
||||||
|
requires = ["setuptools>=68", "wheel"]
|
||||||
|
build-backend = "setuptools.build_meta"
|
||||||
|
|
||||||
|
[project]
|
||||||
|
name = "idea194-attractorforge-verified-attractor"
|
||||||
|
version = "0.1.0"
|
||||||
|
description = "Verified attractor-program compiler and deterministic runtime for mission and biosphere stability"
|
||||||
|
readme = "README.md"
|
||||||
|
requires-python = ">=3.11"
|
||||||
|
license = "MIT"
|
||||||
|
authors = [{name = "OpenCode"}]
|
||||||
|
|
||||||
|
[project.scripts]
|
||||||
|
attractorforge = "idea194_attractorforge_verified_attractor.cli:main"
|
||||||
|
|
||||||
|
[tool.setuptools]
|
||||||
|
package-dir = {"" = "src"}
|
||||||
|
|
||||||
|
[tool.setuptools.packages.find]
|
||||||
|
where = ["src"]
|
||||||
|
|
@ -0,0 +1,20 @@
|
||||||
|
from .compiler import AttractorCompiler, CompilationResult
|
||||||
|
from .models import (
|
||||||
|
AttractorSpec,
|
||||||
|
BasinSketch,
|
||||||
|
ControlStep,
|
||||||
|
PlanCert,
|
||||||
|
PlanDelta,
|
||||||
|
)
|
||||||
|
from .runtime import DeterministicRuntime
|
||||||
|
|
||||||
|
__all__ = [
|
||||||
|
"AttractorCompiler",
|
||||||
|
"AttractorSpec",
|
||||||
|
"BasinSketch",
|
||||||
|
"CompilationResult",
|
||||||
|
"ControlStep",
|
||||||
|
"DeterministicRuntime",
|
||||||
|
"PlanCert",
|
||||||
|
"PlanDelta",
|
||||||
|
]
|
||||||
|
|
@ -0,0 +1,41 @@
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from dataclasses import dataclass
|
||||||
|
|
||||||
|
from .models import Control, State
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class ToyOrbitalAdapter:
|
||||||
|
"""A tiny stable manifold proxy for offline tests.
|
||||||
|
|
||||||
|
State layout: [radial_error, along_track_error].
|
||||||
|
Control layout: [radial_correction, tangential_correction].
|
||||||
|
"""
|
||||||
|
|
||||||
|
damping: float = 0.12
|
||||||
|
coupling: float = 0.03
|
||||||
|
|
||||||
|
def step(self, state: State, control: Control, dt: float) -> State:
|
||||||
|
radial, tangential = state
|
||||||
|
u_r, u_t = control
|
||||||
|
next_radial = radial + dt * (-self.damping * radial + u_r)
|
||||||
|
next_tangential = tangential + dt * (-self.coupling * tangential + u_t)
|
||||||
|
return (next_radial, next_tangential)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class ToyBioreactorAdapter:
|
||||||
|
"""A deterministic chemostat-style proxy for habitat control."""
|
||||||
|
|
||||||
|
dilution: float = 0.08
|
||||||
|
decay: float = 0.04
|
||||||
|
toxin_relaxation: float = 0.15
|
||||||
|
|
||||||
|
def step(self, state: State, control: Control, dt: float) -> State:
|
||||||
|
substrate, biomass, toxin = state
|
||||||
|
feed, purge = control
|
||||||
|
next_substrate = max(0.0, substrate + dt * (feed - self.dilution * substrate - 0.04 * biomass))
|
||||||
|
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))
|
||||||
|
return (next_substrate, next_biomass, next_toxin)
|
||||||
|
|
@ -0,0 +1,60 @@
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import argparse
|
||||||
|
from dataclasses import asdict
|
||||||
|
import json
|
||||||
|
|
||||||
|
from .compiler import AttractorCompiler
|
||||||
|
from .models import AttractorSpec
|
||||||
|
from .runtime import DeterministicRuntime
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> None:
|
||||||
|
parser = argparse.ArgumentParser(prog="attractorforge")
|
||||||
|
parser.add_argument("--kind", choices=["motion_manifold", "bio_basin"], required=True)
|
||||||
|
parser.add_argument("--spec-id", required=True)
|
||||||
|
args = parser.parse_args()
|
||||||
|
|
||||||
|
spec = _default_spec(args.spec_id, args.kind)
|
||||||
|
result = AttractorCompiler().compile(spec)
|
||||||
|
replay = DeterministicRuntime().replay(result.controller, result.plan_delta)
|
||||||
|
print(json.dumps({
|
||||||
|
"plan_cert": asdict(result.plan_cert),
|
||||||
|
"basin_sketch": asdict(result.basin_sketch),
|
||||||
|
"replay_matched": replay.matched,
|
||||||
|
"final_state": replay.final_state,
|
||||||
|
}, sort_keys=True, indent=2))
|
||||||
|
|
||||||
|
|
||||||
|
def _default_spec(spec_id: str, kind: str) -> AttractorSpec:
|
||||||
|
if kind == "motion_manifold":
|
||||||
|
return AttractorSpec(
|
||||||
|
spec_id=spec_id,
|
||||||
|
kind=kind,
|
||||||
|
state_labels=("radial_error", "along_track_error"),
|
||||||
|
target_state=(0.0, 0.0),
|
||||||
|
control_limit=0.8,
|
||||||
|
dt=0.1,
|
||||||
|
horizon=24,
|
||||||
|
basin_radius=1.8,
|
||||||
|
convergence_rate=0.34,
|
||||||
|
safety_margin=0.15,
|
||||||
|
objective="invariant-manifold-transfer",
|
||||||
|
)
|
||||||
|
return AttractorSpec(
|
||||||
|
spec_id=spec_id,
|
||||||
|
kind=kind,
|
||||||
|
state_labels=("substrate", "biomass", "toxin"),
|
||||||
|
target_state=(1.0, 0.8, 0.2),
|
||||||
|
control_limit=0.6,
|
||||||
|
dt=0.1,
|
||||||
|
horizon=24,
|
||||||
|
basin_radius=0.9,
|
||||||
|
convergence_rate=0.28,
|
||||||
|
safety_margin=0.1,
|
||||||
|
objective="microbiome-basin-stabilization",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
|
|
@ -0,0 +1,145 @@
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from dataclasses import dataclass
|
||||||
|
import hashlib
|
||||||
|
import json
|
||||||
|
from typing import Protocol
|
||||||
|
|
||||||
|
from .adapters import ToyBioreactorAdapter, ToyOrbitalAdapter
|
||||||
|
from .models import AttractorSpec, BasinSketch, ControlStep, PlanCert, PlanDelta, State
|
||||||
|
|
||||||
|
|
||||||
|
class StepAdapter(Protocol):
|
||||||
|
def step(self, state: State, control: tuple[float, ...], dt: float) -> State: ...
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class CompiledController:
|
||||||
|
spec: AttractorSpec
|
||||||
|
adapter: StepAdapter
|
||||||
|
gain: tuple[float, ...]
|
||||||
|
|
||||||
|
def control(self, state: State) -> tuple[float, ...]:
|
||||||
|
error = tuple(s - t for s, t in zip(state, self.spec.target_state, strict=True))
|
||||||
|
raw = tuple(-k * e for k, e in zip(self.gain, error, strict=False))
|
||||||
|
return tuple(_clip(v, self.spec.control_limit) for v in raw)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class CompilationResult:
|
||||||
|
controller: CompiledController
|
||||||
|
plan_cert: PlanCert
|
||||||
|
plan_delta: PlanDelta
|
||||||
|
basin_sketch: BasinSketch
|
||||||
|
|
||||||
|
|
||||||
|
def _clip(value: float, limit: float) -> float:
|
||||||
|
return max(-limit, min(limit, value))
|
||||||
|
|
||||||
|
|
||||||
|
def _json(value: object) -> str:
|
||||||
|
return json.dumps(value, sort_keys=True, separators=(",", ":"))
|
||||||
|
|
||||||
|
|
||||||
|
def _hash_text(value: str) -> str:
|
||||||
|
return hashlib.sha256(value.encode("utf-8")).hexdigest()
|
||||||
|
|
||||||
|
|
||||||
|
def _hash_state(state: State) -> str:
|
||||||
|
return _hash_text(_json([round(v, 12) for v in state]))
|
||||||
|
|
||||||
|
|
||||||
|
def _hash_control(control: tuple[float, ...]) -> str:
|
||||||
|
return _hash_text(_json([round(v, 12) for v in control]))
|
||||||
|
|
||||||
|
|
||||||
|
def _merkle_root(items: list[str]) -> str:
|
||||||
|
if not items:
|
||||||
|
return _hash_text("empty")
|
||||||
|
level = items[:]
|
||||||
|
while len(level) > 1:
|
||||||
|
next_level: list[str] = []
|
||||||
|
for idx in range(0, len(level), 2):
|
||||||
|
left = level[idx]
|
||||||
|
right = level[idx + 1] if idx + 1 < len(level) else left
|
||||||
|
next_level.append(_hash_text(left + right))
|
||||||
|
level = next_level
|
||||||
|
return level[0]
|
||||||
|
|
||||||
|
|
||||||
|
class AttractorCompiler:
|
||||||
|
def compile(self, spec: AttractorSpec) -> CompilationResult:
|
||||||
|
spec = spec.normalized()
|
||||||
|
adapter, gain = self._select_template(spec)
|
||||||
|
controller = CompiledController(spec=spec, adapter=adapter, gain=gain)
|
||||||
|
plan_delta = self._simulate(spec, controller)
|
||||||
|
plan_cert = self._certify(spec, controller, plan_delta)
|
||||||
|
basin_sketch = BasinSketch(
|
||||||
|
attractor_id=spec.spec_id,
|
||||||
|
kind=spec.kind,
|
||||||
|
estimated_basin_radius=spec.basin_radius,
|
||||||
|
convergence_rate_bound=spec.convergence_rate,
|
||||||
|
controller_hash=_hash_text(_json({"gain": gain, "target": spec.target_state, "kind": spec.kind})),
|
||||||
|
dt=spec.dt,
|
||||||
|
)
|
||||||
|
return CompilationResult(controller=controller, plan_cert=plan_cert, plan_delta=plan_delta, basin_sketch=basin_sketch)
|
||||||
|
|
||||||
|
def _select_template(self, spec: AttractorSpec) -> tuple[StepAdapter, tuple[float, ...]]:
|
||||||
|
if spec.kind == "motion_manifold":
|
||||||
|
return ToyOrbitalAdapter(), tuple(max(0.35, spec.convergence_rate * 1.7) for _ in spec.target_state)
|
||||||
|
if spec.kind == "bio_basin":
|
||||||
|
return ToyBioreactorAdapter(), (
|
||||||
|
max(0.25, spec.convergence_rate * 1.3),
|
||||||
|
max(0.25, spec.convergence_rate * 1.1),
|
||||||
|
)
|
||||||
|
raise ValueError(f"Unsupported attractor kind: {spec.kind}")
|
||||||
|
|
||||||
|
def _simulate(self, spec: AttractorSpec, controller: CompiledController) -> PlanDelta:
|
||||||
|
state = tuple(v + spec.basin_radius for v in spec.target_state)
|
||||||
|
steps: list[ControlStep] = []
|
||||||
|
for index in range(spec.horizon):
|
||||||
|
control = controller.control(state)
|
||||||
|
steps.append(
|
||||||
|
ControlStep(
|
||||||
|
step_index=index,
|
||||||
|
time=round(index * spec.dt, 12),
|
||||||
|
state=state,
|
||||||
|
control=control,
|
||||||
|
state_hash=_hash_state(state),
|
||||||
|
control_hash=_hash_control(control),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
state = controller.adapter.step(state, control, spec.dt)
|
||||||
|
merkle_root = _merkle_root([step.state_hash + step.control_hash for step in steps])
|
||||||
|
replay_hash = _hash_text(_json([steps[-1].state_hash if steps else "", merkle_root, spec.spec_id]))
|
||||||
|
return PlanDelta(spec_id=spec.spec_id, seed_state=steps[0].state if steps else spec.target_state, steps=tuple(steps), merkle_root=merkle_root, replay_hash=replay_hash)
|
||||||
|
|
||||||
|
def _certify(self, spec: AttractorSpec, controller: CompiledController, delta: PlanDelta) -> PlanCert:
|
||||||
|
radius = spec.basin_radius
|
||||||
|
gain_sum = sum(controller.gain)
|
||||||
|
worst_case_cost = round(spec.horizon * spec.dt * spec.control_limit * 0.5, 12)
|
||||||
|
robustness_margin = round(max(0.0, gain_sum * radius - spec.safety_margin), 12)
|
||||||
|
preconditions = {
|
||||||
|
"state_within_basin": radius,
|
||||||
|
"control_limit": spec.control_limit,
|
||||||
|
"dt": spec.dt,
|
||||||
|
}
|
||||||
|
postconditions = {
|
||||||
|
"distance_to_target_bound": round(radius * max(0.0, 1.0 - spec.convergence_rate), 12),
|
||||||
|
"monotone_energy_decay": True,
|
||||||
|
}
|
||||||
|
witness = _json({"gain": controller.gain, "objective": spec.objective, "target": spec.target_state, "adapter": type(controller.adapter).__name__})
|
||||||
|
witness_hash = _hash_text(witness)
|
||||||
|
certificate_hash = _hash_text(_json({"spec": spec.spec_id, "witness": witness_hash, "merkle": delta.merkle_root, "post": postconditions}))
|
||||||
|
return PlanCert(
|
||||||
|
spec_id=spec.spec_id,
|
||||||
|
kind=spec.kind,
|
||||||
|
objective=spec.objective,
|
||||||
|
preconditions=preconditions,
|
||||||
|
postconditions=postconditions,
|
||||||
|
worst_case_cost=worst_case_cost,
|
||||||
|
robustness_margin=robustness_margin,
|
||||||
|
certificate_hash=certificate_hash,
|
||||||
|
witness_hash=witness_hash,
|
||||||
|
merkle_root=delta.merkle_root,
|
||||||
|
)
|
||||||
|
|
@ -0,0 +1,91 @@
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from dataclasses import dataclass, field
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
|
||||||
|
State = tuple[float, ...]
|
||||||
|
Control = tuple[float, ...]
|
||||||
|
|
||||||
|
|
||||||
|
def _canonical_float(value: float) -> float:
|
||||||
|
return float(f"{value:.12f}")
|
||||||
|
|
||||||
|
|
||||||
|
def _canonical_tuple(values: tuple[float, ...]) -> tuple[float, ...]:
|
||||||
|
return tuple(_canonical_float(v) for v in values)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class AttractorSpec:
|
||||||
|
spec_id: str
|
||||||
|
kind: str
|
||||||
|
state_labels: tuple[str, ...]
|
||||||
|
target_state: State
|
||||||
|
control_limit: float
|
||||||
|
dt: float
|
||||||
|
horizon: int
|
||||||
|
basin_radius: float
|
||||||
|
convergence_rate: float
|
||||||
|
safety_margin: float
|
||||||
|
objective: str
|
||||||
|
metadata: dict[str, Any] = field(default_factory=dict)
|
||||||
|
|
||||||
|
def normalized(self) -> "AttractorSpec":
|
||||||
|
return AttractorSpec(
|
||||||
|
spec_id=self.spec_id,
|
||||||
|
kind=self.kind,
|
||||||
|
state_labels=self.state_labels,
|
||||||
|
target_state=_canonical_tuple(self.target_state),
|
||||||
|
control_limit=_canonical_float(self.control_limit),
|
||||||
|
dt=_canonical_float(self.dt),
|
||||||
|
horizon=int(self.horizon),
|
||||||
|
basin_radius=_canonical_float(self.basin_radius),
|
||||||
|
convergence_rate=_canonical_float(self.convergence_rate),
|
||||||
|
safety_margin=_canonical_float(self.safety_margin),
|
||||||
|
objective=self.objective,
|
||||||
|
metadata=dict(self.metadata),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class ControlStep:
|
||||||
|
step_index: int
|
||||||
|
time: float
|
||||||
|
state: State
|
||||||
|
control: Control
|
||||||
|
state_hash: str
|
||||||
|
control_hash: str
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class BasinSketch:
|
||||||
|
attractor_id: str
|
||||||
|
kind: str
|
||||||
|
estimated_basin_radius: float
|
||||||
|
convergence_rate_bound: float
|
||||||
|
controller_hash: str
|
||||||
|
dt: float
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class PlanCert:
|
||||||
|
spec_id: str
|
||||||
|
kind: str
|
||||||
|
objective: str
|
||||||
|
preconditions: dict[str, Any]
|
||||||
|
postconditions: dict[str, Any]
|
||||||
|
worst_case_cost: float
|
||||||
|
robustness_margin: float
|
||||||
|
certificate_hash: str
|
||||||
|
witness_hash: str
|
||||||
|
merkle_root: str
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class PlanDelta:
|
||||||
|
spec_id: str
|
||||||
|
seed_state: State
|
||||||
|
steps: tuple[ControlStep, ...]
|
||||||
|
merkle_root: str
|
||||||
|
replay_hash: str
|
||||||
|
|
@ -0,0 +1,87 @@
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from dataclasses import dataclass
|
||||||
|
|
||||||
|
from .compiler import CompiledController
|
||||||
|
from .models import ControlStep, PlanDelta, State
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class ReplayResult:
|
||||||
|
final_state: State
|
||||||
|
matched: bool
|
||||||
|
steps: tuple[ControlStep, ...]
|
||||||
|
|
||||||
|
|
||||||
|
class DeterministicRuntime:
|
||||||
|
def execute(self, controller: CompiledController, seed_state: State, horizon: int) -> PlanDelta:
|
||||||
|
state = seed_state
|
||||||
|
steps: list[ControlStep] = []
|
||||||
|
for index in range(horizon):
|
||||||
|
control = controller.control(state)
|
||||||
|
step = ControlStep(
|
||||||
|
step_index=index,
|
||||||
|
time=round(index * controller.spec.dt, 12),
|
||||||
|
state=state,
|
||||||
|
control=control,
|
||||||
|
state_hash=controller_hash_state(state),
|
||||||
|
control_hash=controller_hash_control(control),
|
||||||
|
)
|
||||||
|
steps.append(step)
|
||||||
|
state = controller.adapter.step(state, control, controller.spec.dt)
|
||||||
|
merkle_root = _merkle_root([step.state_hash + step.control_hash for step in steps])
|
||||||
|
replay_hash = _hash_text(f"{seed_state!r}:{merkle_root}:{controller.spec.spec_id}")
|
||||||
|
return PlanDelta(spec_id=controller.spec.spec_id, seed_state=seed_state, steps=tuple(steps), merkle_root=merkle_root, replay_hash=replay_hash)
|
||||||
|
|
||||||
|
def replay(self, controller: CompiledController, delta: PlanDelta) -> ReplayResult:
|
||||||
|
state = delta.seed_state
|
||||||
|
replayed: list[ControlStep] = []
|
||||||
|
for index, original in enumerate(delta.steps):
|
||||||
|
control = controller.control(state)
|
||||||
|
step = ControlStep(
|
||||||
|
step_index=index,
|
||||||
|
time=round(index * controller.spec.dt, 12),
|
||||||
|
state=state,
|
||||||
|
control=control,
|
||||||
|
state_hash=controller_hash_state(state),
|
||||||
|
control_hash=controller_hash_control(control),
|
||||||
|
)
|
||||||
|
replayed.append(step)
|
||||||
|
state = controller.adapter.step(state, control, controller.spec.dt)
|
||||||
|
if step.state_hash != original.state_hash or step.control_hash != original.control_hash:
|
||||||
|
return ReplayResult(final_state=state, matched=False, steps=tuple(replayed))
|
||||||
|
return ReplayResult(final_state=state, matched=True, steps=tuple(replayed))
|
||||||
|
|
||||||
|
|
||||||
|
def _hash_text(value: str) -> str:
|
||||||
|
import hashlib
|
||||||
|
|
||||||
|
return hashlib.sha256(value.encode("utf-8")).hexdigest()
|
||||||
|
|
||||||
|
|
||||||
|
def _json(values: object) -> str:
|
||||||
|
import json
|
||||||
|
|
||||||
|
return json.dumps(values, sort_keys=True, separators=(",", ":"))
|
||||||
|
|
||||||
|
|
||||||
|
def controller_hash_state(state: State) -> str:
|
||||||
|
return _hash_text(_json([round(v, 12) for v in state]))
|
||||||
|
|
||||||
|
|
||||||
|
def controller_hash_control(control: tuple[float, ...]) -> str:
|
||||||
|
return _hash_text(_json([round(v, 12) for v in control]))
|
||||||
|
|
||||||
|
|
||||||
|
def _merkle_root(items: list[str]) -> str:
|
||||||
|
if not items:
|
||||||
|
return _hash_text("empty")
|
||||||
|
level = items[:]
|
||||||
|
while len(level) > 1:
|
||||||
|
next_level: list[str] = []
|
||||||
|
for idx in range(0, len(level), 2):
|
||||||
|
left = level[idx]
|
||||||
|
right = level[idx + 1] if idx + 1 < len(level) else left
|
||||||
|
next_level.append(_hash_text(left + right))
|
||||||
|
level = next_level
|
||||||
|
return level[0]
|
||||||
|
|
@ -0,0 +1,6 @@
|
||||||
|
#!/usr/bin/env bash
|
||||||
|
set -euo pipefail
|
||||||
|
|
||||||
|
python3 -m pip install -e . pytest build >/dev/null
|
||||||
|
pytest -q
|
||||||
|
python3 -m build
|
||||||
|
|
@ -0,0 +1,11 @@
|
||||||
|
from idea194_attractorforge_verified_attractor.cli import _default_spec
|
||||||
|
|
||||||
|
|
||||||
|
def test_default_specs_have_expected_shapes() -> None:
|
||||||
|
motion = _default_spec("s1", "motion_manifold")
|
||||||
|
bio = _default_spec("s2", "bio_basin")
|
||||||
|
|
||||||
|
assert len(motion.target_state) == 2
|
||||||
|
assert len(bio.target_state) == 3
|
||||||
|
assert motion.kind == "motion_manifold"
|
||||||
|
assert bio.kind == "bio_basin"
|
||||||
|
|
@ -0,0 +1,48 @@
|
||||||
|
from idea194_attractorforge_verified_attractor import AttractorCompiler, AttractorSpec
|
||||||
|
from idea194_attractorforge_verified_attractor.runtime import DeterministicRuntime
|
||||||
|
|
||||||
|
|
||||||
|
def test_motion_template_compiles_and_replays() -> None:
|
||||||
|
spec = AttractorSpec(
|
||||||
|
spec_id="motion-demo",
|
||||||
|
kind="motion_manifold",
|
||||||
|
state_labels=("radial_error", "along_track_error"),
|
||||||
|
target_state=(0.0, 0.0),
|
||||||
|
control_limit=0.8,
|
||||||
|
dt=0.1,
|
||||||
|
horizon=18,
|
||||||
|
basin_radius=1.5,
|
||||||
|
convergence_rate=0.3,
|
||||||
|
safety_margin=0.2,
|
||||||
|
objective="invariant-manifold-transfer",
|
||||||
|
)
|
||||||
|
|
||||||
|
result = AttractorCompiler().compile(spec)
|
||||||
|
replay = DeterministicRuntime().replay(result.controller, result.plan_delta)
|
||||||
|
|
||||||
|
assert replay.matched
|
||||||
|
assert result.plan_cert.spec_id == spec.spec_id
|
||||||
|
assert result.basin_sketch.controller_hash
|
||||||
|
assert abs(replay.final_state[0]) < 0.8
|
||||||
|
assert abs(replay.final_state[1]) < 0.8
|
||||||
|
|
||||||
|
|
||||||
|
def test_bio_template_uses_bounded_controls() -> None:
|
||||||
|
spec = AttractorSpec(
|
||||||
|
spec_id="bio-demo",
|
||||||
|
kind="bio_basin",
|
||||||
|
state_labels=("substrate", "biomass", "toxin"),
|
||||||
|
target_state=(1.0, 0.7, 0.15),
|
||||||
|
control_limit=0.6,
|
||||||
|
dt=0.1,
|
||||||
|
horizon=20,
|
||||||
|
basin_radius=0.8,
|
||||||
|
convergence_rate=0.25,
|
||||||
|
safety_margin=0.1,
|
||||||
|
objective="microbiome-basin-stabilization",
|
||||||
|
)
|
||||||
|
|
||||||
|
result = AttractorCompiler().compile(spec)
|
||||||
|
assert all(abs(v) <= spec.control_limit for step in result.plan_delta.steps for v in step.control)
|
||||||
|
assert result.plan_cert.certificate_hash
|
||||||
|
assert result.plan_cert.merkle_root == result.plan_delta.merkle_root
|
||||||
Loading…
Reference in New Issue