Spaces:
Running
Running
File size: 3,438 Bytes
a6a5d8e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | """AYNI-OS HTTP API — verifiable state at any past moment + reciprocity + flow.
Endpoints (ADDITIVE; do not touch existing a11oy routes):
GET /v1/ayni -> per-organ Ayni coefficients + T24 deficit report
GET /v1/replay?at=<ts> -> verifiable reconstructed state at past timestamp T
(event-sourcing replay; honest naming)
GET /v1/tinkuy -> current Kuramoto order parameter r + flow state
Uses FastAPI if available; otherwise exposes a pure-stdlib WSGI/dispatch `handle()`
so the logic is testable and runnable without extra deps (open-source only).
"""
from __future__ import annotations
from dataclasses import asdict
from typing import Optional
from .checkpoint import CheckpointStore
from .ledger import ReciprocityLedger
from .reciprocity_monitor import scan
from .rewind import reconstruct_at
from .tinkuy import TinkuyMonitor
class AyniService:
"""Holds the live ledger, checkpoint store, and tinkuy monitor."""
def __init__(self, ledger: Optional[ReciprocityLedger] = None,
store: Optional[CheckpointStore] = None,
tinkuy: Optional[TinkuyMonitor] = None) -> None:
self.ledger = ledger or ReciprocityLedger()
self.store = store or CheckpointStore()
self.tinkuy = tinkuy or TinkuyMonitor()
# ---- endpoint logic (framework-agnostic) ---------------------------
def get_ayni(self) -> dict:
report = scan(self.ledger)
return {
"alphas": report.alphas,
"deficits": [asdict(d) for d in report.deficits],
"halt": report.halt,
"tripwire": "T24",
"alpha_min": 0.45,
"alpha_balanced": 0.5,
}
def get_replay(self, at: float) -> dict:
st = reconstruct_at(self.ledger, target_ts=float(at), store=self.store)
d = asdict(st)
d["mechanism"] = "event-sourcing-replay" # honest: NOT time travel
d["chain_verified"] = st.chain_ok
return d
def get_tinkuy(self) -> dict:
st = self.tinkuy.state()
d = asdict(st)
d["model"] = "kuramoto-1975-order-parameter"
return d
# ---- stdlib dispatcher (no framework needed) -----------------------
def handle(self, path: str, query: Optional[dict] = None) -> tuple[int, dict]:
query = query or {}
if path == "/v1/ayni":
return 200, self.get_ayni()
if path == "/v1/replay":
if "at" not in query:
return 400, {"error": "missing required query param 'at' (timestamp)"}
return 200, self.get_replay(query["at"])
if path == "/v1/tinkuy":
return 200, self.get_tinkuy()
return 404, {"error": f"no route {path}"}
def build_fastapi_app(service: Optional[AyniService] = None):
"""Return a FastAPI app if FastAPI is installed; else None."""
try:
from fastapi import FastAPI, Query
except Exception:
return None
svc = service or AyniService()
app = FastAPI(title="AYNI-OS", version="1.0-additive")
@app.get("/v1/ayni")
def ayni(): # noqa: ANN202
return svc.get_ayni()
@app.get("/v1/replay")
def replay(at: float = Query(...)): # noqa: ANN202
return svc.get_replay(at)
@app.get("/v1/tinkuy")
def tinkuy(): # noqa: ANN202
return svc.get_tinkuy()
return app
|