#!/usr/bin/env python3 """Oracle7 Workbench - your own Oracle7, on your own machine, with the diagram of everything it is. python oracle7.py --workbench [port] (default 8177; opens on http://127.0.0.1:/) What it does Ask talk to Oracle7; the diagram lights the path each answer took and shows why a decline happened Specialists the 267 admitted specialists and the five systems built the same way, each with its card Build scaffold a specialist, edit it here, run its dev exam as often as you like, its held-out exam once Grow admitted specialists join your model and appear on the diagram; a changed file sends one back to the exam Gaps (off until you turn it on) your declines, grouped into the specialists your model is missing Code describe a function by examples; the code organ composes it and runs it on your examples Film record any answer's path on the diagram as a video file, made in your browser Share package a specialist (its folder, card, exam verdict and hashes) for the community registry Privacy: the server listens on 127.0.0.1 only. Nothing is sent anywhere. The gap journal is off by default, lives in workbench/local/journal.jsonl when on, is never packaged, and the Gaps page deletes it in one click. Your specialists can only ADD answers: one runs only when the core declined for lack of a reading, never when the core answered and never when a check (false premise, private question, units...) stopped the answer. Stdlib only.""" from __future__ import annotations import hashlib import io import json import os import re import shutil import subprocess import sys import tempfile import threading import time import zipfile from collections import Counter, defaultdict from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, urlparse HERE = Path(__file__).resolve().parent LOCAL = Path(os.environ.get("ORACLE7_WORKBENCH_LOCAL", str(HERE / "local"))) SPECS = LOCAL / "specialists" OUTBOX = LOCAL / "outbox" JOURNAL = LOCAL / "journal.jsonl" CONVS = LOCAL / "conversations" SETTINGS = LOCAL / "settings.json" STATIC = HERE / "static" DATA = HERE / "data" for d in (SPECS, OUTBOX, CONVS): d.mkdir(parents=True, exist_ok=True) sys.path.insert(0, str(HERE)) import exam # noqa: E402 import sources # noqa: E402 import mcp # noqa: E402 import verify # noqa: E402 import sensors # noqa: E402 import registry # noqa: E402 ID = re.compile(r"^[a-z][a-z0-9-]{1,40}$") FNAME = re.compile(r"^[A-Za-z0-9_][A-Za-z0-9_.-]{0,60}\.(py|json|md|txt|csv)$") READING_DECLINES = ("act:clarify", "unverified:", "corpus_query", "factoid", "deliberation", "capability:", "arb:", "compose", "family_precedence:low_credit", "verified:", "cube:") CHECK_DECLINES = ("precision_guard", "same_name_guard", "weight:", "chat:unresolved") TYPES = {".html": "text/html; charset=utf-8", ".js": "text/javascript; charset=utf-8", ".css": "text/css; charset=utf-8", ".json": "application/json", ".svg": "image/svg+xml", ".png": "image/png", ".woff2": "font/woff2"} def settings(): try: s = json.loads(SETTINGS.read_text(encoding="utf-8")) except Exception: # noqa: BLE001 s = {} s.setdefault("journal", False) s.setdefault("keep_conversations", True) return s # ---------------------------------------------------------------- conversations (kept on this computer; one switch turns it off) CID = re.compile(r"^[a-z0-9]{6,40}$") def conv_path(cid): if not CID.match(cid or ""): raise ValueError("bad conversation id") return CONVS / ("%s.json" % cid) def conv_load(cid): p = conv_path(cid) return json.loads(p.read_text(encoding="utf-8")) if p.exists() else {"id": cid, "title": "", "created": int(time.time()), "turns": []} def conv_add(cid, turn): c = conv_load(cid) c["turns"].append(turn) c["updated"] = int(time.time()) if not c.get("title"): c["title"] = turn["q"][:70] conv_path(cid).write_text(json.dumps(c, indent=0), encoding="utf-8") def conv_list(): out = [] for p in CONVS.glob("*.json"): try: c = json.loads(p.read_text(encoding="utf-8")) out.append({"id": c["id"], "title": c.get("title") or "(untitled)", "updated": c.get("updated", c.get("created", 0)), "turns": len(c.get("turns", [])), "answered": sum(1 for t in c.get("turns", []) if t.get("status") == "answered")}) except Exception: # noqa: BLE001 pass return sorted(out, key=lambda x: -x["updated"]) def save_settings(s): SETTINGS.write_text(json.dumps(s, indent=1), encoding="utf-8") # ---------------------------------------------------------------- the path an answer took (names from the diagram only) SPEC_GROUP = { "Physical science": "acoustics additive chemistry densities decay electrical electronics energy fluids fuels materials mechanisms optics " "physics springs thermo units carbonmat carboncapture renewables hvac".split(), "Earth & life": "agriculture airquality anatomy biology botany climate ecology entomology forestry genetics geology hydrology marinebio " "medicine meteorology microbiology mycelial mycology ocean ornithology seismology soil water wildlife greenhouse geography".split(), "Computing": "algorithms cyber dataformats embedded git iot javascript linux networking python sql logic statistics calculus geometry".split(), "Space": "astronomy orbits planetary spacecomms spacecraft bodies".split(), "Humanities": "architecture arthistory economics government grammar law library linguistics literature music mythology poetry " "psychology sociology ushistory worldhistory".split(), } def spec_group(sid): if sid.startswith("rel-"): return "Relation readers" base = re.sub(r"-(v\d+|r\d+)$", "", sid).split("-")[0] for g, names in SPEC_GROUP.items(): if base in names: return g return "Everyday" def trace(path, response, status): p, resp = str(path or ""), str(response or "") s = [["L0", "Front door"], ["L1", None]] if "chat:and_parts" in p or "chat:parts" in p: s.append(["L3", "Two-part turns"]) if re.search(r"(^|[>|])chat:(resolved|unresolved)", p): s.append(["L3", "Follow-up resolver"]) if p.startswith("workbench:history"): return [["L0", "Front door"], ["L3", "Conversation engine"], ["L3", "Dialogue organ"], ["L4", "Speaker"], ["L0", "Front door"]] if p.startswith("workbench:help"): return [["L0", "Front door"], ["L5", "Self-explanation"], ["L4", "Speaker"], ["L0", "Front door"]] if p.startswith(("act:", "organ:converse")) or "arb:act_" in p: s.append(["L3", "Conversation acts"]) if "family_precedence" in p or "specialist_precedence" in p: s.append(["L2", "Family precedence"]) m = re.search(r"specialist:([a-z0-9-]+)", p) mine = re.match(r"yours:([a-z0-9-]+)", p) net = re.match(r"network\d*>([a-z0-9 _-]+)", p) if mine: s += [["L2", "Specialist call"], ["S", "Yours: " + mine.group(1)]] elif net and not m: s.append(["L2", "Specialist call"]) for tok in net.group(1).split(): if tok not in ("core_math", "validity"): s.append(["S", spec_group(tok)]) s.append(["E", "Math organ"]) elif m: g = spec_group(m.group(1)) s += [["L2", "Specialist call"], ["S", g]] if g == "Relation readers": s.append(["V", "Relation store"]) elif "exact_router" in p: s += [["L2", "Exact router"], ["V", "Relation store"]] elif "general_reader" in p: s += [["L2", "General reader"], ["V", "Relation store"]] elif "organ:relation" in p: s += [["L2", "Organ dispatch"], ["V", "Relation store"]] elif re.search(r"organ:code|code_organ|graded_code", p): s += [["L2", "Organ dispatch"], ["E", "Code organ"]] elif re.search(r"organ:math|network>|core_math|agent:maths|algebra:", p): s += [["L2", "Organ dispatch"], ["E", "Math organ"]] elif re.search(r"(^|[>|])op:", p): s += [["L2", "Organ dispatch"], ["E", "Calculator"]] elif "deliberation" in p: s += [["L2", "Deliberation"], ["V", "Deep store"]] elif "compose" in p: s += [["L2", "Compose"], ["V", "Deep store"]] elif re.search(r"factoid|cube|corpus_query|verified:", p): s += [["L2", "Factoid & reading"], ["V", "Deep store"]] elif re.search(r"capability:|arb:", p) and "arb:act_" not in p: s += [["L2", "Capability routes"], ["V", "Deep store"]] elif status != "answered" and p.startswith("act:clarify"): s.append(["L2", "Front parser"]) if p.startswith("same_name_guard"): s += [["L3", "Conversation engine"], ["L4", "Same-name check"], ["V", "Relation store"]] if re.match(r"precision_guard:(N1|C1|D1|N2|G3)_", p): s.append(["L4", "Answer checks"]) elif p.startswith("precision_guard:") or "same_name_guard:withheld" in p: s.append(["L4", "Final-answer checks" if re.match(r"precision_guard:(G\d|Q\d)", p) else "Precision guard"]) elif status != "answered" and not p.startswith("act:"): s.append(["L4", "Precision guard"]) if status == "answered": if "From the Wikidata relation store" in resp: s.append(["L4", "Receipts"]) s.append(["L4", "Speaker"]) s.append(["L0", "Front door"]) out = [] for x in s: if not out or out[-1] != x: out.append(x) return out def why_declined(path): p = str(path or "") if exam.CHECK.search(p): return "check", "A check stopped it: the answer would rest on a premise or value it could not verify." if "precision_guard:P" in p: return "reading", "It found a reply but could not verify it against a record, so it declined rather than guess." if p.startswith("act:clarify") or "Front parser" in p: return "parse", "It did not read the question's shape. Rephrasing as a direct question often works." if any(k in p for k in READING_DECLINES): return "reading", "Nothing it holds could verify an answer, so it declined rather than guess." return "other", "It declined rather than guess." # ---------------------------------------------------------------- what it can do (answered by the Workbench itself) HELP_TOPICS = [ ("ask", r"\b(ask|question|answer|decline|why did it|path)\b", "Ask anything in the box at the bottom left. Each answer lights the path it took on the diagram and shows its route. " "When Oracle7 cannot verify an answer it declines and says why (a check stopped it, it did not parse the question, or " "nothing it holds could verify it). 'show path' replays a path; 'route' writes the raw route to the log."), ("scan", r"\b(scan|my (?:pc|computer|laptop|machine))\b", "Start > Scan this computer (or type /scan). It reads this machine's system, CPU, memory, disks, uptime and the processes " "using the most memory - figures and process names only, never your files - and shows them. 'Make it a specialist' adds a " "reader, so 'how much disk space is free on my computer?' gets the value and the call it was read from."), ("connect", r"\b(connect|server|ssh|remote)\b", "Start > Connect a server (or /connect user@host). It uses your own SSH key and runs a fixed list of read-only commands " "(hostname, uptime, load, memory, disks, top processes, failed services). Name the server, then ask 'how much memory is " " using?'. A password is never asked for or stored."), ("files", r"\b(files?|upload|documents?|csv|spreadsheet|notes|tables?)\b", "Start > Add files (or /files): drop .csv, .tsv, .txt or .md files. Tables answer 'what is the of ?' with " "the file, row and column; text answers 'what do my notes say about ...?' by quoting the sentence with its file and line. " "The files are copied into the reader's own folder on this machine."), ("build", r"\b(build|make|create|write|new)\b.*\bspecialists?\b|\bspecialist\b|\bexams?\b", "Build: 'New specialist' scaffolds a folder (manifest, organ, panel of test items, scorer) that you edit here, or 'Add the " "worked example'. Run the dev exam as often as you like and the held-out exam once. It joins your model only if it gets " "more right than the core alone, adds no false yes, has few wrong answers, stays in its lane on 924 general questions, " "leaves identity questions alone and is fast. Commands: /new [words], /exam dev, /exam heldout."), ("grow", r"\b(grow|improve|self.improv|learn|diagram)\b", "Your model grows only through exams: an admitted specialist appears on the diagram's Specialists ring and answers where " "the core declined. Change one of its files and it leaves your model until it passes a new held-out set. With the gap " "journal on (Gaps, or /gaps on) your declines are grouped into the specialists you are missing, and 'Start a specialist' " "turns a group into a new specialist with your questions as its first dev items."), ("code", r"\b(code|coding|program|function|python|script|tests?)\b", "Code (or /code ): ask for a function, e.g. 'check whether a string is a palindrome'. The code organ " "hands over only programs it has run: a small website, palindromes, factorials, vowel and word counts, word frequencies, " "HTTP routes, regex extraction, memoisation, a bank-account class, matrix multiply, fizzbuzz, fibonacci, string reversal, " "JSON round-trips, concurrent work, binary search, tower of hanoi, quicksort, list sums, prime tests and two small games. " "Check what it hands over with tests of your own: Run tests runs them in a separate Python process with a 10 second limit."), ("film", r"\b(film|video|record|movie|clip)\b", "Film (or /film): record answers as a video of the diagram tracing each path beside the question, answer and path. It is " "made in your browser as a .webm file; 'film this' under any answer records just that one."), ("voice", r"\b(eleven ?labs|narrat\w*|voice|voiceover|music|sound|audio|image|images|picture|illustrat\w*)\b", "Film > Narration, sound and images (optional): connect your own ElevenLabs account with its API key to narrate each filmed " "answer in a voice you choose, lay an ambient sound you describe under it, and open each answer with an illustration " "(Gemini 2.5 Flash Image or GPT Image, through ElevenLabs; image generation needs their Pro plan). Generated images are " "labelled as generated in the film and saved in workbench/local/assets. This is the only feature that sends anything off " "your computer: the narration text, the sound description and the image prompts go to ElevenLabs. The key stays in the page " "unless you tick 'remember on this computer'."), ("share", r"\b(share|registry|import|package|hugging ?face)\b", "Share (or /share ): packages a specialist you built - its folder, exam verdict and the SHA-256 of every file - and shows " "the command to upload it to the community registry with your own Hugging Face account. Import takes a shared .zip, which " "must pass its held-out exam on your machine before it joins. Readers of your computer, servers and files are never packaged."), ("specialists", r"\b(specialists?|systems?|deepspace|trafficcop|chameleon|vision|artist)\b", "Specialists lists the 267 admitted specialists (106 by field plus 161 relation readers), each with its held-out score and " "card; search it or type /specialists . Systems describes the five systems built the same way: DeepSpace, Trafficcop, " "Chameleon (builds websites), Vision (sees and reads) and Artist (makes images by citation)."), ("mcp", r"\b(mcp|tools?|model context protocol|plug ?in|integrat\w*)\b", "Start > Connect an MCP server: give the command that starts it (for example npx -y @modelcontextprotocol/server-filesystem " "with a folder) or its URL. The Workbench lists its tools; tick the ones to use, add a few example questions if you like, and " "it builds an adaptive specialist. It routes each question to the tool it fits, fills the arguments from the question (or " "says which one is missing), answers with the tool's result and the exact call it made, and learns from your right/wrong " "marks under each answer. Questions routed to a server are sent to it, so connect servers you trust."), ("history", r"\b(history|previous|past|earlier|old|saved) (?:conversations?|chats?)\b|\bconversations?\b|\bremember\w*\b|\bmemory of\b", "Conversations (the panel on the left; the button at the top opens and collapses it): every conversation is kept on this " "computer. Open one and carry on - the core picks up its context, so follow-ups still work. Rename, search or delete them, " "or switch keeping off. 'Make specialist' turns a conversation into a temporary specialist (7 days unless you choose) that " "answers 'what did we say earlier about...?' and repeated questions by quoting the earlier answer with its date and route. " "It is checked like any reader before it joins, and leaves your model when it expires."), ("verify", r"\b(verify|verified|hash\w*|blockchain|chain|solana|on.chain|tamper\w*|authentic\w*|merkle)\b", "Specialists > Verify this install (or /verify): recomputes the SHA-256 of every shipped part - all 307 specialist folders, " "the admitted fold chain, this workbench, the launcher and the store manifest - and rebuilds the Merkle root published in " "RELEASE_HASHES.json. A changed or missing specialist shows by name. 'Check the anchor on Solana' then asks a public Solana " "RPC (read-only; the only network call) for the release's anchor transaction and confirms its memo carries the same root, " "so your copy is provably the one released. Stores are checked against PACK_MANIFEST.json by oracle7.py --setup."), ("sensors", r"\b(sensors?|iot|esp32|arduino|mqtt|serial|usb|thermometer|probe|telemetry|readings?|home assistant|shelly)\b", "Start > Connect a sensor, or the Sensors tab (or /sensors): an HTTP device or hub that serves JSON or Prometheus metrics " "(ESP32, Shelly, Tasmota, Home Assistant, weather stations), an MQTT topic, a USB/serial device that prints readings, a log " "file a logger appends to, or a simulated garden node to try it without hardware. Test shows the numbers it found and what " "each one measures; Add starts polling it (every 30 s by default, 24 h kept on this computer) and makes a reader you can ask: " "'what is the soil moisture on garden node?', 'temperature on my sensor today?'. Every answer gives the value, its age, its " "source and whether it was measured or simulated; a value outside its physical range is rejected and said so."), ("terminal", r"\b(cli|terminal|command line|shell|script|json output)\b", "Everything that makes sense in a terminal works there too, with the same code: python oracle7.py help lists it. " "ask \"question\" (answer, status and path), chat, specialists [word], card , systems, verify [--chain], mine, new , " "example, exam dev|heldout, scan, sensors, sensor-test
, history. Add --json for machine-readable output."), ("registry", r"\b(oracle7 token|\$?oracle7 registry|registry|register|attest\w*|vouch\w*|on.chain|token|holders?)\b", "Share > Register on Solana: one specialist, one $ORACLE7. The creator of an admitted specialist signs, in their own wallet (Phantom), one Solana transaction that burns exactly 1 $ORACLE7 and carries, in a memo, the " "specialist's fingerprint - the SHA-256 of its files, its held-out verdict and its held-out set, and the release it was examined " "on. Anyone who imports it and sees it pass its held-out exam on their machine can Attest the same way. Check reads every " "registration and attestation back from the chain, confirms it matches the specialist's files now, and shows each signer's " "wallet and how much $ORACLE7 it holds. The records travel with the specialist when it is shared. No key is ever seen or " "stored here; the burn is the only token movement, approved in the creator's wallet. Attesting is free."), ("privacy", r"\b(privacy|private|data|sent|stored|safe)\b", "The Workbench answers on 127.0.0.1 only and sends nothing anywhere. The gap journal is off until you turn it on, stays in " "workbench/local/journal.jsonl, and one click deletes it. Conversations are kept in workbench/local/conversations so you can " "go back to them; switch that off or delete them in the Conversations panel."), ] COMMANDS = ["/help - this guide", "/scan - read this computer", "/connect user@host [port] [key] - read a server", "/files - add files", "/mcp - connect an MCP server", "/sensors - connect and watch sensors", "/new [words] - a new specialist", "/exam dev|heldout - run an exam", "/specialists - find one", "/systems - the five systems", "/gaps on|off - the gap journal", "/code - compose a function", "/film - film the answers", "/share - package one", "/verify - check this install against the published hashes", "/registry - its on-chain registrations and attestations", "/history - open the conversations panel", "/new-chat - start a new conversation", "/remember [days] - make this conversation a temporary specialist", "/clear - clear the conversation"] HELP_Q = re.compile(r"^\s*(?:/help\b|help\b|what (?:can|do) (?:you|it|the workbench|oracle7) do|what (?:are|is) (?:your|the|its) " r"(?:features|commands|abilities)|(?:list|show) (?:me )?(?:the |your |all )?(?:features|commands)|what commands|" r"how do i (?:use|start)\b|how does (?:this|the workbench) work)", re.I) HOWTO_Q = re.compile(r"^\s*(?:how (?:do|can|would) i|how to|can i|can you|is it possible to)\b", re.I) SENSORS_ALL = re.compile(r"\b(?:my|our|the|all(?: the| my)?) sensors\b|\bsensor readings\b|\ball (?:my )?readings\b", re.I) PRONOUN = re.compile(r"\b(it|its|that one|this one|them|they)\b", re.I) HISTORY_Q = re.compile(r"^\s*(?:what|which) (?:did|was|were) (?:i|my) (?:just |first |last |previous |earlier )?(?:ask|asked|question|questions)\b|" r"^\s*what (?:was|were) my (first|last|previous|earlier|latest) questions?\b|" r"^\s*what did you (?:just )?(?:say|answer|tell me)\b|^\s*what (?:have|did) (?:i|we) (?:asked|ask|talk(?:ed)? about|discuss(?:ed)?)\b", re.I) def history_reply(text, sid): """the conversation itself is kept by the Workbench, so it answers questions about it: what was asked first or last, what it just said, what we have covered""" t = str(text or "") if not HISTORY_Q.search(t): return None try: c = conv_load(sid) except ValueError: return None asked = [x for x in c.get("turns", []) if not str(x.get("path", "")).startswith("workbench:")] topic = re.search(r"\babout (.+?)[?.!]*$", t, re.I) if topic: want = set(sources._words(topic.group(1))) - sources.RECALL_FILLER hits = [x for x in asked if want & set(sources._words(x["q"] + " " + str(x.get("response", ""))))] if not hits: return None # not in this conversation: a saved conversation may recall it return "\n\n".join("At %s you asked \"%s\". I %s: %s" % (time.strftime("%H:%M", time.localtime(x.get("t", 0))), x["q"], "answered" if x.get("status") == "answered" else "declined", str(x.get("response", ""))[:400]) for x in hits[-3:]) if not asked: return "Nothing has been asked in this conversation yet." when = lambda x: time.strftime("%H:%M", time.localtime(x.get("t", 0))) if re.search(r"\bfirst\b", t, re.I): x = asked[0] return "Your first question in this conversation (%s) was: \"%s\"" % (when(x), x["q"]) if re.search(r"\bwhat did you (?:just )?(?:say|answer|tell me)\b", t, re.I): x = asked[-1] return "I said (%s), to \"%s\": %s" % (when(x), x["q"], str(x.get("response", ""))[:600]) if re.search(r"\b(just|last|previous|latest)\b", t, re.I): x = asked[-1] return "Your last question (%s) was: \"%s\"" % (when(x), x["q"]) lines = ["%d. %s - %s" % (i, x["q"], "answered" if x.get("status") == "answered" else "declined") for i, x in enumerate(asked[-12:], max(1, len(asked) - 11))] return "In this conversation you have asked %d question%s:\n%s" % (len(asked), "" if len(asked) == 1 else "s", "\n".join(lines)) def help_reply(text): t = str(text or "") if HELP_Q.search(t): body = "\n\n".join("%s. %s" % (k.capitalize(), v) for k, _, v in HELP_TOPICS) return ("Here is everything the Workbench does. The core model answers questions from sourced records and declines what " "it cannot verify. Around it you can read your computer, servers and files, build and examine specialists, find " "your gaps, compose code, film answers and share what you build.\n\n" + body + "\n\nCommands:\n" + "\n".join(COMMANDS)) if HOWTO_Q.search(t): first = ("registry", "terminal", "sensors", "mcp", "voice", "film", "verify", "history", "files", "scan", "code", "share", "grow", "build", "connect", "specialists", "privacy", "ask") topics = sorted(HELP_TOPICS, key=lambda x: first.index(x[0]) if x[0] in first else 99) hits = [v for k, pat, v in topics if re.search(pat, t, re.I)] if hits: return "\n\n".join(hits[:2]) return None # ---------------------------------------------------------------- the engine + your specialists (fill declines only) class Workbench: def __init__(self, engine): self.eng = engine self.lock = threading.Lock() self.mine = {} # id -> exam.Loaded (admitted and unchanged since its exam) self.last = {} # session -> (your specialist id, subject) of its last answer, for follow-ups self.warm = set() # sessions whose context the core holds in this run self.reload() def reload(self): mine = {} for d in sorted(SPECS.iterdir()) if SPECS.exists() else []: st = exam.status(d) if st["state"] == "admitted": try: mine[d.name] = exam.load_specialist(d) except Exception as e: # noqa: BLE001 print("workbench: %s did not load: %s" % (d.name, e), file=sys.stderr) self.mine = mine def chat(self, text, sid): h = help_reply(text) if h: return {"status": "answered", "response": h, "path": "workbench:help"} h = history_reply(text, sid) if h: return {"status": "answered", "response": h, "path": "workbench:history"} # a follow-up to an answer from one of your readers ("what shelf is it on?") is read against that answer's subject prev = self.last.get(sid) if prev and PRONOUN.search(text): sp = self.mine.get(prev[0]) if sp is not None: a = sp.reply(PRONOUN.sub(prev[1], text, count=1), sid) if a and a.get("status") == "answered": a = dict(a) a["path"] = "yours:%s>followup>%s" % (prev[0], a.get("path", "")) return self.remember(text, sid, a) # questions ADDRESSED to your own things (my computer, my notes, a named server, your table's rows) go to their # readers first: the core cannot know them, and its general answer would not be about them. The readers' lane # check bounds how often that can happen on general questions. now = time.time() live = {i: sp for i, sp in self.mine.items() if not sp.m.get("expires") or float(sp.m["expires"]) > now} if SENSORS_ALL.search(text): # "my sensors": every connected sensor answers, one line each lines = [] for i, sp in live.items(): if sp.m.get("organ", {}).get("module", "").startswith("organ_sensor"): a = sp.organ.answer("%s readings %s" % (sp.organ.name, text), sid) if a and a.get("status") == "answered": lines.append(a["response"]) if lines: return self.remember(text, sid, {"status": "answered", "response": "\n".join(lines), "path": "yours:sensors>all"}) for i, sp in live.items(): if sp.m.get("source"): a = sp.reply(text, sid) if a and a.get("status") == "answered": a = dict(a) a["path"] = "yours:%s>%s" % (i, a.get("path", "")) return self.remember(text, sid, a) if sid not in self.warm: self.warm.add(sid) try: prior = [x for x in conv_load(sid).get("turns", []) if x.get("status") == "answered" and not str(x.get("path", "")).startswith(("workbench:", "yours:"))][-2:] except ValueError: prior = [] for x in prior: # replies discarded: this only restores what the core remembers try: self.eng.chat(x["q"], session_id="wb-" + sid) except Exception: # noqa: BLE001 pass r = self.eng.chat(text, session_id="wb-" + sid) self.last.pop(sid, None) # the core answered (or declined): follow-ups are its own path = str(r.get("path", "")) if r.get("status") != "answered" and live and exam.fillable(path): for i, sp in live.items(): a = sp.reply(text, sid) if a and a.get("status") == "answered": a = dict(a) a["path"] = "yours:%s>%s" % (i, a.get("path", "")) a["declined_by"] = path return self.remember(text, sid, a) return r def remember(self, text, sid, a): """hand an answer from your specialist back to the core's conversation memory, so a follow-up ("and in winter?", "what did I just ask?") reads against it like any answer the core gave""" m = re.match(r"^yours:([a-z0-9-]+)>", str(a.get("path", ""))) if m and a.get("subject"): self.last[sid] = (m.group(1), str(a["subject"])) def upd(): try: import chat_context_v2 as cc cc.update("wb-" + sid, text, a) except Exception: # noqa: BLE001 pass try: self.eng._ex.submit(upd).result(timeout=5) except Exception: # noqa: BLE001 pass return a WB = None def specialists_payload(): base = json.loads((DATA / "specialists.json").read_text(encoding="utf-8")) if (DATA / "specialists.json").exists() else {"items": []} yours = [] for d in sorted(SPECS.iterdir()): if d.is_dir(): st = exam.status(d) m = exam.read_manifest(d) or {} yours.append({"id": d.name, "name": m.get("name", d.name), "state": st["state"], "note": st.get("note", ""), "verdict": st.get("verdict")}) base["yours"] = yours return base def gaps(): """declines from the journal, grouped by what is missing: the decline kind, then the domain words the questions share""" rows = [] if JOURNAL.exists(): for line in JOURNAL.read_text(encoding="utf-8").splitlines(): try: rows.append(json.loads(line)) except Exception: # noqa: BLE001 pass dec = [r for r in rows if r.get("status") != "answered"] stop = set("the a an of in on at to is was were are be by for with from and or what which who whom whose where when why how " "did does do it its this that these those i you he she they we my your his her their our me him them as than then " "there here can could would should will shall may might must much many most more about into over after before " "name tell give find please".split()) groups = defaultdict(list) for r in dec: words = [w for w in re.findall(r"[a-z][a-z-]{2,}", r["q"].lower()) if w not in stop] groups[r.get("why", "other")].append((r["q"], words)) out = [] for kind, items in groups.items(): c = Counter(w for _, ws in items for w in set(ws)) themes = [] for w, n in c.most_common(8): if n < 2: break ex = [q for q, ws in items if w in ws][:4] themes.append({"word": w, "count": n, "examples": ex}) out.append({"kind": kind, "count": len(items), "themes": themes, "examples": [q for q, _ in items[:5]]}) out.sort(key=lambda g: -g["count"]) return {"on": settings().get("journal", False), "turns": len(rows), "declined": len(dec), "groups": out} def scaffold(sid, name, words, examples=()): d = SPECS / sid if d.exists(): raise ValueError("a specialist named %s already exists" % sid) shutil.copytree(HERE / "template", d) for f in d.iterdir(): if f.is_file(): cue = "|".join(w.replace("-", "[- ]") for w in words) or "(?!x)x" # words are [a-z0-9-] only: safe in JSON and regex t = f.read_text(encoding="utf-8").replace("__ID__", sid).replace("__NAME__", name.replace('"', "'")).replace( "__CUE__", cue).replace("__MODULE__", "organ_" + sid.replace("-", "_")) f.write_text(t, encoding="utf-8") (d / "organ.py").rename(d / ("organ_%s.py" % sid.replace("-", "_"))) if examples: # questions from your Gaps page become dev items to answer pf = d / "panel.json" panel = json.loads(pf.read_text(encoding="utf-8")) panel["dev"] = [{"q": q, "a": "", "valid": True, "kind": "fact"} for q in examples] + panel["dev"] pf.write_text(json.dumps(panel, indent=1), encoding="utf-8") return d def package(sid): d = SPECS / sid st = exam.status(d) if (exam.read_manifest(d) or {}).get("source"): raise ValueError("source readers read your own computer, servers or files, so they are not packaged for sharing") if st["state"] != "admitted": raise ValueError("only an admitted specialist is packaged (this one is %s); pass its held-out exam first" % st["state"]) files = sorted(p for p in d.rglob("*") if p.is_file() and "__pycache__" not in p.parts) sums = {str(p.relative_to(d)).replace("\\", "/"): hashlib.sha256(p.read_bytes()).hexdigest() for p in files} man = {"id": sid, "state": st["state"], "verdict": st.get("verdict"), "files": sums, "packaged": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "format": "oracle7-specialist/1"} h = hashlib.sha256(json.dumps(sums, sort_keys=True).encode()).hexdigest()[:12] out = OUTBOX / ("%s-%s.zip" % (sid, h)) with zipfile.ZipFile(out, "w", zipfile.ZIP_DEFLATED) as z: for p in files: z.write(p, "%s/%s" % (sid, p.relative_to(d).as_posix())) z.writestr("%s/SHARE_MANIFEST.json" % sid, json.dumps(man, indent=1)) return {"zip": str(out), "sha12": h, "state": st["state"], "command": "huggingface-cli upload Prometheus7/oracle7-specialists \"%s\" specialists/%s --repo-type dataset" % (out, out.name)} def import_zip(path): p = Path(path).expanduser() if not p.is_file() or p.suffix != ".zip": raise ValueError("give the path of a .zip made by Share") with zipfile.ZipFile(p) as z: names = z.namelist() top = {n.split("/")[0] for n in names} if len(top) != 1: raise ValueError("a shared specialist is one folder") sid = top.pop() if not ID.match(sid) or (SPECS / sid).exists(): raise ValueError("bad or existing specialist id: %s" % sid) for n in names: parts = n.split("/") if ".." in parts or n.startswith("/") or (len(parts) > 1 and parts[-1] and not FNAME.match(parts[-1])): raise ValueError("refused file in package: %s" % n) man = json.loads(z.read("%s/SHARE_MANIFEST.json" % sid)) tmp = Path(tempfile.mkdtemp(prefix="o7import-")) z.extractall(tmp) src = tmp / sid for rel, h in man["files"].items(): if hashlib.sha256((src / rel).read_bytes()).hexdigest() != h: shutil.rmtree(tmp) raise ValueError("%s does not match its published hash" % rel) (src / "SHARE_MANIFEST.json").unlink() for v in src.glob("verdict_*.json"): # their verdict is theirs: it takes the exam again here v.rename(v.with_name("shared_" + v.name)) shutil.move(str(src), str(SPECS / sid)) shutil.rmtree(tmp, ignore_errors=True) return {"id": sid, "note": "imported; it joins your model only after it passes its held-out exam on this machine"} def slug(s): return re.sub(r"-+", "-", re.sub(r"[^a-z0-9]+", "-", str(s).lower())).strip("-")[:30] or "x" def machine_probes(rd, on): """questions whose answers the readings already hold, in parts that do not change between two reads""" by = {(r["entity"].split()[0], r["signal"]): r for r in rd} P = [] if ("cpu", "cores") in by: P.append(("How many cores does %s have?" % on, str(by[("cpu", "cores")]["value"]))) if ("system", "hostname") in by: P.append(("What is the hostname of %s?" % on, str(by[("system", "hostname")]["value"]))) if ("memory", "used") in by: P.append(("How much memory is %s using?" % on, by[("memory", "used")]["text"].split(" used of ")[1].split(" (")[0])) if ("disk", "free") in by: P.append(("How much disk space is free on %s?" % on, " free of ")) if ("system", "uptime") in by: P.append(("How long has %s been up?" % on, "up ")) if ("processes", "top_memory") in by: P.append(("What is using the most memory on %s?" % on, "using the most memory")) P += [("How much memory does a computer need?", None), ("What is the capital of France?", None), ("Who wrote Hamlet?", None)] return P def file_probes(d): org = sources.FilesOrgan(folder=d.name + "/files") org.load() P = [] for fname, hdr, rows in org.tables: keys = [r[0].strip() for r in rows if r and r[0].strip()] for r in rows[:: max(1, len(rows) // 6)][:6]: if r and r[0].strip() and keys.count(r[0].strip()) == 1: for j in range(1, min(len(hdr), len(r))): if hdr[j].strip() and r[j].strip(): P.append(("What is the %s of %s?" % (hdr[j].strip(), r[0].strip()), r[j].strip())) break for fname, i, sent in org.lines[:: max(1, len(org.lines) // 5)][:5]: ws = [w for w in sources._words(sent) if len(w) > 4][:3] if len(ws) >= 2: P.append(("What do my notes say about %s?" % " ".join(ws), sent[:30])) P += [("What is the capital of France?", None), ("Who wrote Hamlet?", None)] return P def run_code(code, tests, timeout=10): """your function + your tests, in a separate Python process (isolated mode, a temp dir, a time limit)""" with tempfile.TemporaryDirectory(prefix="o7code-") as t: f = Path(t) / "run.py" f.write_text(code + "\n\n" + tests + "\nprint('ALL TESTS PASSED')\n", encoding="utf-8") try: r = subprocess.run([sys.executable, "-I", str(f)], cwd=t, capture_output=True, text=True, timeout=timeout) out = (r.stdout + r.stderr)[-4000:] return {"ok": r.returncode == 0 and "ALL TESTS PASSED" in r.stdout, "output": out} except subprocess.TimeoutExpired: return {"ok": False, "output": "stopped after %d s" % timeout} # ---------------------------------------------------------------- optional: ElevenLabs for film narration and sound (your key) # The ONLY feature that sends anything off this machine, and only when you connect it: the text of the narration (or the # sound description) goes to api.elevenlabs.io with your key; the audio comes back into your film. The key is kept in this # page's memory, or in workbench/local/settings.json if you tick "remember on this computer". Off by default. EL_API = "https://api.elevenlabs.io/v1" def el_key(b): k = str(b.get("key") or settings().get("elevenlabs_key") or "").strip() if not k: raise ValueError("connect ElevenLabs first (Film tab): paste your API key") if not re.match(r"^[A-Za-z0-9_\-]{20,128}$", k): raise ValueError("that does not look like an ElevenLabs API key") return k def el_call(path, key, body=None, accept="application/json", timeout=90): import urllib.error import urllib.request req = urllib.request.Request(EL_API + path, data=None if body is None else json.dumps(body).encode(), headers={"xi-api-key": key, "Content-Type": "application/json", "Accept": accept}, method="GET" if body is None else "POST") try: with urllib.request.urlopen(req, timeout=timeout) as r: return r.read() except urllib.error.HTTPError as e: raw = e.read()[:2000].decode("utf-8", "replace") try: d = json.loads(raw).get("detail") msg = d.get("message") if isinstance(d, dict) else str(d) except Exception: # noqa: BLE001 msg = raw[:200] raise ValueError("ElevenLabs refused it (%d): %s" % (e.code, msg)) except urllib.error.URLError as e: raise ValueError("could not reach ElevenLabs: %s" % e.reason) # ---------------------------------------------------------------- http class H(BaseHTTPRequestHandler): server_version = "oracle7-workbench" def log_message(self, *a): pass def _send(self, code, obj, ctype="application/json", raw=None): b = raw if raw is not None else json.dumps(obj, default=str).encode() self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(b))) self.send_header("X-Content-Type-Options", "nosniff") self.end_headers() self.wfile.write(b) def _body(self): n = int(self.headers.get("Content-Length") or 0) return json.loads(self.rfile.read(min(n, 2_000_000)) or b"{}") def _local_only(self): host = (self.headers.get("Host") or "").split(":")[0] return host in ("127.0.0.1", "localhost", "[::1]") def do_GET(self): # noqa: N802 if not self._local_only(): return self._send(403, {"error": "the workbench answers on 127.0.0.1 only"}) u = urlparse(self.path) q = {k: v[0] for k, v in parse_qs(u.query).items()} if u.path in ("/", "/index.html"): return self._send(200, None, TYPES[".html"], (STATIC / "index.html").read_bytes()) if u.path.startswith("/static/"): f = (STATIC / u.path[len("/static/"):]).resolve() if STATIC.resolve() in f.parents and f.is_file(): return self._send(200, None, TYPES.get(f.suffix, "application/octet-stream"), f.read_bytes()) return self._send(404, {"error": "not found"}) if u.path == "/api/state": return self._send(200, {"head": WB.eng.tip, "journal": settings().get("journal", False), "yours": sorted(WB.mine), "elevenlabs_remembered": bool(settings().get("elevenlabs_key"))}) if u.path == "/api/sensors": out = [] for sid, c in sorted(sensors.configs().items()): h = sensors.history(sid) latest = {} for r in h: if r["ok"]: latest[r["q"]] = {"v": r["v"], "unit": r["unit"], "t": r["t"]} out.append({"id": sid, "name": c["name"], "kind": c["kind"], "address": c.get("address", ""), "interval": c.get("interval", 30), "status": sensors.STATUS.get(sid), "latest": latest, "rejected": sum(1 for r in h if not r["ok"]), "state": exam.status(SPECS / sid)["state"] if (SPECS / sid).exists() else "none"}) return self._send(200, {"items": out, "quantities": {k: v[0] for k, v in sensors.QUANTITIES.items()}}) if u.path == "/api/sensor/series": rows = [r for r in sensors.history(q.get("id", "")) if r["q"] == q.get("q") and r["ok"]] step = max(1, len(rows) // 300) return self._send(200, {"points": [[int(r["t"]), r["v"]] for r in rows[::step]]}) if u.path == "/api/registry/memo": sid, kind = q.get("id", ""), q.get("kind", "register") if not ID.match(sid) or not (SPECS / sid).is_dir(): return self._send(404, {"error": "no such specialist"}) try: return self._send(200, {"memo": registry.memo_for(SPECS / sid, kind), "mint": registry.ORACLE7_MINT, "rpc": registry.RPC, "burn_raw": registry.BURN_RAW, "decimals": registry.DECIMALS}) except ValueError as e: return self._send(400, {"error": str(e)}) if u.path == "/api/convs": return self._send(200, {"keep": settings().get("keep_conversations"), "items": conv_list()}) if u.path == "/api/conv": return self._send(200, conv_load(q.get("id", ""))) if u.path == "/api/help": return self._send(200, {"topics": [{"topic": k, "text": v} for k, _, v in HELP_TOPICS], "commands": COMMANDS}) if u.path == "/api/specialists": return self._send(200, specialists_payload()) if u.path == "/api/systems": return self._send(200, json.loads((DATA / "systems.json").read_text(encoding="utf-8"))) if u.path == "/api/gaps": return self._send(200, gaps()) if u.path == "/api/card": sid = q.get("id", "") f = DATA / "cards" / (sid + ".md") if re.match(r"^[a-z0-9-]+$", sid) and f.exists(): return self._send(200, {"id": sid, "md": f.read_text(encoding="utf-8")}) return self._send(404, {"error": "no card"}) if u.path == "/api/files": d = SPECS / q.get("id", "") if not ID.match(q.get("id", "")) or not d.is_dir(): return self._send(404, {"error": "no such specialist"}) return self._send(200, {"files": sorted(p.name for p in d.iterdir() if p.is_file()), "status": exam.status(d)}) if u.path == "/api/file": d, name = SPECS / q.get("id", ""), q.get("name", "") if not ID.match(q.get("id", "")) or not FNAME.match(name) or not (d / name).is_file(): return self._send(404, {"error": "no such file"}) return self._send(200, {"name": name, "text": (d / name).read_text(encoding="utf-8")}) if u.path == "/api/exam": return self._send(200, exam.progress(q.get("id", ""))) return self._send(404, {"error": "not found"}) def do_POST(self): # noqa: N802 if not self._local_only() or (self.headers.get("Origin") and not re.match(r"^https?://(127\.0\.0\.1|localhost)(:\d+)?$", self.headers.get("Origin", ""))): return self._send(403, {"error": "the workbench answers its own page on 127.0.0.1 only"}) u = urlparse(self.path) try: b = self._body() except Exception: # noqa: BLE001 return self._send(400, {"error": "bad request"}) try: if u.path == "/api/chat": text, sid = str(b.get("text", "")).strip()[:4000], re.sub(r"[^a-z0-9]", "", str(b.get("sid", "x")))[:40] or "x" if not text: return self._send(400, {"error": "ask something"}) t = time.time() r = WB.chat(text, sid) status = r.get("status") kind, why = (None, None) if status == "answered" else why_declined(r.get("path")) if settings().get("keep_conversations") and not b.get("nokeep") and CID.match(sid): conv_add(sid, {"t": int(time.time()), "q": text, "status": status, "response": str(r.get("response", "")), "path": str(r.get("path", "")), "trace": trace(r.get("path"), r.get("response"), status), "why": why, "ms": int((time.time() - t) * 1000)}) if settings().get("journal"): with open(JOURNAL, "a", encoding="utf-8") as fh: fh.write(json.dumps({"t": int(time.time()), "q": text, "status": status, "why": kind, "path": str(r.get("path"))[:120]}) + "\n") return self._send(200, {"status": status, "response": str(r.get("response", "")), "path": str(r.get("path", "")), "trace": trace(r.get("path"), r.get("response"), status), "why": why, "why_kind": kind, "ms": int((time.time() - t) * 1000)}) if u.path == "/api/essay": # ProseFold v2: an essay composed from this model's own cited records, with the chosen length, reading level, # technical depth and style (profiles measured from the release's corpora); no language model import prosefold.options as PO topic = re.sub(r"\s+", " ", str(b.get("topic", ""))).strip()[:80] if not topic: return self._send(400, {"error": "name a topic"}) pick = lambda k, allowed, d: b.get(k) if b.get(k) in allowed else d # noqa: E731 voice_path = None vt = str(b.get("voice_text", "") or "") if b.get("style") == "custom" and len(vt.split()) >= 150: LOCAL.mkdir(parents=True, exist_ok=True) voice_path = LOCAL / "essay_voice.txt" voice_path.write_text(vt[:200000], encoding="utf-8") elif b.get("style") == "custom": return self._send(400, {"error": "paste at least 150 words of the writing whose voice you want"}) t = time.time() r = PO.write(WB.eng.chat, topic, length=pick("length", PO.LENGTHS, "medium"), level=pick("level", PO.LEVELS, "general"), depth=pick("depth", PO.DEPTHS, "standard"), style=pick("style", PO.STYLES, "plain"), voice_path=voice_path) r["ms"] = int((time.time() - t) * 1000) return self._send(200, r) if u.path == "/api/journal": s = settings() s["journal"] = bool(b.get("on")) save_settings(s) return self._send(200, {"on": s["journal"]}) if u.path == "/api/journal/clear": JOURNAL.unlink(missing_ok=True) return self._send(200, {"cleared": True}) if u.path == "/api/new": sid = str(b.get("id", "")).strip().lower() if not ID.match(sid): return self._send(400, {"error": "an id of lower-case letters, digits and hyphens, 2-41 long"}) words = [w for w in re.findall(r"[a-z][a-z0-9-]+", str(b.get("words", "")).lower())][:20] scaffold(sid, str(b.get("name") or sid)[:80], words, [str(x)[:300] for x in (b.get("examples") or [])][:20]) return self._send(200, {"id": sid}) if u.path == "/api/scan": rd = sources.scan_local() return self._send(200, {"readings": rd}) if u.path == "/api/scan/save": d = exam.make_source(SPECS, "this-computer", "this computer", "MachineOrgan", {"target": "local", "name": "this computer"}, ["computer", "pc", "laptop", "machine", "desktop"]) exam.start_source(d, machine_probes(sources.scan_local(), "my computer"), on_done=WB.reload) return self._send(200, {"id": d.name}) if u.path in ("/api/server", "/api/server/save"): name = re.sub(r"[^A-Za-z0-9 .-]", "", str(b.get("name") or b.get("host") or ""))[:40].strip() or "server" host, user, port, key = str(b.get("host", "")).strip(), str(b.get("user", "")).strip(), int(b.get("port") or 22), str(b.get("key", "")).strip() rd = sources.scan_server(host, user, port, key) if u.path == "/api/server": return self._send(200, {"readings": rd}) d = exam.make_source(SPECS, "server-" + slug(name), name, "MachineOrgan", {"target": "server", "name": name, "host": host, "user": user, "port": port, "key": key}, [w for w in slug(name).split("-") if w]) exam.start_source(d, machine_probes(rd, name), on_done=WB.reload) return self._send(200, {"id": d.name}) if u.path == "/api/files/add": name = re.sub(r"[^A-Za-z0-9 .-]", "", str(b.get("name") or "my files"))[:40].strip() or "my files" sid = str(b.get("id") or "files-" + slug(name)) if not ID.match(sid): return self._send(400, {"error": "bad id"}) d = SPECS / sid if not d.exists(): exam.make_source(SPECS, sid, name, "FilesOrgan", {"folder": sid + "/files", "name": name}, ["notes", "files", "documents"]) saved = [] for f in (b.get("files") or [])[:50]: fn = re.sub(r"[^A-Za-z0-9_.-]", "_", Path(str(f.get("name", ""))).name)[:80] if not re.search(r"\.(csv|tsv|txt|md)$", fn, re.I): continue text = str(f.get("text", "")) if len(text) > 5_000_000: continue (d / "files" / fn).write_text(text, encoding="utf-8") saved.append(fn) if not saved: return self._send(400, {"error": "add .csv, .tsv, .txt or .md files (save other documents as text or CSV first)"}) exam.start_source(d, file_probes(d), on_done=WB.reload) return self._send(200, {"id": sid, "saved": saved}) if u.path == "/api/el/connect": key = el_key(b) voices = json.loads(el_call("/voices", key)).get("voices", []) st = settings() if b.get("remember"): st["elevenlabs_key"] = key else: st.pop("elevenlabs_key", None) save_settings(st) return self._send(200, {"voices": [{"id": v["voice_id"], "name": v.get("name", "")} for v in voices][:80], "remembered": bool(b.get("remember"))}) if u.path == "/api/el/forget": st = settings() st.pop("elevenlabs_key", None) save_settings(st) return self._send(200, {"forgotten": True}) if u.path == "/api/el/speech": key, text = el_key(b), str(b.get("text", ""))[:2500] vid = re.sub(r"[^A-Za-z0-9]", "", str(b.get("voice") or "JBFqnCBsd6RMkjVDRZzb")) audio = el_call("/text-to-speech/%s?output_format=mp3_44100_128" % vid, key, {"text": text, "model_id": "eleven_multilingual_v2"}, accept="audio/mpeg") return self._send(200, None, "audio/mpeg", audio) if u.path == "/api/el/image": # ElevenLabs Image & Video API: create, poll until completed, download the signed content_url key, prompt = el_key(b), str(b.get("prompt", ""))[:1500] model = str(b.get("model") or "gemini-2.5-flash-image") if model not in ("gemini-2.5-flash-image", "gpt-image-1"): return self._send(400, {"error": "model must be gemini-2.5-flash-image or gpt-image-1"}) body = {"model_id": model, "prompt": prompt, "aspect_ratio": "16:9" if model.startswith("gemini") else "3:2"} gid = json.loads(el_call("/flows/image", key, body))["id"] t0, g = time.time(), {} while time.time() - t0 < 180: g = json.loads(el_call("/flows/image/%s" % re.sub(r"[^A-Za-z0-9_-]", "", gid), key)) if g.get("status") in ("completed", "failed"): break time.sleep(2) if g.get("status") != "completed": return self._send(502, {"error": "image not made: %s" % (g.get("failure_reason") or g.get("status") or "timed out")}) import urllib.request try: data = urllib.request.urlopen(g["content_url"], timeout=60).read() except Exception: # noqa: BLE001 - the signed URL normally needs no key; try once with it data = urllib.request.urlopen(urllib.request.Request(g["content_url"], headers={"xi-api-key": key}), timeout=60).read() assets = LOCAL / "assets" assets.mkdir(exist_ok=True) ext = ".png" if "png" in str(g.get("content_mime_type", "")) else ".jpg" if "jp" in str(g.get("content_mime_type", "")) else ".webp" (assets / ("image-%s%s" % (time.strftime("%Y%m%d-%H%M%S"), ext))).write_bytes(data) return self._send(200, None, g.get("content_mime_type") or "image/png", data) if u.path == "/api/el/sound": key, prompt = el_key(b), str(b.get("prompt", ""))[:400] audio = el_call("/sound-generation", key, {"text": prompt, "duration_seconds": min(22.0, max(2.0, float(b.get("seconds") or 20))), "prompt_influence": 0.4}, accept="audio/mpeg") return self._send(200, None, "audio/mpeg", audio) if u.path == "/api/mcp/connect": tools = mcp.connect(str(b.get("name", "")).strip(), str(b.get("command", "")).strip(), str(b.get("url", "")).strip(), str(b.get("token", "")).strip()) return self._send(200, {"tools": tools}) if u.path == "/api/mcp/save": server, tools = str(b.get("name", "")).strip(), [str(t) for t in (b.get("tools") or [])][:40] if server not in mcp.servers() or not tools: return self._send(400, {"error": "connect the server and pick at least one tool"}) sid = "mcp-" + slug(server) ex = [(str(x.get("q", ""))[:300], (str(x.get("want")) if x.get("want") else None)) for x in (b.get("examples") or [])][:20] probes = [p for p in ex if p[0]] + [("What is the capital of France?", None), ("Who wrote Hamlet?", None)] d = exam.make_source(SPECS, sid, "%s tools" % server, "mcp.McpOrgan", {"server": server, "tools": tools, "name": server, "folder": sid}, [w for w in mcp.words(server)] or ["mcp"]) exam.start_source(d, probes, on_done=WB.reload) return self._send(200, {"id": sid}) if u.path == "/api/feedback": m = re.match(r"^yours:([a-z0-9-]+)>", str(b.get("path", ""))) right = bool(b.get("right")) if not m or not (SPECS / m.group(1)).is_dir(): return self._send(200, {"learned": False, "note": "marks teach your own specialists; this answer came from the core"}) d = SPECS / m.group(1) tool = re.search(r"tool ([\w.-]+)\(", str(b.get("response", ""))) if (exam.read_manifest(d) or {}).get("organ", {}).get("module", "").startswith("organ_mcp") or tool: e = mcp.learn(d / "credit.json", tool.group(1), str(b.get("q", "")), right) if tool else {} exam.start_source(d, None, on_done=WB.reload) # the lane check runs again after learning return self._send(200, {"learned": True, "tool": tool.group(1) if tool else None, "credit": e}) return self._send(200, {"learned": False, "note": "noted; only tool specialists learn from marks"}) if u.path == "/api/conv/rename": c = conv_load(str(b.get("id", ""))) c["title"] = str(b.get("title", ""))[:120] conv_path(c["id"]).write_text(json.dumps(c, indent=0), encoding="utf-8") return self._send(200, {"ok": True}) if u.path == "/api/conv/delete": conv_path(str(b.get("id", ""))).unlink(missing_ok=True) return self._send(200, {"ok": True}) if u.path == "/api/conv/delete_all": for p in CONVS.glob("*.json"): p.unlink() return self._send(200, {"ok": True}) if u.path == "/api/conv/keep": st = settings() st["keep_conversations"] = bool(b.get("on")) save_settings(st) return self._send(200, {"keep": st["keep_conversations"]}) if u.path == "/api/conv/specialist": cid, days = str(b.get("id", "")), max(1, min(365, int(b.get("days") or 7))) c = conv_load(cid) done = [t for t in c["turns"] if t.get("status") == "answered" and not str(t.get("path", "")).startswith("workbench:")] if not done: return self._send(400, {"error": "this conversation has no answers to remember yet"}) sid = ("chat-" + slug(c.get("title") or cid))[:34].rstrip("-") + "-" + cid[:4] d = exam.make_source(SPECS, sid, "conversation: " + (c.get("title") or cid)[:60], "ConversationOrgan", {"conv": cid, "name": (c.get("title") or cid)[:60]}, ["conversation", "earlier"]) m = exam.read_manifest(d) m["expires"] = time.time() + days * 86400 m["temporary"] = True (d / "manifest.json").write_text(json.dumps(m, indent=1), encoding="utf-8") probes = [] for t in done[:8]: ws = [w for w in sources._words(t["q"]) if w not in sources.RECALL_FILLER][:3] if ws: probes.append(("What did we say earlier about %s?" % " ".join(ws), t["q"][:40])) probes += [("What is the capital of France?", None), ("Who wrote Hamlet?", None), ("Who directed Jaws?", None)] exam.start_source(d, probes, on_done=WB.reload) return self._send(200, {"id": sid, "expires": m["expires"]}) if u.path == "/api/verify": return self._send(200, verify.verify_install()) if u.path == "/api/verify/chain": return self._send(200, verify.check_anchor(str(b.get("signature") or "") or None)) if u.path == "/api/sensor/test": cfg = {"kind": str(b.get("kind", "")), "address": str(b.get("address", "")).strip(), "user": str(b.get("user", "")), "password": str(b.get("password", ""))} if cfg["kind"] == "http" and not re.match(r"^https?://", cfg["address"]): return self._send(400, {"error": "the address must start with http:// or https://"}) vals, where = sensors.read_source(cfg) return self._send(200, {"source": where, "fields": [{"path": k, "value": v, "guess": sensors.guess_quantity(k)} for k, v in list(vals.items())[:60]]}) if u.path == "/api/sensor/add": name = re.sub(r"[^A-Za-z0-9 .-]", "", str(b.get("name") or "my sensor"))[:40].strip() or "my sensor" fields = [{"path": str(f["path"])[:120], "quantity": str(f["quantity"])[:40], "unit": str(f.get("unit") or "")[:12]} for f in (b.get("fields") or []) if f.get("path") and f.get("quantity")][:30] if not fields: return self._send(400, {"error": "pick at least one reading and say what it measures"}) sid = "sensor-" + slug(name) c = {"id": sid, "name": name, "kind": str(b.get("kind", "")), "address": str(b.get("address", "")).strip(), "interval": max(5, min(3600, int(b.get("interval") or 30))), "fields": fields, "user": str(b.get("user", "")), "password": str(b.get("password", "")), "added": int(time.time())} sensors.save_config(c) sensors.poll_once(c) sensors.start_poller(sid) d = exam.make_source(SPECS, sid, name, "sensors.SensorOrgan", {"sensor": sid, "name": name}, [w for w in slug(name).split("-") if w] or ["sensor"]) qs = sorted({f["quantity"] for f in fields}) probes = [("What is the %s on %s?" % (qq, name), qq) for qq in qs] probes += [("What is the temperature of the Sun?", None), ("How much does a litre of water weigh?", None)] exam.start_source(d, probes, on_done=WB.reload) return self._send(200, {"id": sid}) if u.path == "/api/sensor/remove": sid = str(b.get("id", "")) if not ID.match(sid): return self._send(400, {"error": "bad id"}) for p in (sensors.DIR / ("%s.json" % sid), sensors.DIR / ("%s.jsonl" % sid)): p.unlink(missing_ok=True) shutil.rmtree(SPECS / sid, ignore_errors=True) WB.reload() return self._send(200, {"removed": sid}) if u.path == "/api/solana/rpc": raw = json.dumps(b).encode() out = registry.relay(raw) return self._send(200, None, "application/json", out) if u.path == "/api/registry/record": sid = str(b.get("id", "")) if not ID.match(sid) or not (SPECS / sid).is_dir(): return self._send(400, {"error": "no such specialist"}) return self._send(200, {"records": registry.record(SPECS / sid, str(b.get("kind", "")), str(b.get("signature", "")).strip())}) if u.path == "/api/registry/check": sid = str(b.get("id", "")) if not ID.match(sid) or not (SPECS / sid).is_dir(): return self._send(400, {"error": "no such specialist"}) return self._send(200, registry.check(SPECS / sid)) if u.path == "/api/example": src = HERE / "examples" / "paint-coverage" if (SPECS / src.name).exists(): return self._send(400, {"error": "the example is already in your specialists"}) shutil.copytree(src, SPECS / src.name, ignore=shutil.ignore_patterns("__pycache__", "verdict_*", "exam_log*")) return self._send(200, {"id": src.name}) if u.path == "/api/file": d, name = SPECS / str(b.get("id", "")), str(b.get("name", "")) if not ID.match(str(b.get("id", ""))) or not d.is_dir() or not FNAME.match(name) or name.startswith("verdict_"): return self._send(400, {"error": "that file cannot be written here"}) (d / name).write_text(str(b.get("text", "")), encoding="utf-8") WB.reload() return self._send(200, {"saved": name, "status": exam.status(d)}) if u.path == "/api/exam": sid, split = str(b.get("id", "")), str(b.get("split", "dev")) if not ID.match(sid) or not (SPECS / sid).is_dir() or split not in ("dev", "heldout"): return self._send(400, {"error": "which specialist, and dev or heldout?"}) exam.start(SPECS / sid, split, WB.eng, on_done=WB.reload) return self._send(200, exam.progress(sid)) if u.path == "/api/package": sid = str(b.get("id", "")) if not ID.match(sid) or not (SPECS / sid).is_dir(): return self._send(400, {"error": "no such specialist"}) return self._send(200, package(sid)) if u.path == "/api/import": r = import_zip(str(b.get("path", ""))) return self._send(200, r) if u.path == "/api/code/run": return self._send(200, run_code(str(b.get("code", ""))[:100_000], str(b.get("tests", ""))[:50_000])) except ValueError as e: return self._send(400, {"error": str(e)}) except Exception as e: # noqa: BLE001 return self._send(500, {"error": "%s: %s" % (type(e).__name__, e)}) return self._send(404, {"error": "not found"}) def serve(engine, port=8177): global WB WB = Workbench(engine) sensors.start_all() srv = ThreadingHTTPServer(("127.0.0.1", port), H) print("Oracle7 Workbench (%s) on http://127.0.0.1:%d/" % (engine.tip, port), flush=True) srv.serve_forever()