Publish isolated guest tools

README.md

@@ -1,3 +1,7 @@
-# guest-tools
-
-Guest-safe tools for mesh agents; reviewed tools only
+# Guest tools
+No fleet/admin tools, host credentials, SSH relay aliases, or secretz dependencies.
+- `kev`: original yesno/ask/choose interface via own revocable gateway key. Host injects GPU credential; gateway cap applies.
+- `recall [text] [--with peer] [--recent N]`: only this guest's own a2a history.
+- `osmike-publish FILE`: commits a <=10MiB artifact to context/published in this guest's private repo and returns an owner-login link. No public share is implied.
+- `osmike-cron add --after SECONDS COMMAND ...`, `list`, `remove ID`: local guest jobs. Due jobs execute on the next awake poll. They cannot wake sleeping containers or schedule fleet work.
+Tools are fetched read-only from osmike/guest-tools. Guest credentials are limited to its bus identity, repo-only deploy key and revocable gateway key.

kev

@@ -0,0 +1,249 @@
+#!/usr/bin/env python3
+"""kev - free, fast (~30 ms) yes/no and choice decisions from KEV (kev.osmike.com).
+
+  kev yesno  "<state>" "<question>" [--exit] [--min P]
+  kev ask    "<state>" -q id:"question" [-q id2:"question2" ...] [--json]
+  kev choose "<state>" "<question>" opt:"description" opt2:"description" [...] [--json]
+
+State: a text describing the situation (keep it < ~55k chars; longer is cut to the last 55k).
+Use "-" as the state to read it from stdin, or --state-file FILE.
+
+Output:
+  yesno  -> "yes 0.82" or "no 0.31"  (the number is always p(yes); yes when p >= --min, default 0.5)
+           --exit: exit code 0 = yes, 1 = no, 2 = error (for shell `if kev yesno ... --exit; then`)
+  ask    -> one line per question: "<id> yes 0.82"
+  choose -> first line "<choice> <p>" (+ " near-tie" when the runner-up is within 0.15),
+            then every option with its probability, best first.
+Tips: KEV is sharper on several yes/no questions (`kev ask`) than on one 3-way choice.
+With near-ties, look at the probabilities and apply your own margin.
+
+Token: own revocable GUEST_GATEWAY_KEY; host injects the KEV credential.
+URL:   $OSMIKE_KEV_URL (default https://kev.osmike.com). Timeout 20 s. Exit 2 on any error.
+"""
+import argparse, json, os, re, sys, urllib.error, urllib.request
+
+DEFAULT_URL = "http://172.30.240.1:8770/kev"
+
+MAX_STATE = 55000
+TIMEOUT = 20
+ID_RE = re.compile(r"^[A-Za-z0-9_.-]{1,40}$")
+NEAR_TIE = 0.15
+
+
+class KevError(Exception):
+    pass
+
+
+def _cfg(name):
+    return 'http://172.30.240.1:8770/kev' if name == 'OSMIKE_KEV_URL' else ''
+
+
+def get_token():
+    token=os.environ.get('GUEST_GATEWAY_KEY','')
+    if not token: raise KevError('missing guest gateway key')
+    return token
+
+
+def call_kev(state, questions):
+    url = (_cfg("OSMIKE_KEV_URL") or DEFAULT_URL).rstrip("/") + "/v1/systemone"
+    if len(state) > MAX_STATE:
+        print(f"kev: state is {len(state)} chars, keeping the last {MAX_STATE}", file=sys.stderr)
+        state = state[-MAX_STATE:]
+    if not state.strip():
+        raise KevError("empty state")
+    body = json.dumps({"model": "kev-latest", "state": state, "questions": questions}).encode()
+    req = urllib.request.Request(url, data=body, method="POST", headers={
+        "Authorization": "Bearer " + get_token(), "Content-Type": "application/json"})
+    try:
+        with urllib.request.urlopen(req, timeout=TIMEOUT) as r:
+            data = json.loads(r.read())
+    except urllib.error.HTTPError as e:
+        try:
+            detail = e.read().decode("utf-8", "replace")[:300]
+        except Exception:
+            detail = ""
+        hint = {401: "token rejected", 422: "bad request (only noul/choice questions are valid)",
+                502: "KEV backend down (GPU box/tunnel)", 503: "KEV unavailable", 504: "KEV timeout"}
+        raise KevError(f"KEV HTTP {e.code} ({hint.get(e.code, 'error')}) {detail}".strip())
+    except Exception as e:
+        raise KevError(f"KEV unreachable ({url}): {e.__class__.__name__}: {e}")
+    answers = data.get("answers")
+    if not isinstance(answers, dict):
+        raise KevError("KEV reply has no answers")
+    return answers
+
+
+# ------------------------------------------------------------ parsing ---
+def split_spec(spec, n):
+    """'id:text' -> (id, text). The id must be followed directly by the text (no space after ':'),
+    so a question like "Note: is it broken?" keeps its colon and gets an automatic id q<n>."""
+    if ":" in spec:
+        head, tail = spec.split(":", 1)
+        if ID_RE.match(head) and tail.strip() and not tail[:1].isspace():
+            return head, tail.strip()
+    return f"q{n}", spec.strip()
+
+
+def parse_questions(specs):
+    qs, seen = {}, set()
+    for i, s in enumerate(specs, 1):
+        qid, text = split_spec(s, i)
+        if not text:
+            raise KevError(f"empty question in {s!r}")
+        if qid in seen:
+            raise KevError(f"duplicate question id {qid!r}")
+        seen.add(qid)
+        qs[qid] = {"type": "noul", "instructions": text}
+    if not qs:
+        raise KevError("no questions (use -q id:\"question\")")
+    return qs
+
+
+def parse_options(specs):
+    crit = {}
+    for i, s in enumerate(specs, 1):
+        if ":" in s:
+            name, desc = s.split(":", 1)
+            name, desc = name.strip(), desc.strip()
+        else:
+            name, desc = s.strip(), s.strip()
+        if not name:
+            raise KevError(f"empty option name in {s!r}")
+        if name in crit:
+            raise KevError(f"duplicate option {name!r}")
+        crit[name] = desc or name
+    if len(crit) < 2:
+        raise KevError("choose needs at least 2 options (opt:\"description\")")
+    return crit
+
+
+def read_state(arg, state_file):
+    if state_file:
+        try:
+            return open(state_file, encoding="utf-8", errors="replace").read()
+        except OSError as e:
+            raise KevError(f"cannot read state file: {e}")
+    if arg == "-":
+        return sys.stdin.read()
+    if arg is None:
+        raise KevError("missing state")
+    return arg
+
+
+def noul_p(ans):
+    try:
+        return float(ans["noul"])
+    except Exception:
+        raise KevError(f"unexpected KEV answer: {json.dumps(ans)[:200]}")
+
+
+def fmt_yes(p, thr):
+    return f"{'yes' if p >= thr else 'no'} {p:.2f}"
+
+
+def rank_choice(ans):
+    probs = ans.get("probabilities") or {}
+    ranked = sorted(((k, float(v)) for k, v in probs.items()), key=lambda kv: -kv[1])
+    if not ranked and ans.get("choice"):
+        ranked = [(ans["choice"], float(ans.get("confidence") or 0))]
+    if not ranked:
+        raise KevError(f"unexpected KEV answer: {json.dumps(ans)[:200]}")
+    return ranked
+
+
+# --------------------------------------------------------------- main ---
+def build_parser():
+    p = argparse.ArgumentParser(prog="kev", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    sub = p.add_subparsers(dest="cmd", required=True)
+
+    y = sub.add_parser("yesno", help="one yes/no question")
+    y.add_argument("state", nargs="?")
+    y.add_argument("question", nargs="?")
+    y.add_argument("--exit", action="store_true", help="exit 0 = yes, 1 = no, 2 = error")
+    y.add_argument("--min", type=float, default=0.5, help="p(yes) threshold for 'yes' (default 0.5)")
+    y.add_argument("--json", action="store_true")
+    y.add_argument("--state-file")
+
+    a = sub.add_parser("ask", help="several yes/no questions at once")
+    a.add_argument("state", nargs="?")
+    a.add_argument("-q", "--question", action="append", default=[], metavar='ID:"QUESTION"')
+    a.add_argument("--min", type=float, default=0.5)
+    a.add_argument("--json", action="store_true")
+    a.add_argument("--state-file")
+
+    c = sub.add_parser("choose", help="pick one of several options")
+    c.add_argument("state", nargs="?")
+    c.add_argument("question", nargs="?")
+    c.add_argument("options", nargs="*", metavar='OPT:"DESCRIPTION"')
+    c.add_argument("--json", action="store_true")
+    c.add_argument("--state-file")
+    return p
+
+
+def main(argv=None):
+    p = build_parser()
+    args = p.parse_args(argv)
+    # with --state-file the positional "state" slot holds the next argument
+    if args.state_file:
+        pos = [args.state] + ([args.question] if hasattr(args, "question") else []) + list(getattr(args, "options", []))
+        pos = [x for x in pos if x is not None]
+        args.state = None
+        if args.cmd == "yesno":
+            args.question = pos[0] if pos else None
+        elif args.cmd == "choose":
+            args.question = pos[0] if pos else None
+            args.options = pos[1:]
+        elif pos:
+            raise KevError(f"unexpected argument {pos[0]!r} (use -q)")
+    try:
+        state = read_state(args.state, args.state_file)
+        if args.cmd == "yesno":
+            if not args.question:
+                raise KevError('usage: kev yesno "<state>" "<question>"')
+            ans = call_kev(state, {"q": {"type": "noul", "instructions": args.question}})
+            pv = noul_p(ans.get("q") or {})
+            yes = pv >= args.min
+            print(json.dumps({"answer": "yes" if yes else "no", "p_yes": round(pv, 4)}) if args.json else fmt_yes(pv, args.min))
+            return (0 if yes else 1) if args.exit else 0
+        if args.cmd == "ask":
+            qs = parse_questions(args.question)
+            ans = call_kev(state, qs)
+            out = {}
+            for qid in qs:
+                pv = noul_p(ans.get(qid) or {})
+                out[qid] = {"answer": "yes" if pv >= args.min else "no", "p_yes": round(pv, 4)}
+            if args.json:
+                print(json.dumps(out))
+            else:
+                for qid, v in out.items():
+                    print(f"{qid} {v['answer']} {v['p_yes']:.2f}")
+            return 0
+        if args.cmd == "choose":
+            if not args.question:
+                raise KevError('usage: kev choose "<state>" "<question>" opt:"desc" opt2:"desc"')
+            crit = parse_options(args.options)
+            ans = call_kev(state, {"c": {"type": "choice", "instructions": args.question, "criteria": crit}})
+            ranked = rank_choice(ans.get("c") or {})
+            tie = len(ranked) > 1 and ranked[0][1] - ranked[1][1] < NEAR_TIE
+            if args.json:
+                print(json.dumps({"choice": ranked[0][0], "p": round(ranked[0][1], 4), "near_tie": tie,
+                                  "probabilities": {k: round(v, 4) for k, v in ranked}}))
+            else:
+                print(f"{ranked[0][0]} {ranked[0][1]:.2f}" + (" near-tie" if tie else ""))
+                for k, v in ranked:
+                    print(f"  {k} {v:.2f}")
+            return 0
+    except KevError as e:
+        print(f"kev: {e}", file=sys.stderr)
+        return 2
+    except KeyboardInterrupt:
+        return 2
+    return 2
+
+
+if __name__ == "__main__":
+    try:
+        sys.exit(main())
+    except KevError as e:
+        print(f"kev: {e}", file=sys.stderr)
+        sys.exit(2)

osmike-cron

@@ -0,0 +1,14 @@
+#!/usr/bin/env python3
+"""Guest-only wake-time jobs. Due jobs run when your guest is awake; never wakes the host or uses fleet scheduler."""
+import argparse,json,os,secrets,time
+from pathlib import Path
+p=argparse.ArgumentParser(); sub=p.add_subparsers(dest='action',required=True); add=sub.add_parser('add'); add.add_argument('--after',type=int,required=True,help='seconds'); add.add_argument('command',nargs='+'); sub.add_parser('list'); rm=sub.add_parser('remove'); rm.add_argument('id'); a=p.parse_args()
+d=Path('/data/guest-jobs'); d.mkdir(mode=0o700,exist_ok=True)
+if a.action=='add':
+ if a.after<1: p.error('after must be positive')
+ ident=secrets.token_hex(8); f=d/(ident+'.json'); f.write_text(json.dumps({'due':time.time()+a.after,'command':a.command})); print(ident)
+elif a.action=='list':
+ for f in d.glob('*.json'): print(f.stem,f.read_text())
+else:
+ if not all(c in '0123456789abcdef' for c in a.id): p.error('invalid id')
+ (d/(a.id+'.json')).unlink(missing_ok=True)

osmike-publish

@@ -0,0 +1,14 @@
+#!/usr/bin/env python3
+"""Publish to YOUR PRIVATE repo. Printed URL requires the owner's Gitea login."""
+import os,shutil,subprocess,sys,secrets
+from pathlib import Path
+if len(sys.argv)!=2: sys.exit('usage: osmike-publish FILE (private owner repo link)')
+p=Path(sys.argv[1]).resolve(); root=Path('/data/workspace'); dest=root/'context/published'; dest.mkdir(parents=True,exist_ok=True)
+if not p.is_file() or p.stat().st_size>10*1024*1024: sys.exit('file must exist and be at most 10MiB')
+name=secrets.token_hex(8)+'-'+p.name; shutil.copyfile(p,dest/name)
+for args in [['add','--','context/published/'+name],['-c','user.name='+os.environ['AGENT_NAME'],'-c','user.email='+os.environ['AGENT_NAME']+'@guests.osmike.com','commit','-m','Publish private guest artifact'],['push','origin','HEAD:main']]:
+ r=subprocess.run(['git','-C',str(root)]+args,capture_output=True)
+ if r.returncode: sys.exit('repo publication failed')
+repo=os.environ['GUEST_REPO'].split(':8771/',1)[1].removesuffix('.git')
+from urllib.parse import quote
+print('https://git.osmike.com/'+repo+'/src/branch/main/context/published/'+quote(name))

recall

@@ -0,0 +1,11 @@
+#!/usr/bin/env python3
+"""Search only your own a2a history, authenticated with your guest key."""
+import argparse,json,os,urllib.parse,urllib.request
+p=argparse.ArgumentParser(); p.add_argument('query',nargs='?',default=''); p.add_argument('--with',dest='peer'); p.add_argument('--recent',type=int,default=30); a=p.parse_args()
+url=os.environ['A2A_API']+'/v1/history?'+urllib.parse.urlencode({'limit':min(max(a.recent,1),200)})
+req=urllib.request.Request(url,headers={'X-API-KEY':os.environ['A2A_AGENT_KEY']})
+with urllib.request.urlopen(req,timeout=30) as r: rows=json.load(r).get('messages',[])
+for m in rows:
+ if a.peer and a.peer not in [m.get('from_agent'),m.get('to_agent')]: continue
+ if a.query.lower() not in m.get('message','').lower(): continue
+ print(json.dumps(m,ensure_ascii=False))