Blame for core/capability.py

"""
core/capability.py — what "this capability works" is actually allowed to mean.

A registry that lists a tool as *available* and finds out at call time is
guessing. This is the vocabulary that replaces the guess, and it is a ladder:
each rung is a separate fact, established separately, and no rung implies the
one above it.

    cataloged      the mesh knows it exists — a description, an install line
    wrapped        it is represented as coordinates that can be addressed
    simulatable    mock and dry_run produce a real envelope for it
    reachable      the edge that would run it is paired and online
    authenticated  the session it needs is currently valid at the far end
    callable       an executor for this specific coordinate resolved
    authorized     this identity is permitted to invoke it
    executed       it ran, and there is a receipt

`authenticated` is separate from `reachable` on purpose. A seller's laptop can
be online and paired while Amazon has expired the session or is asking for MFA.
Collapsing the two produces the worst failure this system can have: a district
that looks live, and silently returns yesterday's numbers.

The rule that keeps it honest:

    SEC-Ω describes transport and policy protection.
    This describes whether work can actually be performed.
    Neither implies the other.

A SEC-Ω-protected request can be perfectly secure and still correctly answer
edge_unreachable, executor_missing, unsupported_platform, authorization_denied,
approval_required or execution_failed. A protected channel is not a working
destination, and a catalog entry must never be able to present itself as a live
tool because it happened to be wrapped.
"""
from __future__ import annotations

from typing import Any, Dict, List, Optional

SCHEMA = "signalmesh.capability/v1"

# Ordered. Index is the rung; nothing here is a set of independent flags.
LADDER: List[str] = [
    "cataloged", "wrapped", "simulatable", "reachable",
    "authenticated", "callable", "authorized", "executed",
]

MEANING: Dict[str, str] = {
    "cataloged":   "known to the mesh — described, not represented",
    "wrapped":     "represented as coordinates that can be addressed",
    "simulatable": "mock and dry_run produce a real envelope",
    "reachable":   "the edge that would run it is paired and online",
    "authenticated": "the session it needs is currently valid at the far end",
    "callable":    "an executor resolved for this coordinate",
    "authorized":  "this identity is permitted to invoke it",
    "executed":    "it ran, and there is a receipt",
}

# Why a rung was not reached. These are answers, not errors — a capability that
# honestly reports why it cannot run is working correctly.
BLOCKED: Dict[str, str] = {
    "reachable":  "edge_unreachable",
    "authenticated": "session_expired",
    "callable":   "executor_missing",
    "authorized": "authorization_denied",
    "executed":   "not_yet_invoked",
}


def rung(status: str) -> int:
    try:
        return LADDER.index(status)
    except ValueError:
        return -1


def at_least(status: str, required: str) -> bool:
    """Is this capability at or above the rung an action needs?"""
    a, b = rung(status), rung(required)
    return a >= 0 and b >= 0 and a >= b


def assess(*, wrapped: bool = False, has_coords: bool = False,
           simulatable: bool = False, edge_required: bool = False,
           edge_online: bool = False, session_required: bool = False,
           session_valid: bool = False,
           executor_resolved: Optional[bool] = None,
           authorized: Optional[bool] = None,
           receipt: Optional[str] = None) -> Dict[str, Any]:
    """Climb as far as the evidence goes, then stop and say why.

    Deliberately conservative at every step. `executor_resolved=None` means
    nobody has checked, which is not the same as an executor being present, and
    is therefore not enough to call something callable.
    """
    if receipt:
        return _at("executed", receipt=receipt)

    if not wrapped or not has_coords:
        return _at("cataloged",
                   why="not wrapped — it is described, not represented")
    if not simulatable:
        return _at("wrapped", why="no mock or dry_run behaviour to exercise")

    # An origin that needs no edge is reachable by definition; one that does is
    # reachable only while a paired device is actually connected.
    if edge_required and not edge_online:
        return _at("simulatable", blocked_at="reachable", why=BLOCKED["reachable"],
                   detail="requires a paired meshd edge; none is connected")

    # A live machine is not a live session. This is the rung that stops a
    # district reporting healthy while the far end is asking for MFA.
    if session_required and not session_valid:
        return _at("reachable", blocked_at="authenticated",
                   why=BLOCKED["authenticated"],
                   detail=("the edge is online but the session it needs has "
                           "expired or is being challenged — the owner has to "
                           "sign in again; no stored password can do it for them"))

    if executor_resolved is not True:
        return _at("authenticated" if session_required else "reachable",
                   blocked_at="callable", why=BLOCKED["callable"],
                   detail=("no executor has been resolved for this coordinate. "
                           "A reachable edge is not proof that the thing it "
                           "would run is installed, supported or complete."))

    if authorized is not True:
        return _at("callable", blocked_at="authorized", why=BLOCKED["authorized"],
                   detail="this identity has not been permitted to invoke it")

    return _at("authorized", blocked_at="executed", why=BLOCKED["executed"],
               detail="permitted, but it has not been run")


