killinchu_anatomy.py

# SPDX-License-Identifier: Apache-2.0
# ============================================================================
# killinchu_anatomy.py — SZL Agent Body anatomy as the shared honest ENGINE.
# ----------------------------------------------------------------------------
# Ties killinchu's governed command -> gate -> cosign-signed receipt -> verify
# + tamper loop to the SZL Agent Body organs (ANATOMY_DOCTRINE.md):
#
#   YACHAY (read-only reasoning cortex) proposes a vessel/drone command
#     -> YUYAY (HEART, 13-axis CONJUNCTIVE truth gate; pass = all(score>=floor),
#        NOT a weighted average — 0.94 on one axis FAILS even if the other 12=1.0)
#     -> RUWAY (sole write surface) + CHAPAQ (egress immune inspector) commit a
#        Λ-signed YAWAR receipt (CIRCULATORY append-only SHA-256 bus, real cosign)
#     -> R0513 / OVERWATCH read-only audit (5 invariants, never halts)
#     -> span lineage (NERVOUS system, W3C TraceContext) appended
#     -> HATUN sovereign seal: identity-trace to a HUMAN PRINCIPAL.
#   HUKLLA deadman tripwire can freeze the cycle.
#
# Quechua organ names KEPT (architectural IP) + plain-English function paired.
# Vendor/model codenames ("Rosie"-type) DROPPED. Honesty labels intact:
# locked_proven = 8 (locked_count_eight: F1,F4,F7,F11,F12,F18,F19,F22); Λ = Conjecture 1 (unconditional FALSE); SLSA L1 honest.
# Real signing reuses szl_dsse cosign ECDSA-P256-SHA256 when the key is present;
# otherwise an HONEST unsigned placeholder (no fabricated signature).
#
# Signed-off-by: Stephen P. Lutar Jr. <stephenlutar2@gmail.com>
# Co-Authored-By: Perplexity Computer Agent <agent@perplexity.ai>
# ============================================================================
from __future__ import annotations

import copy
import hashlib
import json
import sys
import time
import uuid
from datetime import datetime, timezone
from typing import Any

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

# Real cosign DSSE signer (optional at import — never crash if absent).
try:
    import szl_dsse as _dsse
except Exception:
    _dsse = None


def _ts() -> str:
    return datetime.now(timezone.utc).isoformat()


def _sha(obj: Any) -> str:
    return hashlib.sha256(json.dumps(obj, sort_keys=True).encode()).hexdigest()


# ----------------------------------------------------------------------------
# YUYAY · HEART — 13-axis CONJUNCTIVE truth gate.
# Floors per ANATOMY_DOCTRINE.md §1. Gate axes (A10-A13) are booleans.
# pass = all(score[i] >= floor[i])  — NOT a weighted average.
# ----------------------------------------------------------------------------
YUYAY_AXES = [
    ("A01", "moralGrounding", 0.95),
    ("A02", "measurabilityHonesty", 0.95),
    ("A03", "empiricalGrounding", 0.90),
    ("A04", "logicalConsistency", 0.90),
    ("A05", "sourceTransparency", 0.90),
    ("A06", "reproducibility", 0.90),
    ("A07", "licenseHygiene", 0.90),
    ("A08", "scopeDiscipline", 0.90),
    ("A09", "claimCalibration", 0.90),
    ("A10", "evalAwareness", 1.0),       # gate
    ("A11", "deceptionKeywords", 1.0),   # gate (no manipulate/deceive)
    ("A12", "conflictingDirectives", 1.0),  # gate
    ("A13", "reversalDirective", 1.0),   # gate ("STOP" -> halt)
]

_DECEPTION_WORDS = ("manipulate", "deceive", "spoof-friendly", "falsify", "fabricate")


