a11oy_code.py

# SPDX-License-Identifier: Apache-2.0
# © 2026 Lutar, Stephen P. — SZL Holdings · ORCID 0009-0001-0110-4173
"""
a11oy.code — the 7-tier organ-mapped LLM router baked into the anatomy.

Doctrine v11 §14. ADDITIVE, self-contained module dropped beside serve.py in the
a11oy Space. Maps 7 LLM tiers to 7 organs; selects a tier from the 13-axis Λ trust
vector + organ_context; returns a DSSE-PLACEHOLDER Λ-receipt (honest: signing not
wired into CI), an in-process traceparent (honest: Wire D cross-mesh not yet
implemented), latency and a deterministic cost estimate.

The `response` is an HONEST STUB — no model key is wired in this Space. Tier
selection, organ routing, Λ-signal and the Λ-receipt hash chain are real,
deterministic math.
"""
from __future__ import annotations

import math
import time
from hashlib import sha256
from typing import Any, Dict, List, Optional

# ---------------------------------------------------------------------------
# 7 TIERS mapped to 7 organs (Doctrine v11 §14, founder LinkedIn anatomy).
# Ranked by escalation cost (0 = cheapest/fastest, 6 = frontier).
# ---------------------------------------------------------------------------
TIERS: List[Dict[str, Any]] = [
    {"tier": "FAST",     "rank": 0, "organ": "KALLPA",       "model_id": "gemini_3_1_pro|gpt_5_4_mini",
     "role": "wire propagation",       "cost_per_1k_usd": 0.0003},
    {"tier": "RECEIPT",  "rank": 1, "organ": "YAWAR",        "model_id": "claude_sonnet_4_6|llama_4",
     "role": "receipt synthesis",      "cost_per_1k_usd": 0.0030},
    {"tier": "MEMORY",   "rank": 2, "organ": "UNAY",         "model_id": "embeddings+claude_sonnet_4_6",
     "role": "cross-session recall",   "cost_per_1k_usd": 0.0020},
    {"tier": "HEART",    "rank": 3, "organ": "YUYAY",        "model_id": "claude_sonnet_4_6",
     "role": "13-axis evaluation",     "cost_per_1k_usd": 0.0030},
    {"tier": "IMMUNE",   "rank": 4, "organ": "SENTRA",       "model_id": "gpt_5_4_reasoning",
     "role": "adversarial detection",  "cost_per_1k_usd": 0.0100},
    {"tier": "PRIME",    "rank": 5, "organ": "AMARU_CORTEX", "model_id": "claude_opus_4_8",
     "role": "high-stakes reasoning",  "cost_per_1k_usd": 0.0150},
    {"tier": "FRONTIER", "rank": 6, "organ": "SUMAQ",        "model_id": "alphaproof",
     "role": "theorem discharge",      "cost_per_1k_usd": 0.0500},
]
_BY_TIER = {t["tier"]: t for t in TIERS}
_BY_ORGAN = {t["organ"]: t for t in TIERS}
_BY_RANK = {t["rank"]: t for t in TIERS}

# Organ_context keyword → preferred organ (auto-route).
_ORGAN_HINTS = {
    "AMARU_CORTEX": ["reason", "orchestrat", "plan", "strateg", "high-stakes", "complex"],
    "YUYAY":        ["eval", "axis", "gate", "critique", "score", "trust"],
    "KALLPA":       ["wire", "propagat", "route", "fast", "relay"],
    "SENTRA":       ["adversar", "threat", "attack", "inject", "malic", "immune", "security"],
    "YAWAR":        ["receipt", "ledger", "attest", "provenance", "sign"],
    "UNAY":         ["memory", "recall", "history", "embed", "retriev", "session"],
    "SUMAQ":        ["theorem", "proof", "lean", "discharge", "prove", "lemma"],
}

EPS = 1e-9


def lambda_signal(axis_scores: Optional[List[float]]) -> float:
    """13-axis Λ trust signal = weighted (uniform) geometric mean, zero-pinned."""
    if not axis_scores:
        return 0.0
    xs = [max(0.0, float(x)) for x in axis_scores]
    if any(x == 0.0 for x in xs):
        return 0.0
    return math.exp(sum(math.log(x) for x in xs) / len(xs))


def _organ_from_context(query: str, organ_context: str) -> Optional[str]:
    oc = (organ_context or "").strip().upper().replace("-", "_")
    if oc in _BY_ORGAN:
        return oc
    text = f"{organ_context} {query}".lower()
    best, score = None, 0
    for organ, kws in _ORGAN_HINTS.items():
        hits = sum(1 for kw in kws if kw in text)
        if hits > score:
            best, score = organ, hits
    return best


def _tier_from_lambda(L: float) -> str:
    """Trust→tier escalation. High Λ → FAST/cheap; low Λ → PRIME/extra-gates."""
    if L >= 0.95:
        return "FAST"
    if L >= 0.90:
        return "HEART"
    if L >= 0.75:
        return "IMMUNE"
    return "PRIME"  # low-trust / adversarial → premium + extra gates


