diff --git a/.github/workflows/lampstand.yml b/.github/workflows/lampstand.yml new file mode 100644 index 0000000..5010cd1 --- /dev/null +++ b/.github/workflows/lampstand.yml @@ -0,0 +1,27 @@ +name: lampstand +on: + push: + paths: + - 'tools/lampstand_parse.py' + - 'tools/lampstand_route.py' + - 'tests/test_lampstand_parse.py' + - 'tests/test_lampstand_route.py' + - '.github/workflows/lampstand.yml' + pull_request: + paths: + - 'tools/lampstand_parse.py' + - 'tools/lampstand_route.py' + - 'tests/test_lampstand_parse.py' + - 'tests/test_lampstand_route.py' + - '.github/workflows/lampstand.yml' +jobs: + verify: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.11' + - run: python -m pip install pytest + - name: Run lampstand parser + router tests + run: python -m pytest tests/test_lampstand_parse.py tests/test_lampstand_route.py -q diff --git a/tests/__pycache__/test_lampstand_parse.cpython-312-pytest-8.3.4.pyc b/tests/__pycache__/test_lampstand_parse.cpython-312-pytest-8.3.4.pyc new file mode 100644 index 0000000..d26ea94 Binary files /dev/null and b/tests/__pycache__/test_lampstand_parse.cpython-312-pytest-8.3.4.pyc differ diff --git a/tests/__pycache__/test_lampstand_route.cpython-312-pytest-8.3.4.pyc b/tests/__pycache__/test_lampstand_route.cpython-312-pytest-8.3.4.pyc new file mode 100644 index 0000000..c6d6640 Binary files /dev/null and b/tests/__pycache__/test_lampstand_route.cpython-312-pytest-8.3.4.pyc differ diff --git a/tests/test_lampstand_parse.py b/tests/test_lampstand_parse.py new file mode 100644 index 0000000..3e2a5b3 --- /dev/null +++ b/tests/test_lampstand_parse.py @@ -0,0 +1,50 @@ +from tools.lampstand_parse import parse + + +def test_discover_contact_lists_in_org(): + p = parse("show me all contact lists in my org") + assert p["intent"] == "ActionShow" + assert p["object"] == "ContactLists" + assert p["purpose"] == "discover" + assert p["space"] == "user-space" + assert "Own" in p["scope"] + assert p["admissible"] is True + + +def test_operate_launch_needs_consent(): + p = parse("open terminal on my laptop") + assert p["intent"] == "ActionLaunch" + assert p["purpose"] == "operate" + assert p["admissible"] == "consent" + + +def test_destructive_delete_is_inadmissible(): + p = parse("delete last week's logs") + assert p["intent"] == "ActionDelete" + assert p["destructive"] is True + assert p["space"] == "system-space" + assert p["admissible"] is False + + +def test_egress_send_needs_consent(): + p = parse("send my contact lists to partners") + assert p["purpose"] == "egress" + assert p["admissible"] == "consent" + + +def test_unrecognized_query_is_fail_closed(): + p = parse("the quick brown fox") + assert p["intent"] is None + assert p["admissible"] is False + + +def test_empty_query_never_raises(): + p = parse("") + assert p["intent"] is None and p["admissible"] is False + + +def test_every_content_token_is_annotated_for_known_query(): + p = parse("delete last week's logs") + # each token carries at least its recognized annotation + assert all(isinstance(t["ann"], list) for t in p["tokens"]) + assert any(a[0] == "Destructive" for t in p["tokens"] for a in t["ann"]) diff --git a/tests/test_lampstand_route.py b/tests/test_lampstand_route.py new file mode 100644 index 0000000..24d628a --- /dev/null +++ b/tests/test_lampstand_route.py @@ -0,0 +1,34 @@ +from tools.lampstand_route import route, to_turn + + +def test_discover_routes_to_ir_allowed(): + r = route("show me all contact lists in my org") + assert r["verdict"] == "allow" + assert "sherlock (IR)" in r["route"] + + +def test_operate_routes_to_consent_plane(): + r = route("open terminal on my laptop") + assert r["verdict"] == "consent" + assert "consent-plane" in r["route"] + + +def test_destructive_routes_to_governor_denied(): + r = route("delete last week's logs") + assert r["verdict"] == "deny" + assert "Governor queue" in r["route"] + + +def test_unrecognized_query_fails_closed_to_deny(): + r = route("the quick brown fox") + assert r["verdict"] == "deny" + assert "no recognized action" in r["reason"] + + +def test_to_turn_projects_a_witness_turn(): + t = to_turn("show me all contact lists in my org") + assert t["user"].startswith("show me") + assert t["purpose"] == "discover" + assert t["admissible"] is True + assert t["tokens"] and all("w" in tok and "ann" in tok for tok in t["tokens"]) + assert t["id"].startswith("turn:") diff --git a/tools/__pycache__/lampstand_parse.cpython-312.pyc b/tools/__pycache__/lampstand_parse.cpython-312.pyc new file mode 100644 index 0000000..c6c1eb7 Binary files /dev/null and b/tools/__pycache__/lampstand_parse.cpython-312.pyc differ diff --git a/tools/__pycache__/lampstand_route.cpython-312.pyc b/tools/__pycache__/lampstand_route.cpython-312.pyc new file mode 100644 index 0000000..d994df5 Binary files /dev/null and b/tools/__pycache__/lampstand_route.cpython-312.pyc differ diff --git a/tools/lampstand_parse.py b/tools/lampstand_parse.py new file mode 100644 index 0000000..6fa8941 --- /dev/null +++ b/tools/lampstand_parse.py @@ -0,0 +1,113 @@ +"""lampstand — deterministic v0 semantic parse for the launcher. + +Turns a natural-language query into typed tokens (the annotation tree the launcher +and Turn Witness surfaces render) plus a resolved intent: {intent, object, scope, +purpose, space, admissible}. This is the rule/lexicon v0 — a transparent baseline to +be superseded by the learned App-Intents model, not a permanent dictionary. Pure +stdlib so it parses identically in CI, on device, and in tests. + +Purposes follow the consent plane: discover / implement / verify / ship / operate / +egress / administer. A destructive operate is marked so it can never self-authorize. +""" +from __future__ import annotations +from typing import Any, Dict, List + +# verb -> (intent concept, purpose) +ACTIONS = { + "show": ("ActionShow", "discover"), "list": ("ActionShow", "discover"), + "find": ("ActionShow", "discover"), "get": ("ActionShow", "discover"), + "search": ("ActionShow", "discover"), + "open": ("ActionLaunch", "operate"), "launch": ("ActionLaunch", "operate"), + "start": ("ActionLaunch", "operate"), "run": ("ActionLaunch", "operate"), + "delete": ("ActionDelete", "operate·destructive"), + "remove": ("ActionDelete", "operate·destructive"), + "wipe": ("ActionDelete", "operate·destructive"), + "send": ("ActionSend", "egress"), "share": ("ActionSend", "egress"), + "email": ("ActionSend", "egress"), + "create": ("ActionCreate", "implement"), "add": ("ActionCreate", "implement"), +} +# noun -> (concept, annotation class, space) +ENTITIES = { + "contact": ("ContactLists", "intent", "user-space"), + "contacts": ("ContactLists", "intent", "user-space"), + "list": ("ContactLists", "intent", "user-space"), + "lists": ("Lists", "type", "user-space"), + "terminal": ("TurtleTerm", "intent", "user-space"), + "log": ("LogData", "entity", "system-space"), + "logs": ("LogData", "entity", "system-space"), + "org": ("Organization", "entity", "user-space"), + "organization": ("Organization", "entity", "user-space"), + "laptop": ("Device", "entity", "user-space"), + "device": ("Device", "entity", "user-space"), +} +RELATIONS = {"in": "Contains", "on": "OnDevice", "to": "SendTo", "from": "From"} +SCOPE = {"my": "Own", "mine": "Own", "last": "TimeWindow", "week": "TimeWindow", + "week's": "TimeWindow", "today": "TimeWindow", "yesterday": "TimeWindow"} + + +def _norm(tok: str) -> str: + return tok.strip().lower().strip(".,!?;:") + + +def parse(query: str) -> Dict[str, Any]: + """Parse a query into annotated tokens + a resolved intent (never raises).""" + words = [w for w in (query or "").split() if w.strip()] + tokens: List[Dict[str, Any]] = [] + intent = obj = None + purpose = None + scopes: List[str] = [] + space = "user-space" + destructive = False + + for w in words: + n = _norm(w) + ann: List[List[str]] = [] + if n in ACTIONS: + concept, pur = ACTIONS[n] + ann.append([concept, "intent"]) + if intent is None: + intent, purpose = concept, pur + if "destructive" in pur: + destructive = True + ann.append(["Destructive", "type"]) + if n in ENTITIES: + concept, cls, sp = ENTITIES[n] + ann.append([concept, cls]) + if cls in ("intent", "entity") and obj is None: + obj, space = concept, sp + if n in RELATIONS: + ann.append([RELATIONS[n], "relation"]) + if n in SCOPE: + s = SCOPE[n] + ann.append([s, "scope"]) + if s not in scopes: + scopes.append(s) + tokens.append({"w": w, "ann": ann}) + + admissible = _admissible(purpose, space, destructive) + return { + "query": query, + "tokens": tokens, + "intent": intent, + "object": obj, + "scope": scopes, + "purpose": purpose, + "space": space, + "destructive": destructive, + "admissible": admissible, + } + + +def _admissible(purpose, space, destructive): + """Fail-closed admissibility verdict: True | 'consent' | False.""" + if purpose is None: + return False # no recognized action → nothing to admit + if destructive: + return False # irreversible acts never self-authorize + if purpose == "discover": + return True + if purpose == "implement": + return True if space in ("user-space", "agent-space") else "consent" + if purpose in ("operate", "egress", "administer"): + return "consent" # explicit owner grant required + return False diff --git a/tools/lampstand_route.py b/tools/lampstand_route.py new file mode 100644 index 0000000..7fc3206 --- /dev/null +++ b/tools/lampstand_route.py @@ -0,0 +1,69 @@ +"""lampstand — route a parsed query to a governed, typed action. + +Takes a query (or a parse dict) and returns a fail-closed routing decision: an +admissible discover routes to the IR (sherlock/holmes); an operate/egress needs an +explicit owner grant (consent-plane); a destructive or unrecognized query is refused +and escalated to the Governor. Nothing routes to execution without a verdict. +""" +from __future__ import annotations +from typing import Any, Dict, Union + +try: # importable both as `tools.lampstand_route` (from repo root) and standalone + from lampstand_parse import parse +except ImportError: # pragma: no cover + from tools.lampstand_parse import parse + + +def route(query_or_parse: Union[str, Dict[str, Any]]) -> Dict[str, Any]: + p = parse(query_or_parse) if isinstance(query_or_parse, str) else query_or_parse + adm = p.get("admissible") + + if adm is True: + verdict = "allow" + chain = (["lampstand", "sherlock (IR)", "holmes"] + if p.get("purpose") == "discover" else ["lampstand", "app-intents"]) + reason = f"admitted: :{p['intent']} · purpose={p['purpose']} · {p['space']}" + elif adm == "consent": + verdict = "consent" + chain = ["lampstand", "app-intents", "consent-plane"] + reason = f"{p['purpose']} on {p['space']} requires an explicit owner grant" + else: # False → fail closed + verdict = "deny" + chain = ["lampstand", "guardrail-fabric", "Governor queue"] + reason = ("no recognized action in query" if p.get("intent") is None + else f"{p['purpose']}: irreversible/inadmissible — escalated to Governor") + + return { + "verdict": verdict, + "route": chain, + "reason": reason, + "intent": p.get("intent"), + "object": p.get("object"), + "purpose": p.get("purpose"), + "space": p.get("space"), + "scope": p.get("scope", []), + } + + +def to_turn(query: str, sys_text: str = "") -> Dict[str, Any]: + """Project a query into a Turn Witness feed turn (surface data/turn-witness.json).""" + p = parse(query) + r = route(p) + tid = "turn:%08x" % (abs(hash(query)) & 0xFFFFFFFF) + return { + "id": tid, + "user": query, + "sys": sys_text or _default_sys(r), + "tokens": [{"w": t["w"], "g": "", "ann": t["ann"]} for t in p["tokens"]], + "purpose": p["purpose"] or "unknown", + "space": p["space"], + "admissible": p["admissible"], + } + + +def _default_sys(r: Dict[str, Any]) -> str: + return { + "allow": f"routed via {r['route'][-1]}", + "consent": "awaiting owner consent", + "deny": "refused — routed to Governor", + }[r["verdict"]]