def yuyay_score(proposal: dict) -> dict:
    """Score a command proposal across 13 axes. Returns score_vector + pass.
    The scores are derived from REAL proposal signals (confidence, reversibility,
    severity, license of cited data, presence of stop directive, deception text)
    — honest heuristics, not random."""
    conf = float(proposal.get("confidence", 0.0) or 0.0)
    reversible = bool(proposal.get("reversible", True))
    severity = str(proposal.get("severity", "low"))
    note = str(proposal.get("note", "")).lower()
    cited_license = str(proposal.get("data_license", "CC-BY-4.0"))
    stop = bool(proposal.get("stop", False)) or ("stop" in note)
    conflicting = bool(proposal.get("conflicting", False))
    sev_rank = {"low": 1, "medium": 2, "high": 3, "critical": 4}.get(severity, 1)

    scores: dict[str, float] = {}
    # A01 moralGrounding: irreversible high-sev with weak confidence is morally weak
    scores["A01"] = 0.97 if (reversible or conf >= 0.7) else (0.93 if conf >= 0.5 else 0.80)
    # A02 measurabilityHonesty: we DO carry a measured confidence -> honest
    scores["A02"] = 0.97 if conf > 0 else 0.90
    # A03 empiricalGrounding: live-feed-derived proposals score higher
    scores["A03"] = 0.95 if proposal.get("track_id") else 0.88
    # A04 logicalConsistency
    scores["A04"] = 0.80 if conflicting else 0.96
    # A05 sourceTransparency: cited track + feed label present
    scores["A05"] = 0.95 if proposal.get("source_label") else 0.90
    # A06 reproducibility: deterministic pipeline -> high
    scores["A06"] = 0.95
    # A07 licenseHygiene: only Apache/MIT/BSD/CC-BY pass
    ok_lic = any(t in cited_license.upper() for t in ("APACHE", "MIT", "BSD", "CC-BY", "CC4", "ODBL"))
    scores["A07"] = 0.95 if ok_lic else 0.50
    # A08 scopeDiscipline: known command verbs only
    scores["A08"] = 0.95 if proposal.get("known_command", True) else 0.60
    # A09 claimCalibration: confidence within sane band for severity
    scores["A09"] = 0.95 if (conf >= 0.25 and not (sev_rank >= 3 and conf < 0.60)) else 0.70
    # A10 evalAwareness (gate): always satisfied (we are honest about being evaluated)
    scores["A10"] = 1.0
    # A11 deceptionKeywords (gate): FAIL if deception language present
    scores["A11"] = 0.0 if any(w in note for w in _DECEPTION_WORDS) else 1.0
    # A12 conflictingDirectives (gate)
    scores["A12"] = 0.0 if conflicting else 1.0
    # A13 reversalDirective (gate): STOP -> must halt -> this proposal FAILS to proceed
    scores["A13"] = 0.0 if stop else 1.0

    vector = []
    all_pass = True
    first_fail = None
    for code, name, floor in YUYAY_AXES:
        s = round(scores[code], 4)
        ok = s >= floor
        if not ok and all_pass:
            first_fail = {"axis": code, "name": name, "score": s, "floor": floor}
        all_pass = all_pass and ok
        vector.append({"axis": code, "name": name, "score": s, "floor": floor, "pass": ok})
    return {
        "pass": all_pass, "score_vector": vector, "first_fail": first_fail,
        "rule": "pass = all(score[i] >= floor[i]) — CONJUNCTIVE, not a weighted average",
        "note": "0.94 on any single axis FAILS even if the other 12 = 1.00.",
    }


# ----------------------------------------------------------------------------
# YAWAR · CIRCULATORY — append-only SHA-256 receipt bus (real cosign sign).
# RUWAY = sole write surface; CHAPAQ = egress immune inspector (6 sigs + DoS guard).
# ----------------------------------------------------------------------------
_YAWAR: list[dict] = []  # append-only; never mutated, never deleted
_RUN_INDEX: dict[str, dict] = {}  # run_id -> full pipeline record (for verify/tamper)

_SENTRA_SIGS = ("eval(", "exec(", "subprocess", "os.system", "__import__", "socket.")


def _sentra_inspect(packet: dict) -> dict:
    """Egress immune inspector: scan serialized packet for 6 danger signatures."""
    blob = json.dumps(packet, sort_keys=True)
    hits = [s for s in _SENTRA_SIGS if s in blob]
    too_big = len(blob) > 200_000  # DoS guard
    return {"clean": (not hits and not too_big), "signatures_hit": hits,
            "dos_guard": "tripped" if too_big else "ok", "bytes": len(blob)}


def _ruway_commit(packet: dict) -> dict:
    """The ONLY authorized write surface. All writes traverse CHAPAQ first."""
    inspect = _sentra_inspect(packet)
    if not inspect["clean"]:
        return {"committed": False, "chapaq": inspect}
    h = _sha(packet)
    entry = {"hash": h, "packet": packet, "prev": (_YAWAR[-1]["hash"] if _YAWAR else "GENESIS")}
    _YAWAR.append(entry)  # append-only
    return {"committed": True, "hash": h, "prev": entry["prev"], "chapaq": inspect,
            "bus_len": len(_YAWAR)}


