chore(sync): mirror backend .py + Dockerfile to Space (hf-sync-backend)

Automated backend sync from szl-holdings/a11oy main via hf-sync-backend.
Updated (differed from the Space): Dockerfile, serve.py, szl_corpus_publish.py, szl_energy_operator.py
Deleted (gone from the repo + Dockerfile COPY set): (none)

Keeps the Space-built backend (serve.py + the Dockerfile-COPY'd .py
modules) identical to GitHub main so the Space never rebuilds from a
stale backend, new endpoints don't 404 there, and orphaned modules
removed from the repo don't linger in the Space tree.

Dockerfile

@@ -406,6 +406,14 @@ COPY szl3d_holographic.py ./szl3d_holographic.py
 # sovereign in-image. Reuses the vendor3d Three.js r160 above — 0 CDN.
 COPY cathedral_genius.html ./cathedral_genius.html
 COPY static/cathedral_app.js ./static/cathedral_app.js
+# ADDITIVE (holographic front-door landing, Dev1): the governed-inference-field
+# hero served at "/" by serve.py.spa_root (cathedral one click in at /cathedral,
+# console at /console). a11oy_landing.html is the page; static/a11oy_landing.js is
+# the ES module served at /landing/app.js. Reuses the vendor3d Three.js r160 above
+# (MIT) via the page importmap — 0 runtime CDN. MUST be per-file COPY'd or "/"
+# falls back to the cathedral/console. Doctrine v11 LOCKED; Λ = Conjecture 1.
+COPY a11oy_landing.html ./a11oy_landing.html
+COPY static/a11oy_landing.js ./static/a11oy_landing.js
 # ADDITIVE: batch-2 sovereign security data module (imported by serve.py; try/except-guarded).
 # ADDITIVE: a11oy.code conversational orchestrator module (imported by serve.py).
 # ADDITIVE (a11oy Code agentic core, 2026-06-10): the GENUINELY-agentic loop + agentic

serve.py

@@ -8002,7 +8002,12 @@ async def spa_root():
     from starlette.responses import Response as _SPA_Resp
     _SPA_TAG = (b'<script src="/vendor/a11oy-operator-widget.js" '
                 b'data-surface="a11oy" defer></script>')
-    for _cand in (Path("/app/cathedral.html"), PAGES_DIR / "console.html", INDEX_HTML):
+    # Front door order: the holographic landing first (governed-inference field,
+    # vendored Three.js r160, live receipt/mesh weave-ins + in-browser WebCrypto
+    # verify), then the cathedral hero, then the console SPA, then the SPA index.
+    # All fallbacks preserved so a missing file never white-screens "/".
+    for _cand in (Path("/app/a11oy_landing.html"), Path("/app/cathedral.html"),
+                  PAGES_DIR / "console.html", INDEX_HTML):
         try:
             _cp = Path(_cand)
             if not _cp.is_file():
@@ -8079,6 +8084,21 @@ async def _cathedral_app_js() -> Response:
 # === end /cathedral canonical genius unification ===
 
 
+# === ADDITIVE: holographic front-door landing ES module (Dev1) ===
+# /landing/app.js serves the governed-inference-field hero (instanced particles +
+# fresnel governance core). ES module, imports "three" via the importmap in
+# a11oy_landing.html -> /hero/vendor3d/three.module.min.js (vendored r160, MIT, 0 CDN).
+# Registered BEFORE the SPA /{full_path:path} catch-all so it wins the ordered match.
+# Signed-off-by: Stephen P. Lutar Jr. <stephenlutar2@gmail.com>
+@app.get("/landing/app.js")
+async def _landing_app_js() -> Response:
+    f = Path("/app/static/a11oy_landing.js")
+    if f.is_file():
+        return FileResponse(str(f), media_type="application/javascript; charset=utf-8",
+                            headers={"Cache-Control": "public, max-age=3600"})
+    return JSONResponse({"error": "landing app.js missing"}, status_code=404)
+
+
 
 # --- Doctrine v13 organ page routes (ADDITIVE; explicit, win over SPA catch-all) ---
 PAGES_DIR = Path("/app/pages")

