ayllu/model_binding.py

# SPDX-License-Identifier: Apache-2.0
"""Honest runtime binding between the Ayllu council and SZL-Forge.

Ayllu personas are task roles sharing A11oy's routed model backend.  They are
not eleven separately trained models.  This module binds each role to a
declared SZL-Forge profile and a bounded set of *proposal* capabilities while
leaving execution, approval, signing, and verification in independent runtime
organs.
"""
from __future__ import annotations

import copy
import json
from typing import Any, Mapping


SCHEMA = "szl.ayllu.model-family-binding/v1"
SECOND_BRAIN_SCHEMA = "szl.khipu.compound-second-brain.v1"
FAMILY_ID = "SZL-Forge-1.5B"
COMPUTE_PLANE = "SZL-Yupaq"
BINDING_STATE = "PROFILE_AWARE_LOCAL_ROUTING_ARTIFACT_BINDING_PARTIAL"

_ALL_COMPUTE_OPERATIONS = (
    "formula.org_lambda.weighted_geomean",
    "quant.sample.pipeline",
    "quantum.qubo.exact_baseline",
    "numerics.external.run",
    "numerics.external.compare",
    "proof.lean.inventory",
    "formula.admission.inventory",
    "brain.corpus.inventory",
    "lake.evidence.inventory",
)

_PROFILE_STATES = {
    "ReceiptAgent-v1": "ARTIFACT_BYTES_RECONCILED_MEASURED_NOT_PROMOTED",
    "BrainNavigator-v1": "SIGNED_RECEIPTS_VALID_ARTIFACT_BINDING_CONFLICT",
    "Operator-v1": "PLANNED_TOOL_CONTRACT_REQUIRED",
    "Sentinel-v1": "PLANNED_SECURITY_ADMISSION_REQUIRED",
    "Anatomy-v1": "PLANNED_ONTOLOGY_ADMISSION_REQUIRED",
}

_PERSONA_BINDINGS: dict[str, dict[str, Any]] = {
    "Amaru": {
        "primary_profile": "Operator-v1",
        "supporting_profiles": ["Anatomy-v1"],
        "proposal_surfaces": ["architecture.review", "anatomy.inspect"],
        "compute_operations": [],
    },
    "Ruwaq": {
        "primary_profile": "Operator-v1",
        "supporting_profiles": [],
        "proposal_surfaces": ["code.plan", "build.review", "compute.submit"],
        "compute_operations": [
            "proof.lean.inventory",
            "formula.admission.inventory",
            "brain.corpus.inventory",
            "lake.evidence.inventory",
        ],
    },
    "Yupaq": {
        "primary_profile": "ReceiptAgent-v1",
        "supporting_profiles": ["BrainNavigator-v1"],
        "proposal_surfaces": ["compute.submit", "receipt.verify", "proof.review"],
        "compute_operations": list(_ALL_COMPUTE_OPERATIONS),
    },
    "Qhaway": {
        "primary_profile": "Sentinel-v1",
        "supporting_profiles": ["Anatomy-v1"],
        "proposal_surfaces": ["simulation.review", "failure.evaluate"],
        "compute_operations": ["quantum.qubo.exact_baseline"],
    },
    "Maskaq": {
        "primary_profile": "BrainNavigator-v1",
        "supporting_profiles": ["ReceiptAgent-v1"],
        "proposal_surfaces": ["brain.query", "evidence.retrieve", "citation.review"],
        "compute_operations": [
            "brain.corpus.inventory",
            "formula.admission.inventory",
            "lake.evidence.inventory",
        ],
    },
    "Hampiq": {
        "primary_profile": "Anatomy-v1",
        "supporting_profiles": ["Sentinel-v1"],
        "proposal_surfaces": ["health.inspect", "remediation.propose"],
        "compute_operations": ["lake.evidence.inventory"],
    },
    "Yanapaq": {
        "primary_profile": "Operator-v1",
        "supporting_profiles": [],
        "proposal_surfaces": ["ops.review", "incident.support"],
        "compute_operations": [],
    },
    "Chaka": {
        "primary_profile": "Operator-v1",
        "supporting_profiles": ["ReceiptAgent-v1"],
        "proposal_surfaces": ["connector.review", "contract.crosswalk"],
        "compute_operations": [],
    },
    "Kamachiq": {
        "primary_profile": "Operator-v1",
        "supporting_profiles": ["ReceiptAgent-v1"],
        "proposal_surfaces": ["route.review", "plan.sequence", "approval.request"],
        "compute_operations": [],
    },
    "Qhatuq": {
        "primary_profile": "ReceiptAgent-v1",
        "supporting_profiles": [],
        "proposal_surfaces": ["risk.review", "quant.compute"],
        "compute_operations": [
            "quant.sample.pipeline",
            "formula.org_lambda.weighted_geomean",
        ],
    },
    "Willakuq": {
        "primary_profile": "ReceiptAgent-v1",
        "supporting_profiles": [],
        "proposal_surfaces": ["receipt.verify", "provenance.review", "archive.propose"],
        "compute_operations": ["lake.evidence.inventory"],
    },
}

