Release 11.0 (full): passage readings held back when the sentence leaves out a question word, the judge names the step it refused, the part format v1, anyone's sealed test, the sleep cycle, and the six orbiters as files in the pack (part 7)

Head oracle_release_runtime_w575. Root 5985f38d66bcdedbdedb74db90432675a32ddce9dfe2d196b3282c40581d889b. Solana anchor: pending Ben's signature. Internal week (seven simulated nights): 0 parts admitted, 0 new wrong, published as measured. Full changelog: CHANGELOG_11.md.

workbench/swarm/close_learn102.py

@@ -15,6 +15,8 @@ import collections
 import copy
 import json
 import re
+
+import steps_budget as SB
 from pathlib import Path
 
 import reader_v4 as R4
@@ -217,7 +219,7 @@ class Closer:
         import time
         rd = self.cov.rd
         seen = seen if seen is not None else set()
-        if time.time() > deadline or len(out) >= 8 or depth > 4:
+        if SB.spent() > deadline or len(out) >= 8 or depth > 4:
             return
         left = set(need) - set(used)
         last = prog[-1]["function"] if prog else ""

workbench/swarm/coverage102.py

@@ -41,6 +41,8 @@ import hashlib
 import json
 import os
 import re
+
+import steps_budget as SB
 import sys
 import time
 from pathlib import Path
@@ -252,7 +254,7 @@ class Coverage:
         except Exception:  # noqa: BLE001
             self.closer = None
         try:                                             # the learned word-to-step alignment (align_learn102.py)
-            A = json.load(open(Path(__file__).resolve().parent / "data" / "alignment102.json"))["p_feature_given_word"]
+            A = json.load(open(os.environ.get("O7_ALIGN") or Path(__file__).resolve().parent / "data" / "alignment102.json"))["p_feature_given_word"]
         except Exception:  # noqa: BLE001
             A = {}
         self.align_t = A
@@ -911,8 +913,6 @@ class Coverage:
             if lit is not None or (f.startswith("FN:") and f[3:] not in SAID_FN):
                 continue
             best = max([self.align_t.get(f, {}).get(w, 0.0) for w in words] + [0.0])
-            if f.startswith("IN:") and f.count(":") == 1 and self.says_word(f, f[3:], next(iter(words), "")):
-                best = max(best, 1.0)
             tot += best
             n += 1
         for st in prog:
@@ -962,7 +962,7 @@ class Coverage:
                 written = b_ == a_ + 1 and re.fullmatch(r"-?\d[\d,]*(?:\.\d+)?|\d{4}-\d{2}-\d{2}|\d{4} \d{2} \d{2}", q[s1:e1]) is not None
                 if not t or (t not in vix and not written) or (s1, e1) in seen or any(x <= s1 and e1 <= y for x, y in seen):
                     continue
-                stored = next(iter(self._vstore.get(t, set()))) if t in vix else None
+                stored = min(self._vstore.get(t, {None}), key=str) if t in vix else None
                 if stored is None:                       # a written number or date the store does not hold: as written
                     m = re.search(re.escape(q[s1:e1]).replace("\ ", r"\W+"), question, re.I)
                     stored = question[m.start():m.end()] if m else q[s1:e1]
@@ -1218,10 +1218,10 @@ class Coverage:
         singles = [x for x in P if len(x[3]) == 1 and x[2][0]["function"] == "Find"]
         res = []
 
-        deadline = time.time() + float(os.environ.get("O7_FLASH_BUDGET", 3.0))
+        deadline = SB.deadline()
 
         def run(pp):
-            if len(res) >= 8 or time.time() > deadline:
+            if len(res) >= 8 or SB.spent() > deadline:
                 return
             for st in pp:
                 if st["function"] == "Find":
@@ -1744,7 +1744,7 @@ class Coverage:
                     m = re.search(re.escape(q[s1:e1]).replace("\\ ", r"\W+"), question, re.I)
                     vals_written.append((s1, e1, question[m.start():m.end()] if m else q[s1:e1]))
         # the flash: from every part, along every lit relation, up to two steps - all at once
-        deadline = time.time() + float(os.environ.get("O7_FLASH_BUDGET", 3.0))
+        deadline = SB.deadline()
         grown = []
         for s0, e0, prog, idset in P:
             own = self.part_words(ws, span(s0, e0), prog)
@@ -1768,7 +1768,7 @@ class Coverage:
             grown.append((s0, e0, prog, own, need, [b_ for b_ in beams if b_[0]], start))
         for s0, e0, prog, own, need, beams, _start in grown:
             for path, used, v in beams:
-                if time.time() > deadline:
+                if SB.spent() > deadline:
                     break
                 nodes = {ids_of[n] for n in np.flatnonzero(v)[:200]}
                 for pp, cw in closing(program(prog, path), nodes, used, need):
@@ -1782,7 +1782,7 @@ class Coverage:
                 need = (A[4] - B[3]) | (B[4] - A[3])
                 for pa_, ua, va in [((), frozenset(), A[6])] + A[5]:     # a part's own items meet the other's light
                     for pb_, ub, vb in [((), frozenset(), B[6])] + B[5]:
-                        if time.time() > deadline:
+                        if SB.spent() > deadline:
                             break
                         meet = va & vb
                         if not meet.any() or meet.sum() > 50:

workbench/swarm/data/flash_threshold.json

@@ -1,254 +1,12 @@
 {
  "format": "oracle7-flash-threshold/1",
- "threshold": 0.79,
+ "threshold": 0.733,
  "at_threshold": {
-  "threshold": 0.79,
-  "right": 44,
+  "threshold": 0.733,
+  "right": 38,
   "wrong": 2,
-  "precision": 0.957
+  "precision": 0.95
  },
- "chosen_by": "lowest score at which the flash's development additions reach precision >= 0.95 (>= 10 right); else the highest-precision point",
- "n_scored": 440,
- "curve": [
-  {
-   "threshold": 0.997,
-   "right": 1,
-   "wrong": 0,
-   "precision": 1.0
-  },
-  {
-   "threshold": 0.968,
-   "right": 12,
-   "wrong": 0,
-   "precision": 1.0
-  },
-  {
-   "threshold": 0.901,
-   "right": 22,
-   "wrong": 1,
-   "precision": 0.957
-  },
-  {
-   "threshold": 0.818,
-   "right": 32,
-   "wrong": 2,
-   "precision": 0.941
-  },
-  {
-   "threshold": 0.798,
-   "right": 43,
-   "wrong": 2,
-   "precision": 0.956
-  },
-  {
-   "threshold": 0.753,
-   "right": 52,
-   "wrong": 4,
-   "precision": 0.929
-  },
-  {
-   "threshold": 0.742,
-   "right": 63,
-   "wrong": 4,
-   "precision": 0.94
-  },
-  {
-   "threshold": 0.719,
-   "right": 73,
-   "wrong": 5,
-   "precision": 0.936
-  },
-  {
-   "threshold": 0.684,
-   "right": 81,
-   "wrong": 8,
-   "precision": 0.91
-  },
-  {
-   "threshold": 0.661,
-   "right": 91,
-   "wrong": 9,
-   "precision": 0.91
-  },
-  {
-   "threshold": 0.64,
-   "right": 96,
-   "wrong": 15,
-   "precision": 0.865
-  },
-  {
-   "threshold": 0.629,
-   "right": 104,
-   "wrong": 18,
-   "precision": 0.852
-  },
-  {
-   "threshold": 0.612,
-   "right": 114,
-   "wrong": 19,
-   "precision": 0.857
-  },
-  {
-   "threshold": 0.6,
-   "right": 122,
-   "wrong": 22,
-   "precision": 0.847
-  },
-  {
-   "threshold": 0.588,
-   "right": 132,
-   "wrong": 23,
-   "precision": 0.852
-  },
-  {
-   "threshold": 0.575,
-   "right": 140,
-   "wrong": 26,
-   "precision": 0.843
-  },
-  {
-   "threshold": 0.565,
-   "right": 150,
-   "wrong": 27,
-   "precision": 0.847
-  },
-  {
-   "threshold": 0.561,
-   "right": 159,
-   "wrong": 29,
-   "precision": 0.846
-  },
-  {
-   "threshold": 0.548,
-   "right": 168,
-   "wrong": 31,
-   "precision": 0.844
-  },
-  {
-   "threshold": 0.532,
-   "right": 176,
-   "wrong": 34,
-   "precision": 0.838
-  },
-  {
-   "threshold": 0.522,
-   "right": 186,
-   "wrong": 35,
-   "precision": 0.842
-  },
-  {
-   "threshold": 0.512,
-   "right": 194,
-   "wrong": 38,
-   "precision": 0.836
-  },
-  {
-   "threshold": 0.501,
-   "right": 203,
-   "wrong": 40,
-   "precision": 0.835
-  },
-  {
-   "threshold": 0.494,
-   "right": 211,
-   "wrong": 43,
-   "precision": 0.831
-  },
-  {
-   "threshold": 0.488,
-   "right": 221,
-   "wrong": 44,
-   "precision": 0.834
-  },
-  {
-   "threshold": 0.484,
-   "right": 231,
-   "wrong": 45,
-   "precision": 0.837
-  },
-  {
-   "threshold": 0.478,
-   "right": 240,
-   "wrong": 47,
-   "precision": 0.836
-  },
-  {
-   "threshold": 0.463,
-   "right": 245,
-   "wrong": 53,
-   "precision": 0.822
-  },
-  {
-   "threshold": 0.458,
-   "right": 253,
-   "wrong": 56,
-   "precision": 0.819
-  },
-  {
-   "threshold": 0.45,
-   "right": 262,
-   "wrong": 58,
-   "precision": 0.819
-  },
-  {
-   "threshold": 0.442,
-   "right": 270,
-   "wrong": 61,
-   "precision": 0.816
-  },
-  {
-   "threshold": 0.433,
-   "right": 280,
-   "wrong": 62,
-   "precision": 0.819
-  },
-  {
-   "threshold": 0.418,
-   "right": 289,
-   "wrong": 64,
-   "precision": 0.819
-  },
-  {
-   "threshold": 0.401,
-   "right": 299,
-   "wrong": 65,
-   "precision": 0.821
-  },
-  {
-   "threshold": 0.38,
-   "right": 308,
-   "wrong": 67,
-   "precision": 0.821
-  },
-  {
-   "threshold": 0.361,
-   "right": 317,
-   "wrong": 69,
-   "precision": 0.821
-  },
-  {
-   "threshold": 0.335,
-   "right": 327,
-   "wrong": 70,
-   "precision": 0.824
-  },
-  {
-   "threshold": 0.313,
-   "right": 335,
-   "wrong": 73,
-   "precision": 0.821
-  },
-  {
-   "threshold": 0.3,
-   "right": 343,
-   "wrong": 76,
-   "precision": 0.819
-  },
-  {
-   "threshold": 0.258,
-   "right": 352,
-   "wrong": 78,
-   "precision": 0.819
-  }
- ]
+ "chosen_by": "10.5: relearned after program_score stopped testing one arbitrary word: lowest score at which development additions reach precision >= 0.95 (>= 10 right)",
+ "n_scored": 260
 }