szl_corpus_publish.py

@@ -115,6 +115,84 @@ def _canon(obj: Any) -> bytes:
     return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
 
 
+# --------------------------------------------------------------------------- #
+# Verify-before-publish gate
+# --------------------------------------------------------------------------- #
+# The published corpus advertises that every ecdsa-p256-dsse-pae receipt verifies
+# against ONE pinned cosign.pub — the same key the CI re-verify guard checks
+# (.github/hf-corpus-guards.json -> cosign_pub_pem). Incident #325: two receipts
+# signed by a transient/rotated key (matching the live org cosign.pub, not the
+# pinned one) were published and could no longer re-verify. To make that
+# impossible going forward, the producer now re-verifies each signed envelope
+# against the SAME pinned key before publishing; an envelope that does not verify
+# is skipped (honestly, like an UNSIGNED one) — never published.
+#
+# The pinned PEM is read from the guard config so producer and guard share one
+# source of truth; the embedded copy is only a fallback for a runtime that does
+# not ship the .github config. Keep it in sync with that file's cosign_pub_pem.
+_CORPUS_COSIGN_PUB_PEM_FALLBACK = (
+    "-----BEGIN PUBLIC KEY-----\n"
+    "MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE/Jlv9FnwJ13l4QIZpr4IbTBUtVZ2\n"
+    "i+O7Jai/s7xsdXvOjmZGYhd36VxNQQahTSjWoYpPrSNhXbt/n7lsgi61xA==\n"
+    "-----END PUBLIC KEY-----\n"
+)
+
+
+def _corpus_verify_pub_pem() -> str:
+    """Return the pinned cosign.pub the corpus re-verifies against. Reads the CI
+    guard config (single source of truth) and falls back to the embedded copy."""
+    cfg_path = os.path.join(
+        os.path.dirname(os.path.abspath(__file__)),
+        ".github", "hf-corpus-guards.json")
+    try:
+        with open(cfg_path, "r", encoding="utf-8") as fh:
+            pem = json.load(fh).get("cosign_pub_pem")
+        if isinstance(pem, str) and "BEGIN PUBLIC KEY" in pem:
+            return pem
+    except Exception:
+        pass
+    return _CORPUS_COSIGN_PUB_PEM_FALLBACK
+
+
+def _ecdsa_envelope_verifies(env: Dict[str, Any], pub_pem: str) -> bool:
+    """True iff at least one of the envelope's signatures verifies over its DSSE
+    PAE against pub_pem. Conservative: any error (incl. missing cryptography)
+    returns False so an envelope we cannot vouch for is NOT published."""
+    try:
+        import base64
+        from cryptography.hazmat.primitives.serialization import load_pem_public_key
+        from cryptography.hazmat.primitives import hashes
+        from cryptography.hazmat.primitives.asymmetric import ec
+        from cryptography.exceptions import InvalidSignature
+    except Exception:
+        return False
+    try:
+        body = base64.b64decode(env.get("payload", "") or b"")
+        pt = str(env.get("payloadType", ""))
+        pae = b"DSSEv1 %d %s %d %s" % (len(pt.encode()), pt.encode(), len(body), body)
+        pub = load_pem_public_key(pub_pem.encode("utf-8"))
+        for s in env.get("signatures") or []:
+            try:
+                pub.verify(base64.b64decode(s.get("sig", "") or ""), pae,
+                           ec.ECDSA(hashes.SHA256()))
+                return True
+            except InvalidSignature:
+                continue
+        return False
+    except Exception:
+        return False
+
+
+def _publishable_against_corpus_key(env: Dict[str, Any], scheme: str) -> bool:
+    """Verify-before-publish gate. ecdsa-p256-dsse-pae receipts are published only
+    if they verify against the pinned corpus cosign.pub, so a transient/rotated-key
+    envelope can never enter the corpus and later fail re-verify. sigstore-keyless
+    receipts are not gated here (verified via their Fulcio cert at re-verify time)."""
+    if scheme != "ecdsa-p256-dsse-pae":
+        return True
+    return _ecdsa_envelope_verifies(env, _corpus_verify_pub_pem())
+
+
 # --------------------------------------------------------------------------- #
 # Receipts
 # --------------------------------------------------------------------------- #