def _at(status: str, *, blocked_at: str = "", why: str = "",
        detail: str = "", receipt: str = "") -> Dict[str, Any]:
    out: Dict[str, Any] = {
        "schema":  SCHEMA,
        "status":  status,
        "rung":    rung(status),
        "means":   MEANING[status],
        "ladder":  LADDER,
        # said out loud on every response, because it is the thing people get wrong
        "note": ("SEC-Ω protects the channel; this describes whether work can be "
                 "performed. Neither implies the other."),
    }
    if blocked_at:
        out["blocked_at"] = blocked_at
        out["reason"] = why
        out["detail"] = detail
    elif why:
        out["why"] = why
    if receipt:
        out["receipt"] = receipt
    return out


if __name__ == "__main__":
    assert LADDER[0] == "cataloged" and LADDER[-1] == "executed"
    assert rung("callable") > rung("reachable") > rung("simulatable")
    assert rung("nonsense") == -1

    assert at_least("executed", "callable")
    assert at_least("callable", "callable")
    assert not at_least("simulatable", "callable")
    assert not at_least("callable", "nonsense")

    # a catalog entry is not a tool, however much it describes itself
    c = assess()
    assert c["status"] == "cataloged", c
    assert "not wrapped" in c["why"], "a plain status must still say why it stopped"

    # wrapped, but nothing to exercise
    w = assess(wrapped=True, has_coords=True)
    assert w["status"] == "wrapped" and "no mock" in w["why"], w

    # the nicoli case: wrapped and simulatable, edge not paired
    n = assess(wrapped=True, has_coords=True, simulatable=True, edge_required=True)
    assert n["status"] == "simulatable" and n["blocked_at"] == "reachable"
    assert n["reason"] == "edge_unreachable"

    # pairing the edge does NOT make it callable — this is the overclaim guard
    paired = assess(wrapped=True, has_coords=True, simulatable=True,
                    edge_required=True, edge_online=True)
    assert paired["status"] == "reachable", paired
    assert paired["blocked_at"] == "callable" and paired["reason"] == "executor_missing"
    assert "not proof" in paired["detail"]

    # an executor that nobody checked is not an executor that exists
    unchecked = assess(wrapped=True, has_coords=True, simulatable=True,
                       edge_required=True, edge_online=True, executor_resolved=None)
    assert unchecked["status"] == "reachable"

    # ── the seller's laptop is on; Amazon is asking for MFA ─────────────────
    # The failure this rung exists to prevent: a district that looks healthy
    # and quietly serves yesterday's numbers.
    stale = assess(wrapped=True, has_coords=True, simulatable=True,
                   edge_required=True, edge_online=True,
                   session_required=True, session_valid=False,
                   executor_resolved=True, authorized=True)
    assert stale["status"] == "reachable", stale
    assert stale["blocked_at"] == "authenticated"
    assert stale["reason"] == "session_expired"
    assert "sign in again" in stale["detail"]
    assert "no stored password" in stale["detail"]

    signed_in = assess(wrapped=True, has_coords=True, simulatable=True,
                       edge_required=True, edge_online=True,
                       session_required=True, session_valid=True,
                       executor_resolved=True, authorized=True)
    assert signed_in["status"] == "authorized", signed_in

    # a session that expires between two calls demotes the capability — being
    # authorized a minute ago is not evidence about now
    assert rung(stale["status"]) < rung(signed_in["status"])

    # and something needing no session is not held at that rung
    nosession = assess(wrapped=True, has_coords=True, simulatable=True,
                       edge_required=True, edge_online=True,
                       session_required=False, executor_resolved=True, authorized=True)
    assert nosession["status"] == "authorized"

    resolved = assess(wrapped=True, has_coords=True, simulatable=True,
                      edge_required=True, edge_online=True, executor_resolved=True)
    assert resolved["status"] == "callable" and resolved["blocked_at"] == "authorized"

    ok = assess(wrapped=True, has_coords=True, simulatable=True,
                edge_required=True, edge_online=True,
                executor_resolved=True, authorized=True)
    assert ok["status"] == "authorized" and ok["reason"] == "not_yet_invoked"

    done = assess(wrapped=True, has_coords=True, simulatable=True,
                  edge_required=True, edge_online=True,
                  executor_resolved=True, authorized=True, receipt="call_abc123")
    assert done["status"] == "executed" and done["receipt"] == "call_abc123"

    # an http origin needs no edge, so reachability is not gated on one
    http = assess(wrapped=True, has_coords=True, simulatable=True,
                  edge_required=False, executor_resolved=True, authorized=True)
    assert http["status"] == "authorized", http

    # every rung says the thing people get wrong
    for state in (c, w, n, paired, resolved, ok, done, http):
        assert "SEC-Ω protects the channel" in state["note"]

    print("capability self-check OK")