_HARD_BOUNDARIES = {
    "personas_are_separate_weights": False,
    "tool_dispatch_active": False,
    "can_execute_external_actions": False,
    "can_approve_own_proposal": False,
    "can_sign_own_evidence": False,
    "can_self_certify_correctness": False,
    "model_output_is_verified_truth": False,
    "automatic_lounge_publish": False,
    "compute_execution_location": "SZL-Yupaq external governed computation plane",
    "binding_rule": "MODEL_PROPOSES; YUPAQ_VALIDATES_SCHEMA; ENGINE_COMPUTES; HONESTY_LABELS; RECEIPT_BINDS",
}


def persona_binding(
    name: str,
    *,
    actual_model: Any = None,
    backend_mode: str | None = None,
    model_attestation: Mapping[str, Any] | None = None,
    grounding: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    """Return one role's immutable model/control-plane binding."""
    canonical = next((key for key in _PERSONA_BINDINGS if key.lower() == (name or "").lower()), None)
    if canonical is None:
        raise KeyError(f"unknown Ayllu persona: {name}")
    binding = copy.deepcopy(_PERSONA_BINDINGS[canonical])
    primary = binding["primary_profile"]
    attestation = copy.deepcopy(dict(model_attestation or {})) or None
    grounding_summary = None
    if grounding:
        grounding_summary = {
            "schema": grounding.get("schema"),
            "state": grounding.get("state"),
            "content_access": grounding.get("content_access"),
            "query_sha256": grounding.get("query_sha256"),
            "evidence_set_sha256": grounding.get("evidence_set_sha256"),
            "handles_sha256": grounding.get("handles_sha256"),
            "augmented_prompt_sha256": grounding.get("augmented_prompt_sha256"),
            "handle_evidence_set_equivalent": grounding.get(
                "handle_evidence_set_equivalent"),
            "citation_validation": copy.deepcopy(
                grounding.get("citation_validation")),
            "rejected_model_output_sha256": grounding.get(
                "rejected_model_output_sha256"),
            "grounded_count": grounding.get("grounded_count"),
        }
    attested_served_model = (
        attestation.get("served_model") if attestation is not None else None)
    model_identity_reconciled = (
        actual_model == attested_served_model
        if isinstance(attested_served_model, str) and attested_served_model
        else None
    )
    binding.update({
        "schema": SCHEMA,
        "persona": canonical,
        "family_id": FAMILY_ID,
        "binding_state": BINDING_STATE,
        "profile_state": _PROFILE_STATES[primary],
        "actual_model": actual_model,
        "backend_mode": backend_mode or "NOT_OBSERVED",
        "actual_model_authority": "turn receipt and router evidence",
        "attested_served_model": attested_served_model,
        "model_identity_reconciled": model_identity_reconciled,
        "model_attestation": attestation,
        "model_attestation_sha256": (
            _canonical_sha256(attestation) if attestation is not None else None
        ),
        "grounding": grounding_summary,
        "grounding_sha256": (
            _canonical_sha256(grounding_summary) if grounding_summary is not None else None
        ),
        "compute_plane": COMPUTE_PLANE,
        "authority": "PROPOSAL_ONLY",
        "hard_boundaries": copy.deepcopy(_HARD_BOUNDARIES),
    })
    return binding


def family_binding(
    *,
    namespace: str = "a11oy",
    backend_status: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    """Return the machine-readable Ayllu-to-Forge family contract."""
    status = dict(backend_status or {})
    profile_runtime = status.get("forge_profiles")
    return {
        "schema": SCHEMA,
        "family_id": FAMILY_ID,
        "binding_state": BINDING_STATE,
        "runtime_backend": status,
        "runtime_backend_is_profile_pinned": False,
        "runtime_profile_status": profile_runtime,
        "profile_pin_requirement": (
            "The exact profile tag must be observed, its immutable weight/blob digest must "
            "match a signed release manifest, and the turn receipt must bind that attestation."
        ),
        "personas": [persona_binding(name) for name in _PERSONA_BINDINGS],
        "compute": {
            "plane_id": COMPUTE_PLANE,
            "capabilities_endpoint": f"/api/{namespace}/v1/compute/capabilities",
            "submit_endpoint": f"/api/{namespace}/v1/compute/jobs",
            "allowed_operations": list(_ALL_COMPUTE_OPERATIONS),
            "dispatch_state": "PROPOSAL_ONLY_NOT_ACTIVE_IN_AYLLU_LOOP",
            "stateful_routes_require_auth": True,
        },
        "hard_boundaries": copy.deepcopy(_HARD_BOUNDARIES),
    }


def second_brain_binding(
    *,
    namespace: str = "a11oy",
    backend_status: Mapping[str, Any] | None = None,
    rag_status: Mapping[str, Any] | None = None,
    signer_ready: bool = False,
) -> dict[str, Any]:
    """Describe the Khipu Second Brain as an evidence-bound compound model.

    The generator tag, persistent retrieval index, controller, and receipt
    verifier are independent components.  Keeping that separation explicit
    prevents an index row count from being mislabeled as parameters or trained
    weights while still exposing one operational system contract.
    """
    backend = copy.deepcopy(dict(backend_status or {}))
    rag = copy.deepcopy(dict(rag_status or {}))
    profile_runtime = (
        (backend.get("forge_profiles") or {}).get("profiles") or {}
    ).get("BrainNavigator-v1") or {}
    exact_tag_observed = bool(profile_runtime.get("available"))
    index_ready = bool(rag.get("built"))
    ready = exact_tag_observed and index_ready
    if ready:
        state = "READY_FOR_GROUNDED_NAVIGATION_ARTIFACT_UNBOUND"
    elif not exact_tag_observed and not index_ready:
        state = "UNAVAILABLE_MODEL_AND_INDEX"
    elif not exact_tag_observed:
        state = "UNAVAILABLE_MODEL_TAG_MISSING"
    else:
        state = "UNAVAILABLE_INDEX_NOT_BUILT"
    return {
        "schema": SECOND_BRAIN_SCHEMA,
        "system_id": "SZL-Khipu-Second-Brain-v1",
        "system_type": "COMPOUND_MODEL_WITH_EXTERNAL_EVIDENCE_MEMORY",
        "state": state,
        "ready_for_grounded_navigation": ready,
        "live_grounded_turn_verified_this_request": False,
        "signer_ready_this_request": bool(signer_ready),
        "promotion_state": "BLOCKED_ARTIFACT_AND_EVAL_GATES",
        "profile": {
            "profile_id": "BrainNavigator-v1",
            "expected_model": profile_runtime.get("expected_model", "khipu:latest"),
            "served_model": profile_runtime.get("served_model"),
            "exact_tag_observed": exact_tag_observed,
            "artifact_binding": "UNBOUND",
            "turn_level_attestation_required": True,
        },
        "memory": {
            "kind": "PERSISTENT_SQLITE_HYBRID_RETRIEVAL_GRAPH",
            "built": index_ready,
            "document_count": rag.get("document_count", rag.get("files")),
            "chunk_count": rag.get("chunk_count", rag.get("chunks")),
            "corpus_chunk_count": rag.get(
                "corpus_chunk_count", rag.get("chunk_count", rag.get("chunks"))
            ),
            "brain_handle_count": rag.get("brain_handle_count", 0),
            "brain_handle_plane": rag.get("brain_handle_plane"),
            "training_authority_rows": rag.get("training_authority_rows", 0),
            "node_count": rag.get("node_count"),
            "edge_count": rag.get("edge_count"),
            "generation_id": rag.get("generation_id"),
            "generation_digest_sha256": rag.get("generation_digest_sha256"),
            "integrity_state": rag.get("integrity_state"),
            "rehydration_state": rag.get("rehydration_state"),
            "corpus": rag.get("corpus"),
            "index_mode": rag.get("mode"),
            "scope_boundary": (
                "Corpus chunks and the canonical 9,464-node Brain handle plane are "
                "separate, independently counted retrieval planes. Handles preserve "
                "source and quarantine metadata and grant no gradient authority."
            ),
            "evidence_access": "HANDLES_ONLY_TO_MODEL; CONTENT_STAYS_IN_CONTROLLER",
        },
        "grounding": {
            "ask_endpoint": f"/api/{namespace}/v1/ayllu/ask",
            "persona": "Maskaq",
            "query_endpoint": f"/api/{namespace}/code/rag/query",
            "required_receipt_fields": [
                "evidence_set_sha256",
                "handles_sha256",
                "augmented_prompt_sha256",
                "grounding_sha256",
                "model_attestation_sha256",
                "turn_output_sha256",
            ],
            "abstain_when_ungrounded": True,
        },
        "training_boundary": {
            "raw_brain_nodes_observed": 9464,
            "raw_brain_nodes_admitted_to_gradients": 0,
            "admission_is_row_level": True,
            "admission_engine": "szl_brain_training_admission.py",
            "admission_contract": "szl.brain-training-admission-report.v2",
            "evidence_security": (
                "ED25519_ROOT_SIGNED_PURPOSE_SCOPED_ISSUER_TOOL_KEY"
            ),
            "required_signed_inputs": [
                "protected_eval_content_sha256_list",
                "purpose_scoped_evidence_trust_store",
                "policy_root_signer",
                "root_signed_policy_bundle",
                "signed_prior_split_ledger_descriptor",
                "exact_split_ledger_head_sha256",
                "reviewer_allowlist",
                "artifact_signing_key",
                "explicit_train_admission_switch",
            ],
            "current_state": (
                "ROW_LEVEL_ADMISSION_ENGINE_IMPLEMENTED_CURRENT_RAW_ROWS_QUARANTINED"
            ),
            "required": [
                "stable_node_id",
                "content_sha256",
                "stable_source_identity",
                "immutable_source_revision",
                "author_and_rightsholder",
                "rights_basis_license_and_permission_scope",
                "privacy_classification_and_signed_pii_clearance",
                "source_timestamp_and_freshness",
                "canonical_state",
                "dedup_group",
                "contamination_result",
                "allowlisted_signed_review",
                "immutable_split",
                "cross_run_split_ledger_binding",
            ],
            "honesty": (
                "All graph nodes may participate in retrieval and evaluation; only "
                "independently admitted rows may enter gradients."
            ),
        },
        "hard_boundaries": {
            "index_is_model_weights": False,
            "retrieval_is_training": False,
            "model_can_read_raw_node_content": False,
            "model_can_write_canonical_memory": False,
            "model_can_self_certify_grounding": False,
        },
    }


def prompt_contract(binding: Mapping[str, Any]) -> str:
    """Serialize the binding into a compact system-prompt control contract."""
    compact = {
        "schema": binding.get("schema"),
        "family_id": binding.get("family_id"),
        "persona": binding.get("persona"),
        "primary_profile": binding.get("primary_profile"),
        "profile_state": binding.get("profile_state"),
        "authority": binding.get("authority"),
        "proposal_surfaces": binding.get("proposal_surfaces"),
        "compute_operations": binding.get("compute_operations"),
        "binding_rule": (binding.get("hard_boundaries") or {}).get("binding_rule"),
    }
    return (
        "A11OY MODEL-BINDING CONTRACT (machine-readable; binding):\n"
        + json.dumps(compact, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
        + "\nYou may propose only. Never claim that a proposal was executed, approved, "
          "signed, kernel-verified, or trained unless an independent receipt is present."
    )


def _canonical_sha256(value: Any) -> str:
    return __import__("hashlib").sha256(json.dumps(
        value, sort_keys=True, separators=(",", ":"), ensure_ascii=False
    ).encode("utf-8")).hexdigest()


__all__ = [
    "BINDING_STATE",
    "COMPUTE_PLANE",
    "FAMILY_ID",
    "SCHEMA",
    "SECOND_BRAIN_SCHEMA",
    "family_binding",
    "persona_binding",
    "prompt_contract",
    "second_brain_binding",
]