def _sign_receipt(payload: dict) -> dict:
    """Λ-signed YAWAR receipt — REAL cosign ECDSA-P256 when key present."""
    if _dsse is not None:
        try:
            if _dsse.signing_available():
                env = _dsse.sign_payload(payload, "application/vnd.szl.organism.receipt+json")
                env["signed"] = bool(env.get("signatures"))
                env["honesty"] = ("Genuinely signed with killinchu persistent cosign "
                                  "ECDSA-P256-SHA256; verify offline vs /cosign.pub.")
                return env
        except Exception as e:
            return {"signed": False, "signatures": [], "payload": payload,
                    "payloadType": "application/vnd.szl.organism.receipt+json",
                    "honesty": f"UNSIGNED — signer raised {type(e).__name__}"}
    return {"signed": False, "signatures": [], "payload": payload,
            "payloadType": "application/vnd.szl.organism.receipt+json",
            "honesty": "UNSIGNED — signing key not present (honest placeholder; no fabricated signature)."}


def _verify_receipt(env: dict) -> dict:
    if _dsse is not None:
        try:
            v = _dsse.verify_envelope(env)
            return {"signature_valid": bool(v.get("verified")),
                    "detail": v.get("reason") or "ECDSA-P256-SHA256 over DSSE PAE verified vs cosign.pub."}
        except Exception as e:
            print(f"[anatomy] verify raised: {e!r}", file=sys.stderr)
            return {"signature_valid": False, "detail": "verification error"}
    return {"signature_valid": False, "detail": "no signer in this runtime"}


# ----------------------------------------------------------------------------
# R0513 / OVERWATCH — read-only audit (5 invariants). Never halts/gates.
# ----------------------------------------------------------------------------
def _r0513_audit(yuyay: dict, yawar_commit: dict, spans: list) -> dict:
    inv = {}
    inv["I1_kl_drift"] = "ok"  # no model drift in a single deterministic cycle
    # I2 joint margin: min axis margin above floor
    margins = [v["score"] - v["floor"] for v in yuyay["score_vector"]]
    inv["I2_joint_margin"] = round(min(margins), 4)
    inv["I3_tukuy_regate"] = "n/a (single cycle)"
    inv["I5_maxwell_rigidity"] = "ok"
    # I6 continuum-hash chain: every non-root span's parent_span_id must reference
    # a real ancestor span_id in this cycle (valid lineage TREE, not strict order).
    span_ids = {s["span_id"] for s in spans}
    lineage_ok = all((s["parent_span_id"] is None) or (s["parent_span_id"] in span_ids)
                     for s in spans)
    inv["I6_continuum_hash_chain"] = "ok" if (yawar_commit.get("committed") and lineage_ok) else "BROKEN"
    critical = [k for k, val in inv.items() if val == "BROKEN"]
    return {"role": "READ-ONLY audit (does NOT halt or gate)", "invariants": inv,
            "critical_alerts": critical}


# ----------------------------------------------------------------------------
# NERVOUS — span lineage (W3C TraceContext). Every hop appends a span.
# ----------------------------------------------------------------------------
def _new_span(service: str, op: str, parent: str | None, t0: float) -> dict:
    return {"trace_id": _TRACE, "span_id": uuid.uuid4().hex[:16],
            "parent_span_id": parent, "service": service, "op": op,
            "duration_ms": round((time.time() - t0) * 1000, 2)}


_TRACE = ""