workbench/swarm/examiner105.py

@@ -0,0 +1,97 @@
+"""examiner105 (10.5) - anyone's sealed test. Write a panel of your own questions, seal it, publish its hash, and run it once: the same discipline
+the release is held to, on your machine.
+  exam seal <panel.json>            check the panel and print its hash, to publish before anything runs
+  exam run <panel.json> [out.json]  run it through the head. A run is once: its hash is logged and refused the second time
+Panel: {"name": ..., "items": [{"q": question, "a": expected answer, "valid": true} or {"q": question, "valid": false}]}.
+At least 59 items (no failure then bounds the error below 5% at 95%), and at least 20% must-decline, so a system that always answers fails."""
+import hashlib
+import json
+import math
+import sys
+from pathlib import Path
+
+MIN_ITEMS, MIN_DECLINE = 59, 0.2
+
+
+def load(path):
+    p = json.load(open(path, encoding="utf-8"))
+    items = p.get("items") if isinstance(p, dict) else None
+    if not isinstance(items, list) or not items:
+        raise ValueError("a panel is {\"items\": [...]}")
+    for i, x in enumerate(items):
+        if not isinstance(x, dict) or not str(x.get("q", "")).strip() or "valid" not in x:
+            raise ValueError("item %d needs q and valid" % i)
+        if x["valid"] and x.get("a") in (None, "", []):
+            raise ValueError("item %d is valid but has no expected answer" % i)
+    seen = {x["q"] for x in items}
+    if len(seen) != len(items):
+        raise ValueError("a question appears twice")
+    must = sum(1 for x in items if not x["valid"])
+    if len(items) < MIN_ITEMS or must < MIN_DECLINE * len(items):
+        raise ValueError("needs at least %d items and %d%% must-decline; this has %d and %d" % (MIN_ITEMS, MIN_DECLINE * 100, len(items), must))
+    return p, items, hashlib.sha256(json.dumps(items, sort_keys=True).encode()).hexdigest()
+
+
+def bound(n, k=0):
+    """the 95% upper bound on the error rate, from k failures in n (exact, by bisection)"""
+    lo, hi = 0.0, 1.0
+    for _ in range(60):
+        m = (lo + hi) / 2
+        tail = sum(math.comb(n, i) * m ** i * (1 - m) ** (n - i) for i in range(k + 1))
+        lo, hi = (m, hi) if tail > 0.05 else (lo, m)
+    return hi
+
+
+def run(path, engine, log, out=None):
+    from handoff101 import answer_of, compare
+    _p, items, sha = load(path)
+    used = log.read_text(encoding="utf-8").split() if log.exists() else []
+    if sha in used:
+        raise ValueError("this panel already ran once: write new items for a new test")
+    rows, tally = [], {"right": 0, "wrong": 0, "declined": 0, "false_yes": 0, "declined_right": 0}
+    for i, x in enumerate(items):
+        r = engine.chat(x["q"], "examiner%d" % i)
+        a = answer_of(r) if r.get("status") == "answered" else None
+        if not x["valid"]:
+            k = "false_yes" if a is not None else "declined_right"
+        elif a is None:
+            k = "declined"
+        else:
+            want = x["a"] if isinstance(x["a"], list) else [x["a"]]
+            k = "right" if any(compare(a, str(w))[0] == "CONFIRMED" for w in want) else "wrong"
+        tally[k] += 1
+        rows.append({"q": x["q"], "got": a, "result": k, "path": str(r.get("path"))})
+    log.parent.mkdir(parents=True, exist_ok=True)
+    import declines
+    with open(log.parent / "holes_from_exams.jsonl", "a", encoding="utf-8") as f:   # every failed question, typed, for the night
+        for x, row, k in ((x, row, row["result"]) for x, row in zip(items, rows)):
+            if k in ("declined", "wrong", "false_yes"):
+                f.write(json.dumps({"q": x["q"], "expect": x.get("a"), "valid": x["valid"], "failed_as": k, "kind": declines.classify(str(row["got"]), row["path"]), "panel": sha}) + "\n")
+    with open(log, "a", encoding="utf-8") as f:
+        f.write(sha + "\n")
+    said = tally["right"] + tally["wrong"] + tally["false_yes"]
+    rep = {"panel": sha, "tally": tally, "items": len(items), "error_bound_95": round(bound(said, tally["wrong"] + tally["false_yes"]), 4) if said else None, "rows": rows}
+    if out:
+        json.dump(rep, open(out, "w", encoding="utf-8"), indent=1, ensure_ascii=False)
+    return rep
+
+
+def main(rest, engine_factory, state):
+    if not rest or rest[0] not in ("seal", "run") or len(rest) < 2:
+        print(__doc__)
+        return 2
+    try:
+        if rest[0] == "seal":
+            _p, items, sha = load(rest[1])
+            print("sealed: %d items\nsha256 %s\nPublish this hash before you run it." % (len(items), sha))
+            return 0
+        rep = run(rest[1], engine_factory(), Path(state) / "examiner_used.txt", rest[2] if len(rest) > 2 else None)
+        print(json.dumps({k: rep[k] for k in ("panel", "items", "tally", "error_bound_95")}, indent=1))
+        return 0
+    except ValueError as e:
+        print("refused: %s" % e)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:], None, "."))

workbench/swarm/own_steps.py

@@ -0,0 +1,51 @@
+#!/usr/bin/env python3
+"""own_steps (11.0, plan 5.3) - the reader's steps in Oracle7's own terms. A program is a chain of store operations; KQA's names for them are an export
+format for outside scoring, and nothing more. The map is one-to-one, so the change is proved by running every labelled program both ways: the same
+answer, every time. usage: own_steps.py    (runs the proof over every program in KQA validation)"""
+import json
+import sys
+
+OWN = {"FindAll": "every_thing", "Find": "named", "FilterConcept": "of_kind", "FilterStr": "with_text", "FilterNum": "with_number", "FilterYear": "with_year", "FilterDate": "with_date",
+       "QFilterStr": "where_text", "QFilterNum": "where_number", "QFilterYear": "where_year", "QFilterDate": "where_date", "Relate": "along", "And": "both", "Or": "either", "What": "name_of",
+       "Count": "how_many", "QueryAttr": "value_of", "QueryAttrUnderCondition": "value_of_where", "QueryAttrQualifier": "qualifier_of", "QueryRelation": "link_between",
+       "QueryRelationQualifier": "qualifier_of_link", "SelectBetween": "pick_between", "SelectAmong": "pick_among", "VerifyStr": "is_text", "VerifyNum": "is_number", "VerifyYear": "is_year", "VerifyDate": "is_date"}
+KQA = {v: k for k, v in OWN.items()}
+assert len(KQA) == len(OWN)
+
+
+def to_own(program):
+    return [dict(st, function=OWN[st["function"]]) for st in program]
+
+
+def to_export(program):
+    return [dict(st, function=KQA[st["function"]]) for st in program]
+
+
+def run_own(store, program):
+    """a program in Oracle7's terms, on the store"""
+    return store.run(to_export(program))
+
+
+if __name__ == "__main__":
+    sys.path[:0] = ["/opt/rel102/t/workbench/swarm"]
+    import steps_kb as K
+    kq = K.KQA()
+    tot = same = 0
+    bad = []
+    for name in ("val",):
+        for x in json.load(open("/opt/rel50/kqa/%s.json" % name)):
+            tot += 1
+
+            def go(f, p):
+                try:
+                    return f(p)
+                except Exception as e:  # noqa: BLE001 - a program that fails must fail the same way
+                    return "ERR %s" % type(e).__name__
+            own = to_own(x["program"])
+            assert to_export(own) == x["program"]
+            a, b = go(kq.run, x["program"]), go(lambda p: run_own(kq, p), own)
+            if a == b or sorted(map(str, a if isinstance(a, (list, set)) else [a])) == sorted(map(str, b if isinstance(b, (list, set)) else [b])):
+                same += 1
+            else:
+                bad.append(x["question"][:80])
+    print(json.dumps({"programs": tot, "same_answer": same, "different": bad[:5]}))