@@ -235,6 +313,13 @@ def on_new_receipt(env: Dict[str, Any], *, extra: Optional[Dict[str, Any]] = Non
             return {"ok": False, "skipped": "not-an-envelope"}
         if not _is_real_signed(env):
             return {"ok": True, "skipped": "unsigned-or-placeholder", "published": 0}
+        scheme = _detect_scheme(env)
+        if not _publishable_against_corpus_key(env, scheme):
+            # Genuinely signed but NOT against the pinned corpus key (e.g. a
+            # transient/rotated key). Refuse to publish rather than poison the
+            # corpus with an envelope that would later fail re-verify (#325).
+            return {"ok": True, "skipped": "signature-not-verifiable-against-corpus-key",
+                    "published": 0, "scheme": scheme}
         bucket = _get_bucket(PREFIX_RECEIPT)
         if bucket is None:
             return {"ok": False, "skipped": "bucket-unavailable", "published": 0}
@@ -257,16 +342,21 @@ def backfill_receipts(envelopes: Iterable[Dict[str, Any]], *, flush: bool = True
     bucket = _get_bucket(PREFIX_RECEIPT, start=False)
     if bucket is None:
         return {"ok": False, "error": "bucket-unavailable"}
-    queued = skipped = 0
+    queued = skipped = skipped_unverifiable = 0
     for env in envelopes:
         if not (isinstance(env, dict) and _is_real_signed(env)):
             skipped += 1
             continue
+        if not _publishable_against_corpus_key(env, _detect_scheme(env)):
+            # signed, but not against the pinned corpus key -> never publish (#325)
+            skipped_unverifiable += 1
+            continue
         rec = make_receipt_record(env)
         wrapped = bucket.make_record(rec, kind="receipt", source=SOURCE, dedup_key=rec["receipt_uid"])
         res = bucket.append(wrapped, kind="receipt", source=SOURCE, auto_flush=False)
         queued += res.get("queued", 0)
-    out: Dict[str, Any] = {"ok": True, "queued": queued, "skipped_unsigned": skipped}
+    out: Dict[str, Any] = {"ok": True, "queued": queued, "skipped_unsigned": skipped,
+                           "skipped_unverifiable": skipped_unverifiable}
     if flush:
         out["flush"] = bucket.flush_queue(force=True)
     return out

szl_energy_operator.py

@@ -130,6 +130,14 @@ THROTTLE_SLEEP_MULT_ENV = "A11OY_ENERGY_THROTTLE_SLEEP_MULT"
 _THROTTLE_SLEEP_MULT_DEFAULT = 4.0
 # Forced posture override for demos/tests (soak|baseline|throttle). Empty => live.
 FORCE_POSTURE_ENV = "A11OY_ENERGY_FORCE_POSTURE"
+# Governed-compute switch (default ON). When ON, each completed REAL GPU job is run
+# through the EXISTING governed turn (a11oy_vertical_feeds.governed_turn: Λ aggregator
+# + locked formulas + deny-by-default gates, sealing a Khipu/DSSE receipt into the Lake)
+# and the honest result is attached to the JobRecord as ADDITIVE governance metadata.
+# This NEVER touches how joules are MEASURED. Off => the prior byte-identical job path.
+GOVERN_COMPUTE_ENV = "A11OY_GOVERN_COMPUTE"
+# Governance vertical/organ the GPU jobs are sealed under (its own Lake DAG).
+GOVERN_VERTICAL = "sovereign-compute"
 # The harvest posture feed (a11oy_harvest_endpoints.handle_posture) can hit live
 # external energy feeds, so it is refreshed at most once per this TTL — NEVER on
 # every sweep. Keeps the hot loop gentle (a fast inter-job interval must not turn
@@ -148,6 +156,116 @@ def _useful_work_enabled() -> bool:
             not in ("0", "false", "no", "off", ""))
 
 
+def _govern_enabled() -> bool:
+    """True unless A11OY_GOVERN_COMPUTE is explicitly falsey. Default ON."""
+    return (os.environ.get(GOVERN_COMPUTE_ENV, "1").strip().lower()
+            not in ("0", "false", "no", "off", ""))
+
+
+def _ungoverned(reason: str) -> dict:
+    """Honest ungoverned governance record — STABLE schema, NO fabricated Λ/receipt."""
+    return {
+        "governed": False,
+        "lambda_score": None,
+        "lambda_pass": None,
+        "decision": None,
+        "receipt_id": None,
+        "dsse_keyid": None,
+        "dsse_signed": False,
+        "doctrine_lambda": None,
+        "vertical": GOVERN_VERTICAL,
+        "reason": reason,
+    }
+
+
+def _govern_turn(kind: str, text: str, model: str,
+                 node_name: str) -> dict:
+    """Run a COMPLETED GPU job through the EXISTING governed turn as ADDITIVE metadata.
+
+    This is the governed-compute boundary: it imports + calls the real
+    `a11oy_vertical_feeds.governed_turn` (which scores the turn through the Λ aggregator
+    + locked formulas + deny-by-default gates, then seals a Khipu receipt into the Lake
+    and signs a DSSE envelope) and distils the honest result.
+
+    Doctrine v11 — NEVER fabricate: any failure (disabled / module absent / call error /
+    receipt not sealed) returns an honest ungoverned record (governed=False, lambda_score
+    None, receipt_id None) with a reason. A job that could not be governed says so; it is
+    never faked as governed. The return schema is STABLE (same keys whether governed or
+    not). This function does NOT touch energy measurement and never raises."""
+    if not _govern_enabled():
+        return _ungoverned(f"disabled ({GOVERN_COMPUTE_ENV}=0)")
+    try:
+        import a11oy_vertical_feeds as _VF  # type: ignore
+    except Exception as e:  # noqa: BLE001 — governance optional; absence stays honest
+        return _ungoverned(f"governance modules unavailable: {type(e).__name__}")
+    try:
+        res = _VF.governed_turn(
+            vertical=GOVERN_VERTICAL,
+            text=text or "",
+            action_kind=f"gpu-{kind}",
+            context={"task": "energy-operator", "model": model, "node": node_name},
+        )
+    except Exception as e:  # noqa: BLE001 — never let governance break the job path
+        return _ungoverned(f"governed_turn failed: {type(e).__name__}")
+    if not isinstance(res, dict):
+        return _ungoverned("governed_turn returned non-dict")
+    lam = res.get("lambda")
+    receipt = res.get("receipt") if isinstance(res.get("receipt"), dict) else {}
+    receipt_id = receipt.get("digest")
+    dsse = res.get("dsse") if isinstance(res.get("dsse"), dict) else {}
+    keyid = None
+    sigs = dsse.get("signatures")
+    if isinstance(sigs, list) and sigs and isinstance(sigs[0], dict):
+        keyid = sigs[0].get("keyid")
+    doctrine = res.get("doctrine") if isinstance(res.get("doctrine"), dict) else {}
+    # Fully governed ONLY when a real Λ score AND a real sealed receipt id are present.
+    fully = (isinstance(lam, (int, float)) and bool(receipt_id))
+    out = {
+        "governed": fully,
+        "lambda_score": lam if isinstance(lam, (int, float)) else None,
+        "lambda_pass": res.get("lambda_pass"),
+        "decision": res.get("decision"),
+        "receipt_id": receipt_id,
+        "dsse_keyid": keyid,
+        "dsse_signed": bool(dsse.get("signed")),
+        "doctrine_lambda": doctrine.get("lambda"),
+        "vertical": res.get("vertical"),
+    }
+    if not fully:
+        out["reason"] = ("Λ scored but receipt not sealed"
+                         if isinstance(lam, (int, float))
+                         else "governed_turn returned no Λ score")
+    return out
+
+
+def _governed_compute_summary(recent_records: list[dict]) -> dict:
+    """HONEST, observable governed-compute block for status(). Derived from the rolling
+    tail of recent JobRecords (no fabricated aggregate). Reports whether governance is
+    enabled, how many of the recent jobs were FULLY governed (real Λ + sealed receipt),
+    and the most recent governance result (Λ score, receipt id, dsse keyid) so a reader
+    can see a GPU job was a governed turn — or honestly was not."""
+    gov_records = [r for r in recent_records
+                   if isinstance(r, dict) and isinstance(r.get("governance"), dict)]
+    fully = [g for g in gov_records if g["governance"].get("governed")]
+    last = gov_records[-1]["governance"] if gov_records else None
+    return {
+        "enabled": _govern_enabled(),
+        "vertical": GOVERN_VERTICAL,
+        "recent_window": len(recent_records),
+        "recent_governed": len(fully),
+        "recent_attempted": len(gov_records),
+        "last": last,
+        "note": (
+            "Each completed GPU job is ADDITIONALLY run through the EXISTING governed "
+            "turn (a11oy_vertical_feeds.governed_turn: Λ aggregator + locked formulas + "
+            "deny-by-default gates, sealing a Khipu/DSSE receipt into the Lake). Λ is "
+            "Conjecture 1 — never a theorem. A job that could not be governed is honestly "
+            "labeled governed=False with NO fabricated Λ score or receipt. Governance is "
+            "metadata only; it NEVER changes how joules are MEASURED."
+        ),
+    }
+
+
 def _price_expensive_threshold() -> float:
     try:
         return float(os.environ.get(PRICE_EXPENSIVE_ENV, _PRICE_EXPENSIVE_DEFAULT))
@@ -509,6 +627,15 @@ class JobRecord:
     # contract is unchanged — these are extra optional fields with safe defaults.
     useful_work: bool = False
     rag_chunk_id: Optional[str] = None
+    # ADDITIVE (governed-compute): the honest result of running THIS completed GPU job
+    # through the existing governed turn (Λ aggregator + locked formulas + Khipu/DSSE
+    # receipt sealed into the Lake). None when governance was not attempted; when
+    # attempted it is a dict carrying governed(bool), lambda_score, decision, receipt_id,
+    # dsse_keyid/dsse_signed and the Conjecture-1 doctrine label — or governed=False + a
+    # reason when it could not be governed (NEVER a fabricated score/receipt). Energy
+    # measurement is unaffected; this is metadata only. Dev2/3/4 contract is unchanged
+    # (extra optional field, safe default).
+    governance: Optional[dict] = None
 
     def to_dict(self) -> dict:
         return {
@@ -519,6 +646,7 @@ class JobRecord:
             "joules_label": self.joules_label, "joules_evidence": self.joules_evidence,
             "ts": self.ts, "seq": self.seq,
             "useful_work": self.useful_work, "rag_chunk_id": self.rag_chunk_id,
+            "governance": self.governance,
         }
 
 
@@ -1052,7 +1180,8 @@ class OperatorDaemon:
                 wall_s: float, exporter_sample: Optional[dict],
                 joules_measured: Optional[float], *,
                 useful_work: bool = False,
-                rag_chunk_id: Optional[str] = None) -> JobRecord:
+                rag_chunk_id: Optional[str] = None,
+                governance: Optional[dict] = None) -> JobRecord:
         now = time.time()
         label = _label_upper(exporter_sample, now=now)
         evidence = _J.joules_evidence(exporter_sample, now=now) if label == LABEL_MEASURED else {}
