#!/usr/bin/env python3 """SHiRE Academy unsupported-request learning bridge. The bridge records safe design-learning requests separately from the protected 371-skill Academy state. It never executes arbitrary prompts, user code, shell commands, network research, certification writes, or Forge permission changes. """ from __future__ import annotations import fcntl import hashlib import json import os import re import subprocess import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any VERSION = "0.1.0-stage3c.3-learning-bridge" ARMOR = Path(os.environ.get("SHIRE_ARMOR_ROOT", "/home/shire3d/ARMOR")) APP = ARMOR / "apps/shire-academy" RUNTIME = APP / "runtime" STATE_PATH = RUNTIME / "learning_bridge_state.json" LOCK_PATH = RUNTIME / "learning_bridge.lock" EVIDENCE_ROOT = Path("/SHiREVault/SHiREAcademy/LearningBridge") CAPABILITY_PATH = RUNTIME / "learning_bridge_capabilities.json" PROTECTED = { "practice_state": ARMOR / "data/academy/design_mastery/runtime/practice_state.json", "autopilot_state": ARMOR / "data/academy/design_mastery/runtime/autopilot_state.json", "commercial_sprint": ARMOR / "data/academy/design_mastery/runtime/commercial_design_sprint.json", "academy_index": ARMOR / "data/academy/design_mastery/runtime/academy_index.sqlite3", } CAPABILITY_PACKS: dict[str, dict[str, Any]] = { "gothic_dice_tower": { "title": "Gothic Cathedral Dice Tower", "description": "Hollow tower shell, alternating ramps, dice clearances, Gothic openings, collection tray and printer-fit validation.", "required_capabilities": [ "hollow shell", "alternating angled ramps", "minimum dice passage clearance", "pointed arch exit", "paired lancet windows", "rose-window recess", "decorative vertical ribs", "integrated collection tray", "printer-aware validation", ], "status": "validation_required", "builder_family": "gothic_dice_tower", "production_approved": False, "forge_allowed": False, }, "organic_sculpture": { "title": "Organic Sculpture and Skull Construction", "description": "Organic sculptural forms require a separate validated modelling route; Fast Builder must not fake them with a plaque.", "required_capabilities": [ "organic primary forms", "anatomical proportion study", "surface continuity", "printable hollowing", "support-aware orientation", "originality and provenance review", ], "status": "curriculum_required", "builder_family": None, "production_approved": False, "forge_allowed": False, }, "articulated_model": { "title": "Articulated Model Design", "description": "Joint clearances, trapped-pin logic and multi-part motion require dedicated deterministic validation.", "required_capabilities": [ "joint architecture", "clearance matrices", "trapped-pin geometry", "print-in-place validation", "motion interference checks", ], "status": "curriculum_required", "builder_family": None, "production_approved": False, "forge_allowed": False, }, "complex_mechanical_assembly": { "title": "Complex Mechanical Assembly", "description": "Mechanisms and assemblies require part relationships, tolerances and interference validation beyond starter templates.", "required_capabilities": [ "assembly decomposition", "mating references", "tolerance stack-up", "interference checks", "printer-aware part splitting", ], "status": "curriculum_required", "builder_family": None, "production_approved": False, "forge_allowed": False, }, "unclassified_design": { "title": "Unclassified Design Family", "description": "Academy must classify the design intent and create a validated template before Fast Builder can generate it.", "required_capabilities": [ "design-family classification", "requirements extraction", "parametric architecture", "deterministic validation", "printer-fit planning", ], "status": "classification_required", "builder_family": None, "production_approved": False, "forge_allowed": False, }, } def utc_now() -> str: return datetime.now(timezone.utc).isoformat() def atomic_json(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_name(path.name + f".tmp.{os.getpid()}") temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") os.replace(temporary, path) def load_json(path: Path, default: dict[str, Any] | None = None) -> dict[str, Any]: try: payload = json.loads(path.read_text(encoding="utf-8")) return payload if isinstance(payload, dict) else dict(default or {}) except (OSError, json.JSONDecodeError): return dict(default or {}) def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def protected_hashes() -> dict[str, str]: return {name: sha256(path) for name, path in PROTECTED.items() if path.is_file()} def safe_text(value: Any, maximum: int) -> str: text = re.sub(r"[\x00-\x1f\x7f]", " ", str(value or "")).strip() text = re.sub(r"\s+", " ", text) return text[:maximum] def safe_stem(value: str, fallback: str) -> str: stem = re.sub(r"[^A-Za-z0-9_-]+", "_", value).strip("_-")[:64] return stem or fallback def verify_network_root() -> dict[str, Any]: EVIDENCE_ROOT.mkdir(parents=True, exist_ok=True) result = subprocess.run( ["findmnt", "-T", str(EVIDENCE_ROOT), "-n", "-o", "SOURCE,FSTYPE"], text=True, capture_output=True, timeout=10, check=False, ) if result.returncode != 0 or not result.stdout.strip(): raise RuntimeError("SHiREVault Learning Bridge root is not mounted") parts = result.stdout.strip().splitlines()[-1].split(maxsplit=1) source = parts[0] fstype = parts[1] if len(parts) > 1 else "" if fstype not in {"cifs", "nfs", "nfs4", "smb3", "fuse.sshfs"}: raise RuntimeError(f"Learning Bridge root is not network-backed: {fstype or 'unknown'}") probe = EVIDENCE_ROOT / f".learning-bridge-probe-{os.getpid()}" probe.write_text(utc_now() + "\n", encoding="utf-8") if not probe.is_file() or probe.stat().st_size == 0: raise RuntimeError("Learning Bridge evidence write probe failed") probe.unlink(missing_ok=True) return {"source": source, "filesystem": fstype, "network_verified": True, "writable": True} def default_state() -> dict[str, Any]: return { "version": VERSION, "mode": "unsupported_request_learning_bridge", "requests": [], "request_count": 0, "latest_request": None, "arbitrary_prompt_execution_allowed": False, "external_network_research_allowed": False, "protected_academy_state_write_allowed": False, "counted_attempts_allowed": False, "certification_write_allowed": False, "forge_permission_write_allowed": False, "created_at": utc_now(), "updated_at": utc_now(), } def capabilities() -> dict[str, Any]: payload = {"version": VERSION, "packs": CAPABILITY_PACKS, "updated_at": utc_now()} stored = load_json(CAPABILITY_PATH) stored_packs = stored.get("packs") if isinstance(stored.get("packs"), dict) else {} merged: dict[str, Any] = {} for key, source in CAPABILITY_PACKS.items(): merged[key] = {**source, **(stored_packs.get(key) or {})} payload["packs"] = merged return payload def mark_validated(family: str, evidence: list[str], matrix_passes: int) -> dict[str, Any]: if family not in CAPABILITY_PACKS: raise ValueError("Unknown capability pack") payload = capabilities() pack = dict(payload["packs"][family]) pack.update({ "status": "validated_draft_template", "validated_at": utc_now(), "validation_evidence": list(evidence), "validation_matrix_passes": int(matrix_passes), "production_approved": False, "forge_allowed": False, }) payload["packs"][family] = pack payload["updated_at"] = utc_now() atomic_json(CAPABILITY_PATH, payload) return pack def classify_prompt(prompt: str) -> tuple[str, str]: lower = prompt.lower() if any(token in lower for token in ("dice tower", "dice-tower", "dice chute", "dice tumbler tower")): return "gothic_dice_tower", "The primary request is a dice tower, not a component tray." if any(token in lower for token in ("skull", "skeleton head", "cranium", "organic statue", "creature bust", "human head")): return "organic_sculpture", "The request requires organic sculptural modelling beyond trusted parametric starter templates." if any(token in lower for token in ("articulated", "print-in-place joint", "flexi toy", "hinged creature", "ball joint")): return "articulated_model", "The request requires validated articulation and clearance logic." if any(token in lower for token in ("gearbox", "gear train", "mechanism", "moving assembly", "kinematic assembly")): return "complex_mechanical_assembly", "The request requires a validated assembly and interference-checking route." return "unclassified_design", "The design family is not yet represented by a trusted Fast Builder template." def queue_request( *, prompt: str, project_name: str, output_format: str, printer_profile: str, material: str, family: str | None = None, reason: str | None = None, system_test: bool = False, ) -> dict[str, Any]: prompt_clean = safe_text(prompt, 1600) if len(prompt_clean) < 8: raise ValueError("Learning Bridge requires a meaningful design description") project_clean = safe_text(project_name, 80) or "SHiRE Learning Request" family_id, inferred_reason = classify_prompt(prompt_clean) family_id = family or family_id if family_id not in CAPABILITY_PACKS: family_id = "unclassified_design" pack = capabilities()["packs"][family_id] network = verify_network_root() before = protected_hashes() LOCK_PATH.parent.mkdir(parents=True, exist_ok=True) with LOCK_PATH.open("a+") as handle: fcntl.flock(handle.fileno(), fcntl.LOCK_EX) state = default_state() state.update(load_json(STATE_PATH)) request_id = "LB-" + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + "-" + uuid.uuid4().hex[:6].upper() request_dir = EVIDENCE_ROOT / ("SystemTests" if system_test else "Requests") / request_id request_dir.mkdir(parents=True, exist_ok=False) request = { "schema": "shire.learning_bridge.request.v1", "request_id": request_id, "project_name": project_clean, "file_stem": safe_stem(project_clean, "SHiRE_Learning_Request"), "prompt": prompt_clean, "family_id": family_id, "family_title": pack["title"], "reason": safe_text(reason or inferred_reason, 500), "status": "validated_template_available" if pack.get("status") == "validated_draft_template" else "queued_for_curriculum_design", "required_capabilities": list(pack.get("required_capabilities") or []), "requested_output_format": safe_text(output_format, 16), "printer_profile": safe_text(printer_profile, 64), "material": safe_text(material, 16), "arbitrary_prompt_execution_allowed": False, "external_network_research_allowed": False, "counted_attempt": False, "certification_write_allowed": False, "forge_permission_write_allowed": False, "production_release": False, "created_at": utc_now(), "system_test": bool(system_test), "evidence": str(request_dir), } plan = { "schema": "shire.learning_bridge.plan.v1", "request_id": request_id, "family_id": family_id, "objective": f"Create and validate a deterministic non-load-bearing draft template for {pack['title']}.", "required_capabilities": request["required_capabilities"], "gates": [ "deterministic parameter parsing", "one valid solid or explicitly validated printable assembly", "printer-fit check", "safe wall and clearance checks", "STEP/STL evidence matrix", "Ray preview before any production decision", ], "automatic_forge_unlock": False, "automatic_certification": False, "production_approval": False, "created_at": utc_now(), } atomic_json(request_dir / "request.json", request) atomic_json(request_dir / "learning-plan.json", plan) (request_dir / "README.txt").write_text( "SHiRE Academy Learning Bridge request\n\n" "This request records a missing or advanced design capability. It does not execute the prompt, certify a skill, unlock Forge, or approve production.\n", encoding="utf-8", ) after = protected_hashes() if before != after: raise RuntimeError("Protected Academy state changed while recording a Learning Bridge request") item = { "request_id": request_id, "project_name": project_clean, "family_id": family_id, "family_title": pack["title"], "status": request["status"], "reason": request["reason"], "required_capabilities": request["required_capabilities"], "created_at": request["created_at"], "evidence": str(request_dir), "protected_academy_state_unchanged": True, } if not system_test: requests = list(state.get("requests") or []) requests.append(item) state.update({ "version": VERSION, "mode": "unsupported_request_learning_bridge", "requests": requests[-100:], "request_count": int(state.get("request_count") or 0) + 1, "latest_request": item, "network": network, "updated_at": utc_now(), "arbitrary_prompt_execution_allowed": False, "external_network_research_allowed": False, "protected_academy_state_write_allowed": False, "counted_attempts_allowed": False, "certification_write_allowed": False, "forge_permission_write_allowed": False, }) atomic_json(STATE_PATH, state) return item def status() -> dict[str, Any]: payload = default_state() payload.update(load_json(STATE_PATH)) payload["version"] = VERSION payload["mode"] = "unsupported_request_learning_bridge" payload["capabilities"] = capabilities()["packs"] payload["evidence_root"] = str(EVIDENCE_ROOT) return payload if __name__ == "__main__": print(json.dumps(status(), indent=2, sort_keys=True))