workbench/swarm/part11.py

@@ -0,0 +1,312 @@
+#!/usr/bin/env python3
+"""part11 (11.0) - THE PART FORMAT, version 1. Growth by declaration: a part is declared data that the format can check, and no kind here
+carries code. It is hashed, so a changed part is a new part, signed by its author (Ed25519, RFC 8032, standard library only), and carries
+its examiner. The format sets the least examiner each kind must have, and an author can only add to it, so nobody can write an easy exam.
+  kinds   table, formula, alias-set, reading-rule, sensor, example-set, composite
+  admit   every structural check, with the reason for a refusal. A part that passes still has to sit its exam
+  apply   runs a declared table or formula; a composite runs its parts in order, with units checked at each join
+usage: part11.py new <part.json> <key.hex> | check <part.json> | keygen"""
+import ast
+import hashlib
+import json
+import math
+import operator
+import sys
+
+FORMAT = "oracle7-part-v1"
+KINDS = ("table", "formula", "alias-set", "reading-rule", "sensor", "example-set", "composite")
+FIELDS = {"format", "kind", "version", "interface", "body", "examiner", "provenance", "reach", "id", "author", "signature"}
+LEAST = {"n": 149, "must_decline_share": 0.2}         # 5% error bound at 95% over 100 candidates, zero failures, 20% must-decline
+
+# ---- Ed25519, RFC 8032 section 6
+_p = 2 ** 255 - 19
+_q = 2 ** 252 + 27742317777372353535851937790883648493
+_d = (-121665 * pow(121666, _p - 2, _p)) % _p
+_I = pow(2, (_p - 1) // 4, _p)
+
+
+def _sha512(b):
+    return hashlib.sha512(b).digest()
+
+
+def _add(P, Q):
+    A, B = (P[1] - P[0]) * (Q[1] - Q[0]) % _p, (P[1] + P[0]) * (Q[1] + Q[0]) % _p
+    C, D = 2 * P[3] * Q[3] * _d % _p, 2 * P[2] * Q[2] % _p
+    E, F, G, H = B - A, D - C, D + C, B + A
+    return (E * F % _p, G * H % _p, F * G % _p, E * H % _p)
+
+
+def _mul(s, P):
+    Q = (0, 1, 1, 0)
+    while s:
+        if s & 1:
+            Q = _add(Q, P)
+        P = _add(P, P)
+        s >>= 1
+    return Q
+
+
+def _eq(P, Q):
+    return (P[0] * Q[2] - Q[0] * P[2]) % _p == 0 and (P[1] * Q[2] - Q[1] * P[2]) % _p == 0
+
+
+def _xrec(y, sign):
+    x2 = (y * y - 1) * pow(_d * y * y + 1, _p - 2, _p) % _p
+    if x2 == 0:
+        return None if sign else 0
+    x = pow(x2, (_p + 3) // 8, _p)
+    if (x * x - x2) % _p:
+        x = x * _I % _p
+    if (x * x - x2) % _p:
+        return None
+    return _p - x if (x & 1) != sign else x
+
+
+_gy = 4 * pow(5, _p - 2, _p) % _p
+_gx = _xrec(_gy, 0)
+_G = (_gx, _gy, 1, _gx * _gy % _p)
+
+
+def _comp(P):
+    zi = pow(P[2], _p - 2, _p)
+    x, y = P[0] * zi % _p, P[1] * zi % _p
+    return int.to_bytes(y | ((x & 1) << 255), 32, "little")
+
+
+def _dec(b):
+    y = int.from_bytes(b, "little")
+    sign = y >> 255
+    y &= (1 << 255) - 1
+    if y >= _p:
+        return None
+    x = _xrec(y, sign)
+    return None if x is None else (x, y, 1, x * y % _p)
+
+
+def _expand(sk):
+    h = _sha512(sk)
+    a = int.from_bytes(h[:32], "little")
+    a &= (1 << 254) - 8
+    a |= 1 << 254
+    return a, h[32:]
+
+
+def public_key(sk):
+    return _comp(_mul(_expand(sk)[0], _G))
+
+
+def sign(sk, msg):
+    a, pre = _expand(sk)
+    A = _comp(_mul(a, _G))
+    r = int.from_bytes(_sha512(pre + msg), "little") % _q
+    R = _comp(_mul(r, _G))
+    h = int.from_bytes(_sha512(R + A + msg), "little") % _q
+    return R + int.to_bytes((r + h * a) % _q, 32, "little")
+
+
+def verify(pk, msg, sig):
+    if len(pk) != 32 or len(sig) != 64:
+        return False
+    A, R = _dec(pk), _dec(sig[:32])
+    s = int.from_bytes(sig[32:], "little")
+    if A is None or R is None or s >= _q:
+        return False
+    h = int.from_bytes(_sha512(sig[:32] + pk + msg), "little") % _q
+    return _eq(_mul(s, _G), _add(R, _mul(h, A)))
+
+
+# ---- the part
+def canonical(part):
+    return json.dumps({k: part[k] for k in sorted(part) if k not in ("id", "signature")}, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()
+
+
+def seal(part, sk):
+    """an author's part, hashed and signed"""
+    p = dict(part, format=FORMAT, author=public_key(sk).hex())
+    p["id"] = hashlib.sha256(canonical(p)).hexdigest()
+    p["signature"] = sign(sk, bytes.fromhex(p["id"])).hex()
+    return p
+
+
+OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.Pow: operator.pow}
+FUNCS = {"sqrt": math.sqrt, "log": math.log, "exp": math.exp, "sin": math.sin, "cos": math.cos, "abs": abs}
+
+
+def _formula_ok(expr, inputs):
+    """a formula is arithmetic over its declared inputs: anything else in it is refused"""
+    try:
+        tree = ast.parse(expr, mode="eval")
+    except SyntaxError:
+        return False
+    for n in ast.walk(tree):
+        if isinstance(n, ast.Name):
+            if n.id not in inputs and n.id not in FUNCS:
+                return False
+        elif isinstance(n, ast.Call):
+            if not (isinstance(n.func, ast.Name) and n.func.id in FUNCS) or n.keywords:
+                return False
+        elif isinstance(n, ast.Constant):
+            if type(n.value) not in (int, float):
+                return False
+        elif not isinstance(n, (ast.Expression, ast.BinOp, ast.UnaryOp, ast.USub, ast.Load) + tuple(OPS)):
+            return False
+    return True
+
+
+def _eval(n, env):
+    if isinstance(n, ast.Expression):
+        return _eval(n.body, env)
+    if isinstance(n, ast.Constant):
+        return n.value
+    if isinstance(n, ast.Name):
+        return env[n.id]
+    if isinstance(n, ast.UnaryOp):
+        return -_eval(n.operand, env)
+    if isinstance(n, ast.BinOp):
+        return OPS[type(n.op)](_eval(n.left, env), _eval(n.right, env))
+    return FUNCS[n.func.id](*[_eval(a, env) for a in n.args])
+
+
+def _typed(xs):
+    return isinstance(xs, list) and xs and all(isinstance(x, dict) and isinstance(x.get("name"), str) and x["name"] and "unit" in x for x in xs)
+
+
+def _body_ok(kind, body, face, known):
+    ins = {x["name"] for x in face["inputs"]} if face.get("inputs") else set()
+    if kind == "table":
+        rows = body.get("rows")
+        return isinstance(rows, list) and rows and all(isinstance(r, list) and len(r) == len(face["inputs"]) + len(face["outputs"]) for r in rows)
+    if kind == "formula":
+        return isinstance(body.get("expr"), str) and _formula_ok(body["expr"], ins) and bool(body.get("source"))
+    if kind == "alias-set":
+        a = body.get("aliases")
+        return isinstance(a, dict) and a and all(isinstance(k, str) and isinstance(v, str) for k, v in a.items())
+    if kind == "reading-rule":
+        return isinstance(body.get("shape"), str) and bool(body["shape"].strip())
+    if kind == "sensor":
+        return isinstance(body.get("channel"), str) and isinstance(body.get("ideal"), list) and len(body["ideal"]) == 2 and body["ideal"][0] < body["ideal"][1] and isinstance(body.get("range"), list) and len(body["range"]) == 2 and body["range"][0] <= body["ideal"][0] and body["ideal"][1] <= body["range"][1]
+    if kind == "example-set":
+        ex = body.get("examples")
+        return isinstance(ex, list) and ex and all(isinstance(x, dict) and x.get("cite") for x in ex)
+    if kind == "composite":
+        path = body.get("path")
+        if not (isinstance(path, list) and len(path) >= 2 and all(p in known for p in path)):
+            return False
+        for a, b in zip(path, path[1:]):
+            need = {x["unit"] for x in known[b]["interface"]["inputs"]}
+            if not {x["unit"] for x in known[a]["interface"]["outputs"]} & need:
+                return False
+        return True
+    return False
+
+
+def admit(part, known=None, trusted=None):
+    """structure, hash, signature, least examiner -> (True, "admitted") or (False, why). known: {id: part} for composites"""
+    if not isinstance(part, dict):
+        return False, "not a part"
+    if set(part) - FIELDS or FIELDS - {"reach"} - set(part):
+        return False, "fields: %s" % sorted((set(part) - FIELDS) | (FIELDS - {"reach"} - set(part)))
+    if part["format"] != FORMAT or part["kind"] not in KINDS:
+        return False, "format or kind"
+    face = part["interface"]
+    if not isinstance(face, dict) or not _typed(face.get("outputs")) or (part["kind"] not in ("sensor", "alias-set", "example-set", "composite") and not _typed(face.get("inputs"))):
+        return False, "interface must be typed, with units"
+    ex = part["examiner"]
+    if not isinstance(ex, dict) or not ex.get("generator") or not isinstance(ex.get("n"), int) or not isinstance(ex.get("must_decline_share"), (int, float)):
+        return False, "no examiner"
+    if ex["n"] < LEAST["n"] or ex["must_decline_share"] < LEAST["must_decline_share"]:
+        return False, "examiner below the format's least: %d items, %d%% must-decline" % (LEAST["n"], LEAST["must_decline_share"] * 100)
+    if ex.get("seed_rule") not in ("block-hash", "fixed-then-audited"):
+        return False, "an examiner's seed must come from outside the author"
+    ran = ex.get("ran")                               # a declared exam isn't an exam: the part carries the record of the one that ran
+    if not isinstance(ran, dict) or not isinstance(ran.get("items"), int) or ran.get("failed") != 0 or ran["items"] < ex["n"]:
+        return False, "no record of a passed exam of %d items" % ex["n"]
+    prov = part["provenance"]
+    if not isinstance(prov, dict) or not prov.get("sources"):
+        return False, "provenance"
+    if not _body_ok(part["kind"], part["body"], face, known or {}):
+        return False, "body doesn't hold for kind %s" % part["kind"]
+    if part["id"] != hashlib.sha256(canonical(part)).hexdigest():
+        return False, "hash"
+    try:
+        good = verify(bytes.fromhex(part["author"]), bytes.fromhex(part["id"]), bytes.fromhex(part["signature"]))
+    except (ValueError, TypeError):
+        good = False
+    if not good:
+        return False, "signature"
+    if trusted is not None and part["author"] not in trusted:
+        return False, "author not trusted here"
+    return True, "admitted"
+
+
+def apply(part, inputs, known=None):
+    """run a declared part on named inputs -> value or None. No kind here runs code from the part"""
+    k = part["kind"]
+    names = [x["name"] for x in part["interface"].get("inputs") or []]
+    if k == "table":
+        key = [inputs.get(n) for n in names]
+        return next((r[len(names):] for r in part["body"]["rows"] if r[:len(names)] == key), None)
+    if k == "formula":
+        return _eval(ast.parse(part["body"]["expr"], mode="eval"), {**FUNCS, **inputs}) if all(n in inputs for n in names) else None
+    if k == "sensor":                                 # a reading passes only inside the sensor's stated range
+        lo, hi = part["body"]["range"]
+        x = next(iter(inputs.values()), None)
+        return x if isinstance(x, (int, float)) and lo <= x <= hi else None
+    if k == "composite":
+        v, cur = inputs, None
+        for pid in part["body"]["path"]:
+            p = known[pid]
+            cur = apply(p, v, known)
+            if cur is None:
+                return None
+            cur = cur[0] if isinstance(cur, list) else cur
+            n = [x["name"] for x in (known[part["body"]["path"][min(part["body"]["path"].index(pid) + 1, len(part["body"]["path"]) - 1)]]["interface"].get("inputs") or [{"name": "x"}])][0]
+            v = {n: cur}
+        return cur
+    return None
+
+
+def _suite():
+    """RFC 8032 vector, then parts that must be admitted and parts that must be refused"""
+    sk = bytes.fromhex("9d61b19deffd5a60ba844af492ec2cc44449c5697b326919703bac031cae7f60")
+    assert public_key(sk).hex() == "d75a980182b10ab7d54bfed3c964073a0ee172f3daa62325af021a68f707511a"
+    assert sign(sk, b"").hex().startswith("e5564300c360ac729086e2cc806e828a84877f1eb8e5d974d873e065224901555fb8821590a33bacc61e39701cf9b46b")
+    ex = {"generator": "store", "seed_rule": "block-hash", "n": 149, "must_decline_share": .2, "ran": {"items": 149, "failed": 0}}
+    base = {"version": 1, "examiner": ex, "provenance": {"sources": ["test"]}, "reach": None}
+    ohm = dict(base, kind="formula", interface={"inputs": [{"name": "v", "unit": "V"}, {"name": "r", "unit": "ohm"}], "outputs": [{"name": "i", "unit": "A"}]}, body={"expr": "v / r", "source": "Ohm's law"})
+    tbl = dict(base, kind="table", interface={"inputs": [{"name": "m", "unit": "symbol"}], "outputs": [{"name": "z", "unit": "count"}]}, body={"rows": [["H", 1], ["He", 2]]})
+    sq = dict(base, kind="formula", interface={"inputs": [{"name": "i", "unit": "A"}], "outputs": [{"name": "p", "unit": "W"}]}, body={"expr": "i ** 2 * 5", "source": "P = I^2 R at 5 ohm"})
+    good = {"formula": seal(ohm, sk), "table": seal(tbl, sk), "formula2": seal(sq, sk)}
+    known = {p["id"]: p for p in good.values()}
+    comp = seal(dict(base, kind="composite", interface={"inputs": ohm["interface"]["inputs"], "outputs": sq["interface"]["outputs"]}, body={"path": [good["formula"]["id"], good["formula2"]["id"]]}), sk)
+    known[comp["id"]] = comp
+    res = {"admitted": [admit(p, known) for p in list(good.values()) + [comp]]}
+    assert all(r[0] for r in res["admitted"]), res
+    assert apply(good["formula"], {"v": 10, "r": 2}) == 5 and apply(good["table"], {"m": "He"}) == [2] and apply(comp, {"v": 10, "r": 2}, known) == 125
+    p = good["formula"]
+    other = bytes(range(32))
+    bad = {"malformed": "nope", "unknown field": dict(p, evil=1), "unsigned": dict(p, signature="00" * 64), "body changed after sealing": dict(p, body={"expr": "v * r", "source": "x"}),
+           "foreign signature": dict(p, signature=sign(other, bytes.fromhex(p["id"])).hex()),
+           "code in a formula": seal(dict(ohm, body={"expr": "__import__('os').system('x')", "source": "x"}), sk), "attribute in a formula": seal(dict(ohm, body={"expr": "v.__class__", "source": "x"}), sk),
+           "undeclared input": seal(dict(ohm, body={"expr": "v / q", "source": "x"}), sk), "no units": seal(dict(ohm, interface={"inputs": [{"name": "v"}], "outputs": [{"name": "i", "unit": "A"}]}), sk),
+           "easy examiner": seal(dict(ohm, examiner=dict(ex, n=5)), sk), "no must-decline": seal(dict(ohm, examiner=dict(ex, must_decline_share=0)), sk), "exam never run": seal(dict(ohm, examiner={k: v for k, v in ex.items() if k != "ran"}), sk), "exam failed": seal(dict(ohm, examiner=dict(ex, ran={"items": 149, "failed": 1})), sk), "exam smaller than declared": seal(dict(ohm, examiner=dict(ex, ran={"items": 20, "failed": 0})), sk), "author's own seed": seal(dict(ohm, examiner=dict(ex, seed_rule="mine")), sk),
+           "no provenance": seal(dict(ohm, provenance={}), sk), "unit mismatch at a join": seal(dict(base, kind="composite", interface=comp["interface"], body={"path": [good["table"]["id"], good["formula2"]["id"]]}), sk),
+           "example with no citation": seal(dict(base, kind="example-set", interface={"outputs": [{"name": "label", "unit": "class"}]}, body={"examples": [{"x": 1}]}), sk)}
+    res["refused"] = {k: admit(v, known) for k, v in bad.items()}
+    wrong = [k for k, v in res["refused"].items() if v[0]]
+    tr = admit(good["table"], known, trusted={"someone else"})
+    assert not tr[0]
+    return {"admitted": len(res["admitted"]), "refused": len(bad) - len(wrong), "of": len(bad), "wrongly admitted": wrong, "why": {k: v[1] for k, v in res["refused"].items()}}
+
+
+if __name__ == "__main__":
+    a = sys.argv[1:]
+    if a[:1] == ["suite"] or not a:
+        print(json.dumps(_suite(), indent=1))
+    elif a[0] == "keygen":
+        import secrets
+        print(secrets.token_hex(32))
+    elif a[0] == "new":
+        print(json.dumps(seal(json.load(open(a[1], encoding="utf-8")), bytes.fromhex(open(a[2]).read().strip())), indent=1, ensure_ascii=False))
+    elif a[0] == "check":
+        print(admit(json.load(open(a[1], encoding="utf-8"))))

workbench/swarm/part_submit.py

@@ -0,0 +1,81 @@
+"""part_submit (11.0) - a part in the format joins exactly as an internal one does. It is checked against the format, then becomes a specialist in your workspace
+through the same builders the swarm uses (a table through to_workbench, a formula through compose), and it joins only by the same exam: `exam <id> dev`, then
+`exam <id> heldout`. usage: oracle7.py part check|submit <part.json>"""
+import json
+import os
+import random
+import re
+import shutil
+import sys
+import tempfile
+from pathlib import Path
+
+import exam_v1
+import part11 as P
+import run as swarm_run
+
+
+def _id(part, stem):
+    name = re.sub(r"[^a-z0-9]+", "-", str(part["provenance"].get("name") or part["id"][:12]).lower()).strip("-")
+    return ("%s-%s" % (stem, name))[:40].rstrip("-")
+
+
+def table(part):
+    """rows as a table the swarm's own organ reads, with the swarm's own exam panel"""
+    b = part["body"]
+    info, rows = b.get("property"), b.get("records")
+    if not isinstance(info, dict) or not info.get("label") or not isinstance(rows, list) or not all(isinstance(r, dict) and r.get("label") and "value" in r for r in rows):
+        return None, "a table part to install needs body.property and body.records"
+    info = dict({"aliases": [], "datatype": "string", "pid": "part"}, **info)
+    rows = [dict({"qid": "P%d" % i, "aliases": []}, **r) for i, r in enumerate(rows)]
+    tmp = Path(tempfile.mkdtemp(prefix="o7part-"))
+    here = Path(__file__).resolve().parent
+    json.dump({"property": info, "rows": rows, "source": {"repo": "part", "sha256": part["id"], "licence": part["provenance"].get("licence", "see provenance")}}, open(tmp / "table.json", "w", encoding="utf-8"), ensure_ascii=False)
+    for f in ("organ_table_v2.py", "scorer_table_v1.py", "narrow_v1.py"):
+        shutil.copy(here / f, tmp / f)
+    json.dump(exam_v1.write(info, rows), open(tmp / "panel_v1.json", "w", encoding="utf-8"), ensure_ascii=False)
+    names = [info["label"]] + [a for a in info["aliases"] if len(a) >= 3]
+    json.dump({"name": "Part: %s" % info["label"], "sources": [{"name": "part %s" % part["id"][:16], "licence": part["provenance"].get("licence", "see provenance")}],
+               "accept": {"kinds": ["lookup"], "declines_in_domain": ["unknown_item"], "context_regex": r"\b(?:%s)\b" % "|".join(re.escape(n.lower()) for n in sorted(names, key=len, reverse=True))}, "notes": "Submitted in the part format."},
+              open(tmp / "manifest.json", "w", encoding="utf-8"), ensure_ascii=False)
+    d = swarm_run.to_workbench(tmp, _id(part, "part"), built_from={"questions": ["What is the %s of %s?" % (info["label"], r["label"]) for r in rows[:3]], "gap": info["label"]})
+    shutil.rmtree(tmp, ignore_errors=True)
+    return d, "installed"
+
+
+def formula(part):
+    """worked examples from the formula, written by the format's own arithmetic, composed by the swarm's composer: it has to arrive at the same program"""
+    face = part["interface"]
+    rng = random.Random(part["id"])
+    lo, hi = part["body"].get("example_range", [1, 50])
+    lines = ["title: %s" % (part["provenance"].get("name") or part["id"][:12]), "inputs: " + "; ".join("%s (%s)" % (x.get("label", x["name"]), x["unit"]) for x in face["inputs"]),
+             "output: %s (%s)" % (face["outputs"][0]["name"], face["outputs"][0]["unit"]), "source: %s" % part["body"]["source"]]
+    for _ in range(16):
+        v = {x["name"]: round(rng.uniform(lo, hi), 3) for x in face["inputs"]}
+        lines.append("%s -> %s" % ("; ".join(str(v[x["name"]]) for x in face["inputs"]), round(P.apply(part, v), 6)))
+    r = swarm_run.compose("\n".join(lines), sid=_id(part, "part"), anyway=True)
+    return r.get("specialist"), r["message"]
+
+
+def main(rest):
+    if len(rest) < 2 or rest[0] not in ("check", "submit"):
+        print(__doc__)
+        return 2
+    part = json.load(open(rest[1], encoding="utf-8"))
+    ok, why = P.admit(part)
+    print("format: %s" % why)
+    if not ok or rest[0] == "check":
+        return 0 if ok else 1
+    go = {"table": table, "formula": formula}.get(part["kind"])
+    if go is None:
+        print("kind %s is held as a part; it has no specialist to install" % part["kind"])
+        return 0
+    d, msg = go(part)
+    print(msg)
+    if d:
+        print("now the same exam as any part:\n  python oracle7.py exam %s dev\n  python oracle7.py exam %s heldout" % (Path(d).name, Path(d).name))
+    return 0 if d else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))