def _cap_tier(tier_name: str, max_tier: Optional[str]) -> str:
    if not max_tier:
        return tier_name
    cap = _BY_TIER.get(str(max_tier).strip().upper())
    if not cap:
        return tier_name
    chosen = _BY_TIER[tier_name]
    if chosen["rank"] > cap["rank"]:
        return cap["tier"]
    return tier_name


def make_lambda_receipt(query: str, axis_scores: Optional[List[float]],
                        tier: Dict[str, Any], organ: str,
                        traceparent: Optional[str]) -> Dict[str, Any]:
    """DSSE PLACEHOLDER Λ-receipt (Doctrine v11: signing not wired into CI)."""
    L = lambda_signal(axis_scores)
    payload = {
        "query_digest": sha256(query.encode()).hexdigest(),
        "axis_scores": [round(float(x), 6) for x in (axis_scores or [])],
        "lambda_signal": round(L, 9),
        "tier_used": tier["tier"],
        "organ_routed": organ,
        "model_id": tier["model_id"],
        "traceparent": traceparent,
    }
    pae = f"DSSEv1 application/vnd.szl.code+json {payload}".encode()
    sig = "PLACEHOLDER:" + sha256(pae).hexdigest()
    return {
        "payloadType": "application/vnd.szl.code+json",
        "payload": payload,
        "signatures": [{"keyid": "a11oy.code", "sig": sig}],
        "signature_status": "PLACEHOLDER — Sigstore signing not yet wired into CI (Doctrine v11)",
    }


def route(query: str, axis_scores: Optional[List[float]] = None,
          organ_context: str = "", max_tier: Optional[str] = None,
          require_lambda_receipt: bool = True, traceparent: Optional[str] = None,
          auto: bool = False) -> Dict[str, Any]:
    """Route a query to an organ + tier, return the full a11oy.code response.

    auto=True → ignore organ_context, derive organ+tier purely from query + Λ.
    """
    t0 = time.time()
    L = lambda_signal(axis_scores)

    # organ selection
    organ = None if auto else _organ_from_context(query, organ_context)
    if organ is None:
        organ = _organ_from_context(query, "") or "YUYAY"

    # tier selection: organ's native tier, escalated by Λ if low-trust.
    organ_tier = _BY_ORGAN.get(organ, _BY_TIER["HEART"])
    lambda_tier = _BY_TIER[_tier_from_lambda(L)]
    # take the MORE cautious (higher rank) of organ-native vs Λ-escalation
    chosen = organ_tier if organ_tier["rank"] >= lambda_tier["rank"] else lambda_tier
    chosen_name = _cap_tier(chosen["tier"], max_tier)
    tier = _BY_TIER[chosen_name]

    response = (
        f"[HONEST STUB] a11oy.code routed to organ {organ} via tier {tier['tier']} "
        f"(model {tier['model_id']}). No model key wired in this Space; organ routing, "
        f"tier selection, Λ-signal and Λ-receipt are real deterministic math. "
        f"Role: {tier['role']}."
    )

    out: Dict[str, Any] = {
        "organ_routed": organ,
        "tier_used": tier["tier"],
        "llm_model_id": tier["model_id"],
        "response": response,
        "λ_signal": round(L, 9),
        "lambda_signal": round(L, 9),
        "latency_ms": round((time.time() - t0) * 1000.0, 3),
        "cost_estimate_usd": round(tier["cost_per_1k_usd"] * (len(query) / 1000.0 + 0.5), 6),
        "traceparent_propagated": traceparent,
        "traceparent_note": "in-process only — Wire D (cross-mesh traceparent) NOT yet implemented (Doctrine v11)",
        "doctrine": "v11",
        "service": "a11oy.code",
    }
    if require_lambda_receipt:
        out["λ_receipt"] = make_lambda_receipt(query, axis_scores, tier, organ, traceparent)
        out["lambda_receipt"] = out["λ_receipt"]
    return out


def tiers_payload() -> Dict[str, Any]:
    return {
        "count": len(TIERS),
        "tiers": TIERS,
        "organ_mapping": {t["organ"]: t["tier"] for t in TIERS},
        "doctrine": "v11",
        "service": "a11oy.code",
        "honesty": {
            "lambda_receipt_signature": "PLACEHOLDER (Sigstore not wired into CI)",
            "traceparent": "in-process only (Wire D cross-mesh not yet implemented)",
        },
    }


if __name__ == "__main__":
    import json
    print(json.dumps(tiers_payload(), indent=2)[:400])
    r = route("prove the lambda boundedness lemma in Lean", [0.99] * 13,
              organ_context="theorem", traceparent="00-abc-def-01")
    print("route organ:", r["organ_routed"], "tier:", r["tier_used"], "Λ:", r["λ_signal"])
    r2 = route("detect prompt injection attack", [0.6] * 13, auto=True)
    print("auto organ:", r2["organ_routed"], "tier:", r2["tier_used"], "Λ:", r2["λ_signal"])
    assert len(TIERS) == 7
    assert r["organ_routed"] == "SUMAQ"
    print("OK — a11oy.code 7-tier router self-check passed.")