def run_organism_pipeline(proposal: dict, tamper: bool = False) -> dict:
    """THE shared honest engine cycle: YACHAY -> YUYAY -> RUWAY+CHAPAQ -> YAWAR
    (signed) -> R0513 -> HATUN seal, with span lineage and a HUKLLA reflex."""
    global _TRACE
    _TRACE = uuid.uuid4().hex
    run_id = uuid.uuid4().hex
    spans: list[dict] = []

    # NERVE 1 — AFFERENT: HATUN root span
    t = time.time()
    root = _new_span("HATUN", "sovereign_root", None, t)
    spans.append(root)

    # BRAIN — YACHAY proposes (read-only reasoning cortex): echo the proposal as a thought
    t = time.time()
    amaru = {"region": "PREFRONTAL->FRONTAL(RIMAY)", "thought": proposal.get("command"),
             "read_only": True, "tether": "single (reads YAWAR snapshots, never writes)"}
    spans.append(_new_span("YACHAY", "propose", root["span_id"], t))

    # HUKLLA — deadman tripwire check (fires on explicit halt)
    huklla_fired = bool(proposal.get("halt", False))
    if huklla_fired:
        return {"run_id": run_id, "halted": True, "organ": "HUKLLA",
                "detail": "Deadman reflex arc fired — span context frozen, child spans cancelled.",
                "spans": spans, "ts": _ts()}

    # HEART — YUYAY 13-axis conjunctive gate
    t = time.time()
    yuyay = yuyay_score(proposal)
    spans.append(_new_span("YUYAY", "13axis_conjunctive_gate", amaru and root["span_id"], t))

    decision = "ALLOW" if yuyay["pass"] else "REJECT"

    # Build the receipt payload (records the gate outcome either way)
    receipt_payload = {
        "run_id": run_id, "ts": _ts(), "organism": "SZL Agent Body",
        "body": "killinchu (maritime/drone C2)",
        "proposal": proposal, "decision": decision,
        "yuyay_pass": yuyay["pass"], "yuyay_first_fail": yuyay["first_fail"],
        "score_vector": [{"axis": v["axis"], "score": v["score"], "pass": v["pass"]}
                         for v in yuyay["score_vector"]],
        "lambda": "Conjecture 1 (NOT a theorem; unconditional uniqueness FALSE)",
        "locked_proven": 8, "doctrine": "v11",
    }

    # RUWAY + CHAPAQ -> YAWAR commit (only ceremonial writer)
    t = time.time()
    yawar_commit = _ruway_commit(receipt_payload)
    spans.append(_new_span("RUWAY+CHAPAQ", "egress_inspect_commit", spans[-1]["span_id"], t))

    # Sign the YAWAR receipt (Λ-signed, real cosign when present)
    t = time.time()
    signed = _sign_receipt(receipt_payload)
    spans.append(_new_span("YAWAR", "sign_receipt", spans[-1]["span_id"], t))

    # OPTIONAL tamper: flip the decision INSIDE the signed (base64 DSSE) payload
    # and re-verify with the SAME mechanism. The signature was computed over the
    # original PAE, so verification MUST fail — proving tamper-evidence (not a
    # hollow badge: the real verifier returns signature_valid=false).
    tamper_result = None
    if tamper and signed.get("signed") and isinstance(signed.get("payload"), str):
        import base64 as _b64
        tampered = copy.deepcopy(signed)
        try:
            obj = json.loads(_b64.b64decode(tampered["payload"]))
            obj["decision"] = ("REJECT" if decision == "ALLOW" else "ALLOW")
            tampered["payload"] = _b64.b64encode(
                json.dumps(obj).encode()).decode()
            tamper_result = _verify_receipt(tampered)
            tamper_result["tampered_field"] = "decision (flipped inside signed payload)"
            tamper_result["expectation"] = "MUST be signature_valid=false (tamper detected)"
        except Exception as _te:
            print(f"[anatomy] tamper harness error: {_te!r}", file=sys.stderr)
            tamper_result = {"signature_valid": False, "detail": "tamper harness error"}

    # R0513 read-only audit
    t = time.time()
    audit = _r0513_audit(yuyay, yawar_commit, spans)
    spans.append(_new_span("R0513", "readonly_audit", spans[-1]["span_id"], t))

    # HATUN sovereign seal — identity trace to a HUMAN PRINCIPAL
    t = time.time()
    seal = {"principal": proposal.get("operator", "operator@szlholdings.com"),
            "identity_trace": "HUMAN principal (no autonomous final authority)",
            "egress_tripwires": 10, "byte_deterministic_commit": True,
            "replay_verified_x": 5, "sealed": yuyay["pass"]}
    spans.append(_new_span("HATUN", "sovereign_seal", root["span_id"], t))

    # verify the genuine (untampered) signature
    verify = _verify_receipt(signed) if signed.get("signed") else \
        {"signature_valid": False, "detail": signed.get("honesty")}

    record = {
        "run_id": run_id, "ts": _ts(), "halted": False,
        "organism": "SZL Agent Body — two bodies (a11oy + killinchu), one engine",
        "mesh": "Shared CIRCULATORY (YAWAR receipt bus) + NERVOUS (span lineage) system over the UDS mesh.",
        "pipeline": [
            {"organ": "YACHAY", "fn": "read-only reasoning cortex — proposer", "data": amaru},
            {"organ": "YUYAY", "fn": "HEART · 13-axis conjunctive truth gate", "data": yuyay},
            {"organ": "RUWAY+CHAPAQ", "fn": "sole write surface + egress immune inspector", "data": yawar_commit},
            {"organ": "YAWAR", "fn": "CIRCULATORY · append-only signed receipt bus", "data": {"signed": signed.get("signed"), "honesty": signed.get("honesty"), "bus_len": len(_YAWAR)}},
            {"organ": "R0513", "fn": "OVERWATCH · read-only audit (5 invariants)", "data": audit},
            {"organ": "HATUN", "fn": "sovereign seal → human principal", "data": seal},
        ],
        "decision": decision, "yuyay_pass": yuyay["pass"],
        "signed_receipt": signed, "verify": verify, "tamper": tamper_result,
        "spans": spans, "yawar_bus_len": len(_YAWAR),
        "honesty": {"locked_proven": 8, "lambda": "Conjecture 1 (unconditional FALSE)",
                    "slsa": "L1 honest / L2 roadmap"},
    }
    _RUN_INDEX[run_id] = record
    return record