workbench/swarm/ref_measure.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""ref_measure - the sleep week's reference set measured in a fresh process, as the head stands after a night (11-04). Sleep's only change
+that lives in memory is the alignment table the learner keeps; everything else (built specialists, retired parts) is on disk. So a fresh
+head on the same tree and workbench, with that night's table loaded, is the head sleep has made, and a clean state folder makes every
+measure a first ask. Runs beside the next night: the week takes about half the time.
+usage: ref_measure.py <reference.json> <out.json>   (env: O7_TREE_RUN, ORACLE7_WORKBENCH_LOCAL, ORACLE7_STATE, O7_ALIGN optional)"""
+import collections
+import importlib.util
+import json
+import os
+import sys
+import time
+
+if os.environ.get("PYTHONHASHSEED") != "0":
+    os.environ["PYTHONHASHSEED"] = "0"
+    os.execv(sys.executable, [sys.executable] + sys.argv)
+REF, OUTF = sys.argv[1], sys.argv[2]
+T = os.environ["O7_TREE_RUN"]
+spec = importlib.util.spec_from_file_location("o7", T + "/oracle7.py")
+o7 = importlib.util.module_from_spec(spec)
+sys.argv = ["oracle7.py"]
+spec.loader.exec_module(o7)
+sys.path[:0] = [T + "/workbench", T + "/workbench/swarm"]
+import server                                         # noqa: E402
+eng = o7.Engine()
+wb = server.WB = server.Workbench(eng)
+rt = getattr(eng, "_rt", None)
+for _ in range(60):
+    if rt is not None and ("facts" in rt._organs or "error" in rt._organs):
+        break
+    time.sleep(5)
+from handoff101 import answer_of, compare             # noqa: E402
+
+align = os.environ.get("O7_ALIGN")
+if align:                                             # the learner's kept table, loaded the way sleep105.dream_learner loads it
+    cov = rt._organs["facts"].coverage
+    A = json.load(open(align, encoding="utf-8"))["p_feature_given_word"]
+    cov.align_t = A
+    cov.align_w = collections.defaultdict(dict)
+    for f, ws in A.items():
+        for w, pr in ws.items():
+            cov.align_w[w][f] = pr
+    cov.lex.words |= set(cov.align_w)
+    if hasattr(cov, "_dir_mass"):
+        del cov._dir_mass
+
+
+def right(answer, want):
+    if answer is None or want in (None, "", []):
+        return False
+    return any(compare(str(answer), str(w))[0] == "CONFIRMED" for w in (want if isinstance(want, list) else [want]))
+
+
+reference = json.load(open(REF, encoding="utf-8"))
+n = wrong = 0
+rows = []
+for k, it in enumerate(reference):
+    r = wb.chat(it["q"], "ref%d" % k)
+    a = answer_of(r) if r.get("status") == "answered" else None
+    ok = a is not None and right(a, it["expect"])
+    n += ok
+    wrong += a is not None and not ok
+    rows.append({"k": k, "status": r.get("status"), "right": ok, "path": str(r.get("path"))[:80]})
+json.dump({"right": n, "wrong": wrong, "of": len(reference), "align": align, "rows": rows}, open(OUTF, "w", encoding="utf-8"))
+print("done", n, wrong, flush=True)

workbench/swarm/skeleton102.py

@@ -18,6 +18,8 @@ import collections
 import copy
 import json
 import re
+
+import steps_budget as SB
 import time
 
 import reader_v4 as R4
@@ -254,10 +256,10 @@ class Skeletons:
         self._pc = {}                                   # prefix -> what the store returns, once per parse
         self._said = {}                                 # (function, inputs) -> the judge's "said by a word" verdict on that step alone
         self._jc = {}                                   # step-set -> the judge's verdict on the whole program (order does not change it)
-        deadline = time.time() + budget
+        deadline = SB.deadline()
         out = []
         for sk in self.sk:
-            if time.time() > deadline or len(out) >= 40:
+            if SB.spent() > deadline or len(out) >= 40:
                 break
             slots = collections.Counter(s for _f, _d, sl in sk for s in sl if s in ("E", "V", "C"))
             if any(slots[t] > len(fill[t]) for t in ("E", "V")):
@@ -286,7 +288,7 @@ class Skeletons:
         return key in self._ckeys_c[concept]
 
     def _dfs(self, question, sk, fill, prog, results, i, usedEV, out, deadline):
-        if time.time() > deadline or len(out) >= 40:
+        if SB.spent() > deadline or len(out) >= 40:
             return
         rd = self.rd
         if i == len(sk):
@@ -387,7 +389,7 @@ class Skeletons:
             ran.append((empty, ins, p2, res))
         ran.sort(key=lambda t_: t_[0])
         for empty, ins, p2, res in ran:
-            if time.time() > deadline or len(out) >= 40:
+            if SB.spent() > deadline or len(out) >= 40:
                 return
             r2 = results + [i] if empty else results
             u2 = {"E": set(usedEV["E"]), "V": set(usedEV["V"])}

workbench/swarm/sleep105.py

@@ -0,0 +1,457 @@
+#!/usr/bin/env python3
+"""sleep105 (10.5) - THE SLEEP CYCLE, the roadmap's five steps, connecting what the release already has. A day's turns go through the Workbench chat, the
+same door a person uses, and are written to a chained ledger. Then, each night:
+  1 REPLAY        the day's ledger is re-run. Declined turns become typed holes (declines.classify); answered turns are re-checked against everything
+                  admitted that day, and a changed answer is reported.
+  2 CONSOLIDATION paths the router walked repeatedly become candidate composite parts, admitted only if they give the same answers as the path they
+                  replace and pass an exam on fresh inputs (the day's questions with their numbers nudged).
+  3 DREAMING      each hole is searched for across formulas (grow60: the library's chains, built by the swarm), stores (declines.resolve, and held-back readings a known
+                  answer confirms, which feed the learner) and specialists (swarm build: a table from Wikidata, through its own held-out exam).
+  4 FORGETTING    unused or dominated parts are retired with a receipt. A retired part stays in the content-addressed store, so replay can still reach it.
+  5 MORNING       reach gained, holes filled, answers changed, parts retired, with no new wrong answer as the bar. Reports are chained.
+Days are Oracle7's own failures from v37 to v38 (failure_days.py) unless O7_DAYS names sealed days. Simulated nights: each day is a batch.
+usage: sleep105.py <gate rows.jsonl for the controls> <out dir> [nights] [turns per day]"""
+import collections
+import hashlib
+import importlib.util
+import json
+import os
+import random
+import re
+import shutil
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+if os.environ.get("PYTHONHASHSEED") != "0":           # the same order oracle7.py and the gate run in
+    os.environ["PYTHONHASHSEED"] = "0"
+    os.execv(sys.executable, [sys.executable] + sys.argv)
+
+rows, OUT, NIGHTS, PER = sys.argv[1], Path(sys.argv[2]), int(sys.argv[3]) if len(sys.argv) > 3 else 7, int(sys.argv[4]) if len(sys.argv) > 4 else 300
+for d in ("parts", "store", "ledger"):
+    (OUT / d).mkdir(parents=True, exist_ok=True)
+T = os.environ["O7_TREE_RUN"]
+LOCAL = Path(os.environ["ORACLE7_WORKBENCH_LOCAL"])
+(LOCAL / "specialists").mkdir(parents=True, exist_ok=True)
+spec = importlib.util.spec_from_file_location("o7", T + "/oracle7.py")
+o7 = importlib.util.module_from_spec(spec)
+sys.argv = ["oracle7.py"]
+spec.loader.exec_module(o7)
+sys.path[:0] = [T + "/workbench", T + "/workbench/swarm"]
+import server                                         # noqa: E402
+import exam                                           # noqa: E402
+eng = o7.Engine()
+wb = server.WB = server.Workbench(eng)
+rt = getattr(eng, "_rt", None)
+for _ in range(60):
+    if rt is not None and ("facts" in rt._organs or "error" in rt._organs):
+        break
+    time.sleep(5)
+import declines                                       # noqa: E402
+import grow60 as G                                    # noqa: E402
+import gen60                                          # noqa: E402
+import part11 as P11                                  # noqa: E402
+import run as swarm_run                               # noqa: E402
+from handoff101 import answer_of, compare             # noqa: E402
+
+KEYF = OUT / "author.key"
+if not KEYF.exists():
+    KEYF.write_text(os.urandom(32).hex())
+    KEYF.chmod(0o600)
+KEY = bytes.fromhex(KEYF.read_text().strip())
+REPEAT, GROUP, UNUSED_NIGHTS, CAP = 3, 3, 3, 4        # walked 3 times; 3 example questions (swarm build's own minimum); unused 3 nights; systems a night (grow60's CAP)
+GROW = Path(os.environ.get("O7_GROW", str(OUT / "grow")))
+if not GROW.exists():
+    shutil.copytree(Path(T) / "root" / "organs10" / "physics", GROW)
+LIB = gen60.Library(json.load(open(Path(T) / "root" / "organs10" / "library" / "derived.json", encoding="utf-8")))
+STATE = {"started": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "run": os.environ.get("O7_RUN", "Trial, on Oracle7's own failures"), "nights": NIGHTS, "per_day": PER, "night": 0, "phase": "starting",
+         "history": [], "parts": [], "retired": [], "exams": [], "changed": [], "finds": [], "found_right": 0, "planted": [], "reference": [], "totals": {"turns": 0, "holes": 0, "found": 0, "admitted": 0, "reach": 0, "wrong": 0, "false_yes": 0}}
+
+
+def save():
+    STATE["updated"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
+    tmp = OUT / "state.tmp"
+    tmp.write_text(json.dumps(STATE, indent=1, default=str), encoding="utf-8")
+    tmp.replace(OUT / "state.json")
+
+
+def phase(text, cur=None):
+    STATE["phase"] = text
+    if cur is not None:
+        STATE["current"] = dict(cur)
+    save()
+
+
+def right(answer, want):
+    if answer is None or want in (None, "", []):
+        return False
+    return any(compare(str(answer), str(w))[0] == "CONFIRMED" for w in (want if isinstance(want, list) else [want]))
+
+
+def turn(q, sid):
+    r = wb.chat(q, sid)
+    return r, (answer_of(r) if r.get("status") == "answered" else None)
+
+
+class Ledger:
+    def __init__(self, path):
+        self.path, self.prev = Path(path), "0" * 64
+
+    def add(self, e):
+        e = dict(e, prev=self.prev)
+        e["hash"] = self.prev = hashlib.sha256(json.dumps(e, sort_keys=True, default=str).encode()).hexdigest()
+        with open(self.path, "a", encoding="utf-8") as f:
+            f.write(json.dumps(e, default=str) + "\n")
+
+
+def sig(path):
+    """a router path with what varies taken out"""
+    return re.sub(r"\b[0-9a-f]{8,}\b", "#", re.sub(r"\d+(\.\d+)?", "#", str(path)))
+
+
+def nudge(q, rng):
+    """a day's question with its numbers moved: a fresh input of the same form"""
+    return re.sub(r"(?<![\w.])\d+(?:\.\d+)?(?![\w.])", lambda m: str(max(1, round(float(m.group(0)) * rng.uniform(1.1, 1.9)))), q)
+
+
+def keep_part(body, kind, name, ran, sources, face=None):
+    """a passed exam -> a sealed part in the 11.0 format, stored by its hash, never overwritten"""
+    p = P11.seal({"kind": kind, "version": 1, "interface": face or {"inputs": [{"name": "question", "unit": "text"}], "outputs": [{"name": "answer", "unit": "text"}]}, "body": body,
+                  "examiner": {"generator": "variations of the day's questions", "seed_rule": "fixed-then-audited", "n": ran["items"], "must_decline_share": .2, "ran": ran}, "provenance": {"sources": sources, "name": name}, "reach": None}, KEY)
+    (OUT / "store" / (p["id"] + ".json")).write_text(json.dumps(p, indent=1), encoding="utf-8")
+    return p
+
+
+# ---- 1 REPLAY
+def replay(day_rows, night):
+    holes, changed = [], []
+    for k, e in enumerate(day_rows):
+        r, a = turn(e["q"], "replay%d-%d" % (night, k))
+        if e["status"] == "answered":
+            if a != e["answer"] and not str(e["path"]).startswith("yours:formula"):
+                changed.append({"q": e["q"], "was": e["answer"], "now": a, "path": str(r.get("path"))})
+        elif a is None:
+            holes.append({"q": e["q"], "kind": declines.classify(str(r.get("response", "")), str(r.get("path", ""))), "path": str(r.get("path", "")), "expect": e.get("expect"), "valid": e.get("valid", True),
+                          "held": (r.get("provenance") or {}).get("held_back")})
+        if k % 10 == 0:
+            phase("replay: turn %d of %d" % (k, len(day_rows)))
+    return holes, changed
+
+
+# ---- 2 CONSOLIDATION
+def consolidate(day_rows, night):
+    """repeated paths through one part become a direct route, only if it answers as the walked path does"""
+    walked = collections.defaultdict(list)
+    for e in day_rows:
+        m = re.search(r"(?:yours|specialist):([\w.-]+)", str(e.get("path", "")))
+        if e["status"] == "answered" and m:
+            walked[(sig(e["path"]), m.group(1))].append(e)
+    done = []
+    rng = random.Random(night)
+    for (s, sid), es in sorted(walked.items(), key=lambda kv: -len(kv[1])):
+        if len(es) < REPEAT or sid not in wb.mine:
+            continue
+        sp, same = wb.mine[sid], 0
+        for e in es:
+            a = sp.reply(e["q"], "route")
+            same += bool(a) and a.get("status") == "answered" and right(answer_of(a), e["answer"])
+        fresh = [nudge(e["q"], rng) for e in es for _ in range(50)][:149]
+        agree = 0
+        for k, q in enumerate(fresh):
+            a = sp.reply(q, "route")
+            _r, b = turn(q, "fresh%d" % k)
+            agree += (a is None and b is None) or (bool(a) and b is not None and right(answer_of(a), b))
+        rec = {"night": night, "kind": "route", "steps": [s[:90]], "shape": s, "right": same, "declined": 0, "outcome": "passed" if same == len(es) and agree == len(fresh) and len(fresh) >= 149 else "differs from the path it would replace (%d of %d, %d of %d fresh)" % (same, len(es), agree, len(fresh))}
+        STATE["exams"].append(rec)
+        if rec["outcome"] == "passed":
+            keep_part({"shape": s, "route_to": sid}, "reading-rule", s, {"items": len(fresh), "failed": 0}, ["paths walked on night %d" % night])
+            done.append({"kind": "route", "shape": s, "id": sid, "night": night, "steps": [s[:90]], "reach": 0, "source": "route"})
+    return done
+
+
+# ---- 3 DREAMING
+def dream_stores(hole, sid):
+    """declines.resolve: a part, the part's source, or the question read into a form the core reads"""
+    def ask_core(text):
+        r = eng.chat(text, sid + "c")
+        return {"status": r.get("status"), "response": str(r.get("response", "")), "path": str(r.get("path", ""))}
+    return declines.resolve(hole["q"], ask_core, wb.mine, server.SPECS)
+
+
+def dream_formulas(holes, night, rep):
+    """grow60's round over the night's holes: chains the formula library holds, chosen by how many holes they close, built by the swarm and admitted by its exam"""
+    planner = G.Planner(G.load_parts(str(GROW)))
+    gaps = [h for h in holes if G.ask(planner, h["q"]).get("kind") == "no_part"]
+    needs = [G.gap_needs(LIB, h["q"], INSTALLED) for h in gaps]
+    chosen = G.choose([n for n in needs if n], CAP, LIB)
+    rep["formula_gaps"] = sum(1 for n in needs if n)
+    if not chosen:
+        return []
+    phase("dreaming: the swarm builds %d systems from the formula library" % len(chosen))
+    G.build(str(GROW), LIB, chosen, OUT / "grow_runs", night)
+    INSTALLED.update(chosen)
+    before = set(planner.parts)
+    after = G.Planner(G.load_parts(str(GROW)))
+    new = sorted(set(after.parts) - before)
+    out = []
+    for pid in new:
+        p = after.parts[pid]
+        out.append({"kind": "formula", "shape": pid, "id": pid, "night": night, "steps": ["%s from %s" % (p["output"], ", ".join(p["inputs"]))], "reach": 0, "source": "formula library"})
+    STATE["exams"].append({"night": night, "kind": "formula", "steps": ["%d systems" % len(chosen)], "right": len(new), "declined": 0, "outcome": "%d parts admitted by the swarm's exam" % len(new)})
+    return out
+
+
+def dream_specialists(holes, night):
+    """holes that ask for the same property go to swarm build, as its example questions, then through its exam. Variations come from the builder's panel"""
+    groups = collections.defaultdict(list)
+    for h in holes:
+        g = (h.get("gap") or {})
+        if g.get("build") == "table":
+            groups[g["property"].lower()].append(h)
+    out = []
+    for prop, hs in sorted(groups.items(), key=lambda kv: -len(kv[1])):
+        if len(hs) < GROUP:
+            continue
+        phase("dreaming: building a table of %s from %d holes" % (prop, len(hs)))
+        try:
+            b = swarm_run.build([h["q"] for h in hs[:6]])
+        except Exception as e:  # noqa: BLE001 - a failed build costs that candidate
+            STATE["exams"].append({"night": night, "kind": "table", "steps": [prop], "right": 0, "declined": 0, "outcome": "build failed: %s" % str(e)[:80]})
+            continue
+        d = b.get("specialist")
+        rec = {"night": night, "kind": "table", "steps": [prop], "right": 0, "declined": 0, "outcome": str(b.get("message"))[:110]}
+        STATE["exams"].append(rec)
+        if not d:
+            continue
+        v = exam.run(d, "dev", eng)
+        if v["decision"] not in ("PASS", "ADMIT"):
+            rec["outcome"] = "dev: " + "; ".join(v["reasons"])[:110]
+            continue
+        v = exam.run(d, "heldout", eng)
+        rec.update(right=v["with_specialist"]["ok"], outcome=v["decision"] + (": " + "; ".join(v["reasons"])[:100] if v["reasons"] else ""))
+        if v["decision"] == "ADMIT":
+            wb.reload()
+            keep_part({"rows": [[prop, Path(d).name]]}, "table", Path(d).name, {"items": v["n"], "failed": 0}, ["swarm build from %d holes" % len(hs)], {"inputs": [{"name": "property", "unit": "text"}], "outputs": [{"name": "specialist", "unit": "id"}]})
+            out.append({"kind": "table", "shape": prop, "id": Path(d).name, "night": night, "steps": [prop], "reach": 0, "source": "swarm build"})
+    return out
+
+
+def dream_learner(holes, night, rep):
+    """a held-back reading the day's known answer confirms becomes a labelled pair for the learner, and the alignment is relearned with them. The new table stays only if
+    every reading it newly states is right and no must-decline gains an answer: a confirmation is data, and the exam decides"""
+    cov = rt._organs["facts"].coverage
+    new = [h for h in holes if h.get("held") and h.get("valid", True) is not False and right(h["held"]["answer"], h.get("expect"))]
+    rep["confirmed"] = len(new)
+    PAIRS.extend({"question": h["q"], "program": h["held"]["program"]} for h in new)
+    if not new:
+        return []
+    phase("dreaming: relearning the alignment with %d confirmed readings" % len(PAIRS), rep)
+    pf, out = OUT / "pairs.json", OUT / ("alignment_%d.json" % night)
+    pf.write_text(json.dumps(PAIRS), encoding="utf-8")
+    subprocess.run([sys.executable, "/opt/rel105/dev/align_learn105.py", str(out)], cwd=T, env=dict(os.environ, O7_EXTRA_PAIRS=str(pf)), capture_output=True, timeout=3000)
+
+    def use(path):
+        A = json.load(open(path, encoding="utf-8"))["p_feature_given_word"]
+        cov.align_t = A
+        cov.align_w = collections.defaultdict(dict)
+        for f, ws in A.items():
+            for w, pr in ws.items():
+                cov.align_w[w][f] = pr
+        cov.lex.words |= set(cov.align_w)
+        if hasattr(cov, "_dir_mass"):
+            del cov._dir_mass
+    was = STATE.get("alignment") or str(Path(T) / "workbench" / "swarm" / "data" / "alignment102.json")
+    use(out)
+    said = wrong = 0
+    bad = None
+    phase("exam: every held reading and control against the new table", rep)
+    for k, h in enumerate(h for h in holes if h.get("held")):
+        _r, a = turn(h["q"], "learn%d-%d" % (night, k))
+        if a is not None:
+            said += 1
+            if h.get("valid", True) is False or not right(a, h.get("expect")):
+                wrong += 1
+                bad = bad or h["q"][:100]
+    rec = {"night": night, "kind": "learner", "steps": ["alignment relearned with %d confirmed readings" % len(PAIRS)], "right": said - wrong, "declined": 0, "outcome": "kept: %d newly stated, none wrong" % said if said and not wrong else ("no reading rose to the threshold" if not said else "refused, %d wrong: %s" % (wrong, bad))}
+    STATE["exams"].append(rec)
+    if said and not wrong:
+        STATE["alignment"] = str(out)
+        keep_part({"examples": [{"question": x["question"], "cite": "confirmed against the known answer"} for x in PAIRS]}, "example-set", "confirmed readings, night %d" % night, {"items": max(149, len(new)), "failed": 0}, ["held-back readings confirmed on night %d" % night], {"outputs": [{"name": "alignment", "unit": "table"}]})
+        return [{"kind": "learner", "shape": "alignment %d" % night, "id": "reader90", "night": night, "steps": [rec["steps"][0]], "reach": 0, "source": "the learner"}]
+    use(was)
+    return []
+
+
+# ---- the days
+days = json.load(open(os.environ.get("O7_DAYS", "/opt/rel105/failure_days.json"), encoding="utf-8"))
+items = [json.loads(x) for x in open(rows, encoding="utf-8") if x.strip()]
+controls = [x for x in items if x.get("valid") is False and x.get("status") != "answered" and not x["id"].startswith("REG")]
+random.Random(106).shuffle(controls)
+REF_N = int(os.environ.get("O7_REF_N", "200"))
+reference, days = days[-REF_N:], days[:len(days) - REF_N]
+STATE["pool"] = {"valid": len(days), "controls": len(controls), "reference": len(reference)}
+
+
+def measure():
+    n = wrong = 0
+    for k, it in enumerate(reference):
+        _r, a = turn(it["q"], "ref%d" % k)
+        if a is not None:
+            n += right(a, it["expect"])
+            wrong += not right(a, it["expect"])
+    return n, wrong
+
+
+def measure_fork(night):
+    """the reference measured in a fresh process (ref_measure.py), as the head stands after night `night`, while the next night runs (11-04:
+    about half the time). The only state sleep keeps in memory, the learner's alignment table, is passed by path; built specialists are on
+    disk. A clean state folder each night makes every measure a first ask: in one process a second ask moved 2 of 100 answers. (A fork
+    was tried first: the head's threads leave a forked copy waiting on a lock no thread will release.)"""
+    out = OUT / "ref" / ("night%d.json" % night)
+    out.parent.mkdir(exist_ok=True)
+    rf = OUT / "ref" / "reference.json"
+    if not rf.exists():
+        rf.write_text(json.dumps(reference), encoding="utf-8")
+    st = OUT / "ref" / ("state%d" % night)
+    st.mkdir(exist_ok=True)
+    env = dict(os.environ, ORACLE7_STATE=str(st), PYTHONHASHSEED="0")
+    if STATE.get("alignment"):
+        env["O7_ALIGN"] = STATE["alignment"]
+    p = subprocess.Popen([sys.executable, str(Path(__file__).resolve().parent / "ref_measure.py"), str(rf), str(out)], env=env,
+                         stdout=open(str(out) + ".log", "w"), stderr=subprocess.STDOUT)
+    return p, out
+
+
+def measured(pending):
+    p, out = pending
+    p.wait()
+    r = json.loads(out.read_text())
+    return r["right"], r["wrong"]
+
+
+def write_report(rep, a_, w_):
+    global chain
+    rep["reference_right"], rep["reference_wrong"] = a_, w_
+    STATE["reference"].append({"night": rep["night"], "right": a_, "wrong": w_, "of": len(reference)})
+    body = json.dumps(dict(rep, prev=chain), sort_keys=True)
+    chain = hashlib.sha256(body.encode()).hexdigest()
+    with open(OUT / "reports.jsonl", "a", encoding="utf-8") as f:
+        f.write(json.dumps({"report": json.loads(body), "hash": chain}) + "\n")
+    save()
+    print(json.dumps(rep), flush=True)
+
+
+INSTALLED, parts, retired, chain, PAIRS = set(), [], [], "0" * 64, []
+planner_live = G.Planner(G.load_parts(str(GROW)))
+pending = (measure_fork(0), None)                  # (the process measuring, the report waiting on it); night 0 has no report
+if os.environ.get("O7_REF_CHECK"):                 # test runs only: night 0 measured here too, to show the fresh process measures the same head
+    STATE["ref_check"] = dict(zip(("right", "wrong"), measure()))
+    save()
+per_c = max(1, PER // 6)
+for night in range(1, NIGHTS + 1):
+    STATE["night"] = night
+    day = days[(night - 1) * PER:night * PER] + [dict(x, valid=False) for x in controls[(night - 1) * per_c:night * per_c]]
+    if not day:
+        break
+    random.Random(night).shuffle(day)
+    led = Ledger(OUT / "ledger" / ("day%d.jsonl" % night))
+    rep = {"night": night, "turns": len(day), "holes": 0, "found": 0, "gained": 0, "wrong": 0, "false_yes": 0, "admitted": 0, "retired": 0, "changed": 0, "candidates": 0, "failed": 0}
+    t0 = time.time()
+    day_rows = []
+    for k, it in enumerate(day):
+        r, a = turn(it["q"], "n%d-%d" % (night, k))
+        path = str(r.get("path", ""))
+        if a is None:                                 # the formula parts this run has grown
+            p = G.ask(planner_live, it["q"])
+            if p.get("status") == "answered":
+                a, path = str(p.get("value")), "yours:formula>" + str(p.get("plan"))[:80]
+        e = {"q": it["q"], "status": "answered" if a is not None else str(r.get("status")), "answer": a, "path": path, "expect": it.get("expect"), "valid": it.get("valid", True)}
+        led.add(e)
+        day_rows.append(e)
+        used = next((p for p in parts if p["id"] in path), None)
+        if a is not None:
+            if it.get("valid", True) is False:
+                rep["false_yes"] += 1
+                if used:
+                    used["bad"] = "false yes: " + it["q"][:120]
+            elif used:
+                if right(a, it.get("expect")):
+                    rep["gained"] += 1
+                    used["reach"] += 1
+                    used["last_used"] = night
+                else:
+                    rep["wrong"] += 1
+                    used["bad"] = "wrong: %s -> %s" % (it["q"][:100], str(a)[:60])
+        if k % 10 == 0:
+            phase("day: turn %d of %d" % (k, len(day)), rep)
+    holes, changed = replay(day_rows, night)
+    rep["changed"] = len(changed)
+    STATE["changed"] = (STATE["changed"] + [dict(c, night=night) for c in changed])[-40:]
+    rep["holes"] = len(holes)
+    phase("consolidation", rep)
+    new = consolidate(day_rows, night)
+    open_holes = []
+    for k, h in enumerate(holes):
+        if h.get("valid") is False:
+            continue
+        if k % 10 == 0:
+            phase("dreaming: stores, hole %d of %d" % (k, len(holes)), rep)
+        d = dream_stores(h, "h%d-%d" % (night, k))
+        h["gap"] = d.get("gap")
+        if d["status"] == "answered":
+            f = {"night": night, "q": h["q"], "answer": str(d["response"])[:200], "source": d.get("kind"), "right": right(d["response"], h.get("expect")), "valid": True, "steps": [d.get("fix")]}
+            rep["found"] += 1
+            STATE["found_right"] += int(f["right"])
+            STATE["finds"] = (STATE["finds"] + [f])[-60:]
+        else:
+            open_holes.append(h)
+    phase("dreaming: formulas", rep)
+    new += dream_formulas(open_holes, night, rep)
+    new += dream_specialists(open_holes, night)
+    new += dream_learner(holes, night, rep)
+    planner_live = G.Planner(G.load_parts(str(GROW)))
+    rep["admitted"] = len(new)
+    # an exam record can stand for several parts (the swarm's formula exam), so refusals are counted as records and candidates as
+    # what was admitted plus what was refused (11-04: candidates minus admitted had gone negative)
+    kept = ("passed", "kept", "admit")
+    rep["failed"] = sum(1 for x in STATE["exams"] if x.get("night") == night and not str(x.get("outcome", "")).lower().startswith(kept)
+                        and "parts admitted" not in str(x.get("outcome", "")))
+    rep["candidates"] = rep["admitted"] + rep["failed"]
+    for p in new:
+        p["last_used"] = night
+    parts += new
+    # ---- 4 FORGETTING
+    keep = []
+    for p in parts:
+        if p.get("bad") or night - p.get("last_used", p["night"]) >= UNUSED_NIGHTS:
+            rep["retired"] += 1
+            retired.append({"id": p["id"], "shape": p["shape"], "night": night, "receipt": p.get("bad") or "unused for %d nights" % UNUSED_NIGHTS})
+            if (server.SPECS / p["id"]).exists():
+                shutil.move(str(server.SPECS / p["id"]), str(OUT / "store" / p["id"]))
+                wb.reload()
+        else:
+            keep.append(p)
+    parts = keep
+    # ---- 5 MORNING
+    rep["seconds"] = round(time.time() - t0)
+    STATE["parts"] = [{k: p.get(k) for k in ("id", "kind", "shape", "steps", "source", "night", "reach")} for p in parts]
+    STATE["retired"] = retired
+    tt = STATE["totals"]
+    for a, b in (("turns", "turns"), ("holes", "holes"), ("found", "found"), ("admitted", "admitted"), ("reach", "gained"), ("wrong", "wrong"), ("false_yes", "false_yes")):
+        tt[a] += rep[b]
+    STATE.pop("current", None)
+    STATE["history"].append(rep)
+    phase("measuring the reference set")
+    fork, waiting = pending
+    a_, w_ = measured(fork)
+    if waiting is None:
+        STATE["reference"].append({"night": 0, "right": a_, "wrong": w_, "of": len(reference)})
+    else:
+        write_report(waiting, a_, w_)
+    pending = (measure_fork(night), rep)
+fork, waiting = pending
+write_report(waiting, *measured(fork))
+STATE["phase"] = "finished"
+save()
+print("done", flush=True)

workbench/swarm/steps_budget.py

@@ -0,0 +1,15 @@
+"""steps_budget (10.5) - search is bounded by counted steps, so a reading is the same on any machine under any load"""
+LIMIT = 12000
+
+
+class Budget:
+    n = 0
+
+
+def deadline():
+    return Budget.n + LIMIT
+
+
+def spent():
+    Budget.n += 1
+    return Budget.n

workbench/verify.py

@@ -126,6 +126,10 @@ def verify_install(root=None):
     for f in ("oracle7.py", "PACK_MANIFEST.json"):
         if (INSTALL / f).exists():
             parts[f] = hashlib.sha256((INSTALL / f).read_bytes()).hexdigest()
+    for top in ("orbiters", "records"):              # recipe v5 (10.5): each orbiter, and the records the changelog points to
+        for d in sorted(p for p in (INSTALL / top).glob("*") if p.is_dir() and p.name not in SKIP_DIRS):
+            if top + "/" + d.name in pub["parts"] or any(k.startswith(top + "/") for k in pub["parts"]):
+                parts[top + "/" + d.name] = part_hash(d)[0]
     if any(k.startswith("data/") for k in pub["parts"]) and (INSTALL / "PACK_MANIFEST.json").exists():   # recipe v4 (10.2)
         manifest = json.loads((INSTALL / "PACK_MANIFEST.json").read_text(encoding="utf-8"))
         parts.update({k: v[0] for k, v in data_parts(R, manifest).items()})