@@ -1084,7 +1213,7 @@ class OperatorDaemon:
             node=node_name, model=model, kind=kind, tokens=int(tokens), wall_s=wall_s,
             joules_measured=billable_j,
             joules_label=label, joules_evidence=evidence, ts=_now_iso(), seq=seq,
-            useful_work=useful_work, rag_chunk_id=rag_chunk_id)
+            useful_work=useful_work, rag_chunk_id=rag_chunk_id, governance=governance)
         self._emit(rec)
         return rec
 
@@ -1113,12 +1242,15 @@ class OperatorDaemon:
         j_before = (sample_before or {}).get("joules_measured_total")
         rag_chunk_id: Optional[str] = None
         embed_vec: list[float] = []
+        governed_text = ""  # the turn text handed to the governed turn (post-meter)
         t0 = time.time()
         try:
             if kind == "generate":
                 prompt = _GEN_PROMPTS[self._state.seq % len(_GEN_PROMPTS)]
-                tokens, _ = _ollama_generate(node.base_url, node.gen_model, prompt)
+                tokens, completion = _ollama_generate(node.base_url, node.gen_model, prompt)
                 model = node.gen_model
+                # The governed turn scores the full turn (prompt + completion).
+                governed_text = f"{prompt}\n{completion}".strip()
             else:
                 # USEFUL WORK: embed a REAL un-embedded corpus chunk when available
                 # (honest fallback to canned text otherwise). The joule meter window