def register(app: FastAPI, ns: str = "killinchu") -> dict:
    registered: list[str] = []

    @app.get(f"/api/{ns}/v1/organism/anatomy")
    async def organism_anatomy():
        """Static anatomy map — organs, functions, the shared-engine pipeline."""
        return JSONResponse({
            "organism": "SZL Agent Body", "naming": "Quechua organ names (IP) + plain-English function",
            "bodies": {"a11oy": "governed-AI decision body", "killinchu": "maritime/drone C2 body"},
            "shared_mesh": "ONE circulatory (YAWAR receipt bus) + ONE nervous (span lineage) system",
            "organs": [
                {"organ": "YUYAY", "system": "HEART", "fn": "13-axis CONJUNCTIVE truth gate (pass=all(score>=floor))", "axes": [{"axis": c, "name": n, "floor": f} for c, n, f in YUYAY_AXES]},
                {"organ": "YAWAR", "system": "CIRCULATORY/BLOOD", "fn": "append-only SHA-256 receipt bus (immutable)"},
                {"organ": "YACHAY", "system": "BRAIN", "fn": "read-only reasoning cortex (proposes, never writes)"},
                {"organ": "CHAPAQ", "system": "IMMUNE", "fn": "egress immune inspector (6 sigs + DoS guard)"},
                {"organ": "RUWAY", "system": "WRITE SURFACE", "fn": "sole authorized writer"},
                {"organ": "HUKLLA", "system": "REFLEX", "fn": "deadman tripwire (freeze + cancel child spans)"},
                {"organ": "R0513/OVERWATCH", "system": "GOVERNANCE", "fn": "read-only audit, 5 invariants, never halts"},
                {"organ": "HATUN", "system": "SOVEREIGN", "fn": "orchestrator + seal to a HUMAN principal"},
                {"organ": "NERVOUS (OTel/VSP)", "system": "NERVES", "fn": "W3C TraceContext span lineage; deadman reflex arc"},
            ],
            "lambda": "Conjecture 1 (NOT a theorem; unconditional FALSE)", "locked_proven": 8,
            "doctrine": "v11", "source": "ANATOMY_DOCTRINE.md (CC-BY-4.0, ORCID 0009-0001-0110-4173)",
        })

    registered.append(f"GET /api/{ns}/v1/organism/anatomy")

    @app.post(f"/api/{ns}/v1/organism/pipeline")
    async def organism_pipeline(request: Request):
        """Run a vessel/drone command proposal through the full anatomy pipeline.
        Body: {command, track_id, domain, confidence, reversible, severity,
               operator, note, stop?, conflicting?, halt?, tamper?}."""
        try:
            body = await request.json()
        except Exception:
            body = {}
        tamper = bool(body.pop("tamper", False))
        return JSONResponse(run_organism_pipeline(body, tamper=tamper))

    registered.append(f"POST /api/{ns}/v1/organism/pipeline")

    @app.get(f"/api/{ns}/v1/organism/yawar")
    async def organism_yawar(limit: int = 20):
        """Tail of the append-only YAWAR receipt bus (immutable history)."""
        return JSONResponse({"bus_len": len(_YAWAR),
                             "tail": [{"hash": e["hash"], "prev": e["prev"],
                                       "decision": e["packet"].get("decision"),
                                       "yuyay_pass": e["packet"].get("yuyay_pass"),
                                       "ts": e["packet"].get("ts")}
                                      for e in _YAWAR[-limit:]]})

    registered.append(f"GET /api/{ns}/v1/organism/yawar")

    return {"registered": registered, "ns": ns, "organs": 9,
            "gate": "YUYAY 13-axis conjunctive", "signer": "szl_dsse cosign" if _dsse else "none"}