Spaces:
Running
Running
| """ | |
| 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") | |