@@ -1127,6 +1259,7 @@ class OperatorDaemon:
                 text, rag_chunk_id = self._pick_corpus_embed_job()
                 tokens, embed_vec = _ollama_embed(node.base_url, node.embed_model, text)
                 model = node.embed_model
+                governed_text = (text or "").strip()
         except Exception:  # noqa: BLE001 — node failed mid-job: DEGRADED, never faked
             with self._lock:
                 self._node_status[node.name] = "DEGRADED"
@@ -1149,11 +1282,18 @@ class OperatorDaemon:
         stored = False
         if rag_chunk_id is not None and embed_vec:
             stored = self._store_rag_vector(rag_chunk_id, embed_vec)
+        # GOVERNED COMPUTE (ADDITIVE): run the completed turn through the EXISTING governed
+        # turn (Λ aggregator + locked formulas + Khipu/DSSE receipt sealed into the Lake).
+        # This is OUTSIDE the metered window (after meter_after) and CANNOT change the joule
+        # label/billable energy — governance is metadata only. Honest: a failure yields
+        # governed=False with no fabricated Λ/receipt; the job still records exactly as before.
+        governance = _govern_turn(kind, governed_text, model, node.name)
         # The label is decided off the AFTER sample (the fresh reading at job end).
         rec = self._commit(node.name, model, kind, tokens, wall_s,
                             sample_after, joules_measured,
                             useful_work=stored,
-                            rag_chunk_id=(rag_chunk_id if stored else None))
+                            rag_chunk_id=(rag_chunk_id if stored else None),
+                            governance=governance)
         return rec.to_dict()
 
     def _run_corpus_embed_batch(self, node: NodeCfg, meter_before: Optional[dict],
@@ -1264,6 +1404,7 @@ class OperatorDaemon:
                     "live RAG dense index. Useful work changes WHAT is computed, "
                     "never HOW joules are measured — the MEASURED gate is unchanged."
                 ),
+                "governed_compute": _governed_compute_summary(self._last_records),
                 "recent_jobs": [_public_job(j) for j in self._last_records[-10:]],
                 "exporter": _JOULE_METER_PUBLIC,
                 "honesty": (