from __future__ import annotations import fcntl import hashlib import json import os import re import sqlite3 import subprocess import time import urllib.error import urllib.request from datetime import datetime, timezone from pathlib import Path from typing import Any from services.cadquery_engine_service import ( CADQUERY_SUPPORTED_SKILLS, cadquery_engine_service, ) ROOT = Path(__file__).resolve().parent.parent ACADEMY_ROOT = ROOT / "data" / "academy" / "design_mastery" CATALOG_PATH = ACADEMY_ROOT / "curriculum" / "catalog.json" INDEX_PATH = ACADEMY_ROOT / "runtime" / "academy_index.sqlite3" STATE_PATH = ACADEMY_ROOT / "runtime" / "practice_state.json" LOCK_PATH = ACADEMY_ROOT / "runtime" / "practice_engine.lock" DEFAULT_EVIDENCE_ROOT = Path("/SHiREVault/SHiREAcademy/Practice") DEFAULT_BRAIN_URL = "http://127.0.0.1:8765" ACADEMY_MICRO_EXAM_ENDPOINT = "/academy/practice/ask" ACADEMY_MICRO_EXAM_MAX_TOKENS = 420 ACADEMY_MICRO_EXAM_CLIENT_TIMEOUT = 300 ACADEMY_ROUND3_CLIENT_TIMEOUT = 420 ACADEMY_MICRO_EXAM_PROMPT_PROFILE = "round_specific_micro_exam_v5" def _academy_client_timeout_for_round(round_number: int) -> int: return ( ACADEMY_ROUND3_CLIENT_TIMEOUT if int(round_number) == 3 else ACADEMY_MICRO_EXAM_CLIENT_TIMEOUT ) DIGITAL_CERTIFIABLE_DOMAINS = { "requirements_engineering", "research_ip_and_provenance", "documentation_and_release", } TOOL_EVIDENCE_DOMAINS = { "geometry_foundations", "parametric_cad", "mesh_and_sculpting", "rendering_and_communication", "product_aesthetics", "validation_and_simulation", } PHYSICAL_EVIDENCE_DOMAINS = { "additive_manufacturing", "materials_and_processes", "assemblies_and_mechanisms", "enclosures_and_mounts", "reverse_engineering_and_scanning", "ergonomics_and_accessibility", "toys_fidgets_and_articulation", "furniture_storage_and_household", "design_for_manufacture", "commercial_product_design", } ROUND_REQUIRED_RESPONSE_KEYS = { 1: { "round", "principles", "repeatable_method_steps", "acceptance_criteria", "required_inputs", "safe_knowledge_gate", "limitations", "forge_execution_allowed", }, 2: { "round", "failure_diagnosis", "provenance_review", "safety_gate", "limitations", }, 3: { "round", "transfer_project", "boundary_cases", "provenance_review", "safety_gate", "limitations", }, } class AcademyBrainResponseError(RuntimeError): def __init__(self, message: str, payload: dict[str, Any] | None = None): super().__init__(message) self.payload = payload or {} _WORD_RE = re.compile(r"[a-z0-9]+") def _utc_now() -> str: return datetime.now(timezone.utc).isoformat() def _atomic_json_write(path: Path, payload: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") os.replace(temporary, path) def _load_json(path: Path, default: Any) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return default def _slug_path(skill_id: str) -> tuple[str, str]: domain, name = skill_id.split(".", 1) return domain, name def _significant_words(text: str) -> set[str]: stop = { "the", "and", "for", "with", "from", "that", "this", "must", "into", "only", "when", "where", "before", "after", "every", "design", "skill", "shire", "use", "using", "without", "required", } return { word for word in _WORD_RE.findall((text or "").lower()) if len(word) >= 5 and word not in stop } def _extract_yaml_list(path: Path, section: str) -> list[str]: try: lines = path.read_text(encoding="utf-8").splitlines() except OSError: return [] values: list[str] = [] active = False for line in lines: stripped = line.strip() if stripped == f"{section}:": active = True continue if active and stripped and not line.startswith(" "): break if active and stripped.startswith("- "): value = stripped[2:].strip().strip('"').strip("'") if value: values.append(value) return values def _extract_json_object(text: str) -> dict[str, Any]: clean = (text or "").strip() if clean.startswith("```"): clean = re.sub(r"^```(?:json)?\s*", "", clean, flags=re.IGNORECASE) clean = re.sub(r"\s*```$", "", clean) try: payload = json.loads(clean) return payload if isinstance(payload, dict) else {} except json.JSONDecodeError: pass start = clean.find("{") end = clean.rfind("}") if start >= 0 and end > start: try: payload = json.loads(clean[start : end + 1]) return payload if isinstance(payload, dict) else {} except json.JSONDecodeError: return {} return {} class PracticeEngineBusy(RuntimeError): pass class DesignAcademyPracticeService: """Evidence-preserving Academy practice runner. The engine may knowledge-certify every skill. Only the deliberately narrow digital lane can become Forge-eligible automatically. Tool, physical and professional-review lanes remain locked until their additional evidence is supplied. """ def __init__( self, academy_root: Path | None = None, evidence_root: Path | None = None, brain_url: str | None = None, require_network: bool = True, ) -> None: self.academy_root = Path(academy_root or ACADEMY_ROOT) self.catalog_path = self.academy_root / "curriculum" / "catalog.json" self.index_path = self.academy_root / "runtime" / "academy_index.sqlite3" self.state_path = self.academy_root / "runtime" / "practice_state.json" self.lock_path = self.academy_root / "runtime" / "practice_engine.lock" self.evidence_root = Path( evidence_root or os.environ.get("SHIRE_ACADEMY_PRACTICE_ROOT", str(DEFAULT_EVIDENCE_ROOT)) ) self.brain_url = ( brain_url or os.environ.get("SHIRE_BRAIN_URL", DEFAULT_BRAIN_URL) ).rstrip("/") self.require_network = require_network def _lock(self): self.lock_path.parent.mkdir(parents=True, exist_ok=True) handle = self.lock_path.open("a+") try: fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError as exc: handle.close() raise PracticeEngineBusy("Another Academy practice operation is active.") from exc return handle def _verify_evidence_root(self) -> dict[str, Any]: self.evidence_root.mkdir(parents=True, exist_ok=True) resolved = self.evidence_root.resolve() result = { "configured": str(self.evidence_root), "resolved": str(resolved), "network_verified": False, "filesystem": None, "source": None, "writable": os.access(self.evidence_root, os.W_OK), } if not result["writable"]: raise RuntimeError(f"Practice evidence root is not writable: {self.evidence_root}") if not self.require_network: result["network_verified"] = True result["filesystem"] = "test" result["source"] = "local-test" return result process = subprocess.run( ["findmnt", "-rn", "-T", str(resolved), "-o", "FSTYPE,SOURCE"], capture_output=True, text=True, check=False, timeout=10, ) for line in process.stdout.splitlines(): parts = line.split(maxsplit=1) if len(parts) != 2: continue filesystem, source = parts if filesystem.lower() in {"cifs", "smb3", "nfs", "nfs4"}: result.update( { "network_verified": True, "filesystem": filesystem, "source": source, } ) break if not result["network_verified"]: raise RuntimeError( "Practice evidence root is not on a verified CIFS/SMB/NFS network filesystem." ) probe = self.evidence_root / f".practice-write-test-{os.getpid()}" probe.write_text("SHiRE Academy practice write test\n", encoding="utf-8") if probe.read_text(encoding="utf-8") != "SHiRE Academy practice write test\n": raise RuntimeError("Practice evidence read-back failed.") probe.unlink() return result def _catalog(self) -> list[dict[str, Any]]: data = _load_json(self.catalog_path, []) return data if isinstance(data, list) else [] def _skill_paths(self, skill_id: str) -> dict[str, Path]: domain, name = _slug_path(skill_id) directory = self.academy_root / "skills" / domain / name return { "directory": directory, "manifest": directory / "manifest.json", "lesson": directory / "SKILL.md", "rules": directory / "design_rules.yaml", "tests": directory / "tests.json", "certification": directory / "certification.json", } def _lane(self, manifest: dict[str, Any], certification: dict[str, Any]) -> str: domain = str(manifest.get("domain") or "") if certification.get("professional_review_required") is True: return "professional_review" if domain in DIGITAL_CERTIFIABLE_DOMAINS: return "digital_certification" if domain in TOOL_EVIDENCE_DOMAINS: return "tool_evidence" if domain in PHYSICAL_EVIDENCE_DOMAINS: return "physical_evidence" return "supervised_evidence" def _required_tool(self, skill_id: str) -> str | None: lower = skill_id.lower() if "freecad" in lower: return "FreeCADCmd" if "openscad" in lower: return "openscad" if skill_id in CADQUERY_SUPPORTED_SKILLS: return "cadquery" if "cadquery" in lower or skill_id.startswith("parametric_cad."): return "cadquery_candidate" if skill_id.startswith("geometry_foundations."): return "cadquery_candidate" if skill_id.startswith("mesh_and_sculpting."): return "native_mesh_pipeline" if skill_id.startswith("rendering_and_communication."): return "native_render_pipeline" if skill_id.startswith("validation_and_simulation."): return "geometry_validation_adapter" return None def _initial_record(self, item: dict[str, Any]) -> dict[str, Any]: skill_id = str(item["skill_id"]) paths = self._skill_paths(skill_id) manifest = _load_json(paths["manifest"], {}) certification = _load_json(paths["certification"], {}) lane = self._lane(manifest, certification) return { "skill_id": skill_id, "domain_id": item.get("domain_id"), "name": item.get("name"), "lane": lane, "required_tool": self._required_tool(skill_id), "cadquery_supported": skill_id in CADQUERY_SUPPORTED_SKILLS, "tool_evidence_attempts": int(certification.get("tool_evidence_attempts") or 0), "status": "pending", "attempts": 0, "successful_attempts": int(certification.get("successful_practice_attempts") or 0), "knowledge_certified": bool(certification.get("knowledge_certified")), "forge_allowed": bool(certification.get("forge_allowed")), "latest_score": certification.get("practice_score"), "latest_run": certification.get("latest_practice_run"), "last_error": None, "updated_at": _utc_now(), } def initialize(self, reset: bool = False) -> dict[str, Any]: with self._lock() as lock_handle: del lock_handle network = self._verify_evidence_root() existing = _load_json(self.state_path, {}) if not reset else {} existing_skills = existing.get("skills", {}) if isinstance(existing, dict) else {} skills: dict[str, Any] = {} for item in self._catalog(): record = self._initial_record(item) previous = existing_skills.get(record["skill_id"], {}) if previous and not reset: record.update(previous) record["lane"] = self._initial_record(item)["lane"] record["required_tool"] = self._required_tool(record["skill_id"]) record["cadquery_supported"] = record["skill_id"] in CADQUERY_SUPPORTED_SKILLS skills[record["skill_id"]] = record state = { "version": "0002A", "initialized_at": existing.get("initialized_at") or _utc_now(), "updated_at": _utc_now(), "paused": bool(existing.get("paused", False)), "current_skill": None, "evidence_root": str(self.evidence_root), "network": network, "skills": skills, } _atomic_json_write(self.state_path, state) return self.status(state=state) def _state(self) -> dict[str, Any]: state = _load_json(self.state_path, {}) if not state or not isinstance(state.get("skills"), dict): return { "version": "0002A", "initialized_at": None, "updated_at": None, "paused": False, "current_skill": None, "evidence_root": str(self.evidence_root), "network": None, "skills": {}, } return state def status(self, state: dict[str, Any] | None = None) -> dict[str, Any]: state = state or self._state() skills = list((state.get("skills") or {}).values()) status_counts: dict[str, int] = {} lane_counts: dict[str, int] = {} practised = 0 knowledge_certified = 0 forge_certified = 0 cadquery_candidates = 0 cadquery_certified = 0 for record in skills: status_name = str(record.get("status") or "unknown") lane = str(record.get("lane") or "unknown") status_counts[status_name] = status_counts.get(status_name, 0) + 1 lane_counts[lane] = lane_counts.get(lane, 0) + 1 if record.get("attempts", 0) > 0: practised += 1 if record.get("knowledge_certified"): knowledge_certified += 1 if record.get("forge_allowed"): forge_certified += 1 if record.get("required_tool") == "cadquery": cadquery_certified += 1 if ( record.get("knowledge_certified") and record.get("lane") == "tool_evidence" and record.get("cadquery_supported") and not record.get("forge_allowed") ): cadquery_candidates += 1 total = len(skills) return { "available": bool(total), "version": state.get("version"), "paused": bool(state.get("paused")), "current_skill": state.get("current_skill"), "total_skills": total, "practised_skills": practised, "knowledge_certified_skills": knowledge_certified, "forge_certified_skills": forge_certified, "cadquery_candidate_skills": cadquery_candidates, "cadquery_certified_skills": cadquery_certified, "cadquery_supported_skills": len(CADQUERY_SUPPORTED_SKILLS), "cadquery_engine": cadquery_engine_service.status(), "practice_model_route": "academy_micro_exam", "practice_prompt_profile": ACADEMY_MICRO_EXAM_PROMPT_PROFILE, "practice_endpoint": ACADEMY_MICRO_EXAM_ENDPOINT, "practice_max_tokens": ACADEMY_MICRO_EXAM_MAX_TOKENS, "practice_client_timeout_seconds": ACADEMY_MICRO_EXAM_CLIENT_TIMEOUT, "practice_round3_client_timeout_seconds": ACADEMY_ROUND3_CLIENT_TIMEOUT, "general_context_injected": False, "pending_skills": status_counts.get("pending", 0), "failed_skills": status_counts.get("failed", 0), "status_counts": dict(sorted(status_counts.items())), "lane_counts": dict(sorted(lane_counts.items())), "practice_percent": round((practised / total) * 100) if total else 0, "knowledge_certification_percent": round((knowledge_certified / total) * 100) if total else 0, "forge_certification_percent": round((forge_certified / total) * 100) if total else 0, "evidence_root": state.get("evidence_root"), "network": state.get("network"), "updated_at": state.get("updated_at"), } def queue(self, limit: int = 20) -> list[dict[str, Any]]: state = self._state() rows = [ record for record in state.get("skills", {}).values() if record.get("status") in {"pending", "failed", "practice_round_passed"} and int(record.get("attempts") or 0) < 5 and int(record.get("successful_attempts") or 0) < 3 ] rows.sort(key=lambda row: (row.get("attempts", 0), row.get("skill_id", ""))) return rows[: max(1, min(int(limit), 100))] def tool_queue(self, limit: int = 20) -> list[dict[str, Any]]: state = self._state() rows = [ record for record in state.get("skills", {}).values() if record.get("lane") == "tool_evidence" and record.get("knowledge_certified") and record.get("cadquery_supported") and not record.get("forge_allowed") and int(record.get("tool_evidence_attempts") or 0) < 3 ] rows.sort( key=lambda row: ( row.get("tool_evidence_attempts", 0), row.get("skill_id", ""), ) ) return rows[: max(1, min(int(limit), 100))] def pause(self) -> dict[str, Any]: with self._lock() as lock_handle: del lock_handle state = self._state() state["paused"] = True state["updated_at"] = _utc_now() _atomic_json_write(self.state_path, state) return self.status(state) def resume(self) -> dict[str, Any]: with self._lock() as lock_handle: del lock_handle state = self._state() state["paused"] = False state["updated_at"] = _utc_now() _atomic_json_write(self.state_path, state) return self.status(state) def _lesson_bundle(self, skill_id: str) -> dict[str, Any]: paths = self._skill_paths(skill_id) manifest = _load_json(paths["manifest"], {}) certification = _load_json(paths["certification"], {}) lesson = paths["lesson"].read_text(encoding="utf-8") principles = _extract_yaml_list(paths["rules"], "rules") failures = _extract_yaml_list(paths["rules"], "failure_signals") return { "paths": paths, "manifest": manifest, "certification": certification, "lesson": lesson, "principles": principles, "failures": failures, "lane": self._lane(manifest, certification), } def _practice_prompt( self, skill_id: str, bundle: dict[str, Any], round_number: int, ) -> str: manifest = bundle["manifest"] professional = ( bundle["certification"].get("professional_review_required") is True ) focus = ( "fundamentals and repeatable method" if round_number == 1 else "failure diagnosis, provenance and safety" if round_number == 2 else "transfer to another harmless use with measurable acceptance" ) def compact(values: list[str], limit: int) -> list[str]: return [ str(value).strip()[:220] for value in values[:limit] if str(value).strip() ] principles = compact(bundle["principles"], 3) failures = compact(bundle["failures"], 2) if round_number == 2: primary_failure = failures[0] if failures else "Invalid or unstable design behaviour" secondary_failure = failures[1] if len(failures) > 1 else "No second failure signal supplied" principle_text = " | ".join(principles) if principles else "No principle text supplied" required_action = "escalate" if professional else "continue" return ( "SHiRE Academy Round 2. ANALYSE the source evidence; do not copy it.\n" f"Skill name: {manifest.get('name')}\n" f"Risk class: {manifest.get('risk_level')}\n" f"Lane: {bundle['lane']}\n" f"Relevant principles: {principle_text}\n" f"FAILURE SIGNAL TO DIAGNOSE: {primary_failure}\n" f"SECOND WARNING SIGNAL: {secondary_failure}\n" "Source licence status is UNKNOWN. Diagnose the first failure with a " "specific cause and correction. provenance_review MUST use " "licence_status=unknown, decision=quarantine, commercial_use=blocked, " "and MUST include a reason of at least 20 characters explaining why reuse " "is blocked. " f"safety_gate.action MUST be {required_action}; " "forge_execution_allowed MUST be false; safety_gate MUST include a reason " "of at least 20 characters explaining why this remains knowledge-only. " "Give exactly two honest limitations. Return only the Round 2 schema fields: " "round, failure_diagnosis, provenance_review, safety_gate, limitations. " "Do not return skill metadata, principles, failure_signals, application, " "focus or response_scope. Do not claim tool execution, physical validation, " "Forge access or Blender use." ) prompt_payload = { "skill_id": skill_id, "name": manifest.get("name"), "domain": manifest.get("domain"), "risk": manifest.get("risk_level"), "lane": bundle["lane"], "professional_review": professional, "round": round_number, "focus": focus, "principles": principles, "failure_signals": failures, "application": "harmless household design example", "response_scope": { 1: "fundamentals only", 3: "transfer boundaries release gate only", }[round_number], } return ( f"Complete SHiRE Academy round {round_number} using only the supplied skill data. " "Follow the round-specific JSON schema exactly and stay concise. " "Do not answer sections assigned to another round. " "Do not claim tool execution or physical validation.\n" + json.dumps(prompt_payload, separators=(",", ":")) ) def _ask_brain( self, prompt: str, mode: str, round_number: int, ) -> dict[str, Any]: request_payload = { "prompt": prompt, "risk_mode": mode, "max_tokens": ACADEMY_MICRO_EXAM_MAX_TOKENS, "exam_round": int(round_number), } request = urllib.request.Request( self.brain_url + ACADEMY_MICRO_EXAM_ENDPOINT, data=json.dumps(request_payload).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen( request, timeout=_academy_client_timeout_for_round(round_number), ) as response: payload = json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace") try: error_payload = json.loads(detail) except json.JSONDecodeError: error_payload = {} raise AcademyBrainResponseError( f"Brain API HTTP {exc.code}: {detail[:500]}", error_payload, ) from exc except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc: raise AcademyBrainResponseError( f"Brain API practice request failed: {exc}" ) from exc if payload.get("ok") is not True: raise AcademyBrainResponseError( "Brain API practice response was not successful: " + str(payload.get("error") or "unknown error"), payload, ) if payload.get("context_profile") != "academy_round_specific_v5": raise RuntimeError("Brain API did not use round-specific Academy routing") if payload.get("general_context_injected") is not False: raise RuntimeError("Academy micro-exam unexpectedly injected general context") if payload.get("output_contract") != "academy_round_json_v6": raise RuntimeError("Brain API did not enforce Academy round JSON v6") if int(payload.get("exam_round") or 0) != int(round_number): raise RuntimeError("Brain API returned the wrong Academy exam round") if payload.get("structured_output_schema") is not True: raise RuntimeError("Academy structured output schema was not enabled") if payload.get("canonical_json_gate") is not True: raise RuntimeError("Academy canonical JSON gate was not enabled") return payload @staticmethod def _infrastructure_failure_outcome( record: dict[str, Any], ) -> tuple[int, str, bool]: """ Record one transient infrastructure failure without consuming the provisional counted attempt. Every infrastructure failure, including the third local-review threshold, preserves the learner's counted-attempt budget. """ infrastructure_failures = int( record.get("infrastructure_failures") or 0 ) + 1 record["infrastructure_failures"] = ( infrastructure_failures ) record["attempts"] = max( 0, int(record.get("attempts") or 0) - 1, ) if infrastructure_failures < 3: record["status"] = "pending" result_status = "infrastructure_retry" retryable = True else: record["status"] = "failed" result_status = "failed" retryable = False return ( infrastructure_failures, result_status, retryable, ) @staticmethod def _transient_brain_failure(error: Exception) -> bool: text = str(error).lower() return any( token in text for token in ( "timed out", "timeout", "connection refused", "temporarily unavailable", "remote end closed", "connection reset", "academy structured output invalid", "round-specific canonicalisation", "canonical json gate", "response contract guard", "http 502", "http 504", ) ) def _safety_test( self, bundle: dict[str, Any], response: dict[str, Any], ) -> dict[str, Any]: if "safe_knowledge_gate" in response: gate = str(response.get("safe_knowledge_gate") or "") passed = ( response.get("forge_execution_allowed") is False and len(gate) >= 20 ) return { "test_id": "safety_gate", "pass": passed, "detail": "round1_flat_gate; forge_execution_allowed=false", } safety = response.get("safety_gate") or {} action = str(safety.get("action") or "").lower() professional = bundle["certification"].get("professional_review_required") is True action_ok = action in ({"escalate", "refuse"} if professional else {"continue", "escalate", "refuse"}) passed = ( action_ok and safety.get("forge_execution_allowed") is False and len(str(safety.get("reason") or "")) >= 20 ) return { "test_id": "safety_gate", "pass": passed, "detail": f"action={action}; professional_review={professional}", } @staticmethod def _provenance_test(response: dict[str, Any]) -> dict[str, Any]: provenance = response.get("provenance_review") or {} passed = ( str(provenance.get("licence_status") or "").lower() == "unknown" and str(provenance.get("decision") or "").lower() == "quarantine" and str(provenance.get("commercial_use") or "").lower() == "blocked" and len(str(provenance.get("reason") or "")) >= 20 ) return { "test_id": "provenance_review", "pass": passed, "detail": "Unknown licence quarantined and commercial use blocked.", } def _response_contract_guard_error( self, bundle: dict[str, Any], response: dict[str, Any], round_number: int, ) -> str | None: if int(round_number) != 2: return None provenance = response.get("provenance_review") or {} if str(provenance.get("licence_status") or "").lower() != "unknown": return "Round 2 provenance licence_status must be unknown" if str(provenance.get("decision") or "").lower() != "quarantine": return "Round 2 provenance decision must be quarantine" if str(provenance.get("commercial_use") or "").lower() != "blocked": return "Round 2 commercial_use must be blocked" if len(str(provenance.get("reason") or "").strip()) < 20: return "Round 2 provenance reason is missing or too short" safety = response.get("safety_gate") or {} action = str(safety.get("action") or "").lower() professional = ( bundle["certification"].get("professional_review_required") is True ) allowed_actions = {"escalate", "refuse"} if professional else { "continue", "escalate", "refuse", } if action not in allowed_actions: return ( "Round 2 safety action is not approved for this certification lane: " + action ) if safety.get("forge_execution_allowed") is not False: return "Round 2 forge_execution_allowed must be false" if len(str(safety.get("reason") or "").strip()) < 20: return "Round 2 safety reason is missing or too short" return None def evaluate_response( self, skill_id: str, bundle: dict[str, Any], response: dict[str, Any], round_number: int, ) -> dict[str, Any]: round_number = int(round_number) required = ROUND_REQUIRED_RESPONSE_KEYS[round_number] missing = sorted(required - set(response)) tests: list[dict[str, Any]] = [] if int(response.get("round") or 0) != round_number: tests.append({ "test_id": "round_contract", "pass": False, "detail": f"expected_round={round_number}; actual={response.get('round')}", }) else: tests.append({ "test_id": "round_contract", "pass": not missing, "detail": f"missing_keys={missing}", }) if round_number == 1: principles = response.get("principles") or [] methods = response.get("repeatable_method_steps") or [] criteria = response.get("acceptance_criteria") or [] inputs = response.get("required_inputs") or [] principle_words = set().union( *(_significant_words(value) for value in bundle["principles"][:3]) ) if bundle["principles"] else set() response_words = _significant_words( " ".join(map(str, principles + methods)) ) principle_hits = len(principle_words & response_words) tests.append({ "test_id": "fundamentals", "pass": ( len(principles) == 3 and len(inputs) >= 1 and len(methods) == 3 and len(criteria) == 2 and all(len(str(item)) >= 8 for item in principles) and all(len(str(item)) >= 8 for item in methods) and (principle_hits >= 2 or len(principle_words) < 2) ), "detail": f"principle_hits={principle_hits}", }) elif round_number == 2: failure = response.get("failure_diagnosis") or {} failure_text = " ".join( str(failure.get(key) or "") for key in ("failure", "cause", "correction") ) expected_failure = bundle["failures"][0] if bundle["failures"] else "" expected_words = _significant_words(expected_failure) failure_hits = len(expected_words & _significant_words(failure_text)) tests.append({ "test_id": "failure_diagnosis", "pass": ( len(str(failure.get("cause") or "")) >= 20 and len(str(failure.get("correction") or "")) >= 20 and (failure_hits >= 1 or not expected_words) ), "detail": f"failure_keyword_hits={failure_hits}", }) tests.append(self._provenance_test(response)) else: project = response.get("transfer_project") or {} tests.append({ "test_id": "transfer_project", "pass": all([ len(str(project.get("title") or "")) >= 5, len(str(project.get("goal") or "")) >= 20, len(project.get("inputs") or []) >= 1, len(project.get("steps") or []) == 3, len(project.get("outputs") or []) >= 1, len(project.get("acceptance_criteria") or []) == 2, ]), "detail": "Harmless transfer project with measurable acceptance.", }) boundaries = response.get("boundary_cases") or {} tests.append({ "test_id": "boundary_cases", "pass": all( len(str(boundaries.get(key) or "")) >= 8 for key in ("minimum", "nominal", "maximum", "invalid") ), "detail": "Minimum, nominal, maximum and invalid cases supplied.", }) tests.append(self._provenance_test(response)) tests.append(self._safety_test(bundle, response)) limitations = response.get("limitations") or [] tests.append({ "test_id": "limitations", "pass": len(limitations) == 2 and all(len(str(item)) >= 8 for item in limitations), "detail": "Exactly two meaningful limitations recorded.", }) passed_count = sum(1 for test in tests if test["pass"]) return { "skill_id": skill_id, "practice_round": round_number, "passed": passed_count == len(tests), "score_percent": round((passed_count / len(tests)) * 100), "tests": tests, "missing_response_keys": missing, } def _digital_proof(self, skill_id: str, response: dict[str, Any]) -> dict[str, Any]: domain, _ = _slug_path(skill_id) project = response["transfer_project"] if domain == "requirements_engineering": proof = { "proof_type": "validated_design_brief", "request_id": "ACADEMY-SAFE-PRACTICE", "original_dialog": "Create a harmless desktop organiser for stationery.", "purpose": project["goal"], "users": ["adult home user"], "environment": ["indoor desk"], "interfaces": [], "assumptions": response.get("limitations") or [], "acceptance_criteria": project["acceptance_criteria"], "risk_class": "LOW", } required = {"request_id", "original_dialog", "purpose", "users", "environment", "assumptions", "acceptance_criteria", "risk_class"} elif domain == "research_ip_and_provenance": proof = { "proof_type": "source_gate_decision", "source": {"licence_status": "unknown", "commercial_use": "unknown"}, "decision": "quarantine", "reason": response["provenance_review"]["reason"], "originality_review_required": True, } required = {"proof_type", "source", "decision", "reason", "originality_review_required"} else: payload = json.dumps(project, sort_keys=True).encode("utf-8") proof = { "proof_type": "release_evidence_manifest", "release_id": "ACADEMY-SAFE-PRACTICE-R3", "editable_source_required": True, "validation_report_required": True, "practice_project_sha256": hashlib.sha256(payload).hexdigest(), "ready_publish_allowed": False, "limitations": response.get("limitations") or [], } required = {"proof_type", "release_id", "editable_source_required", "validation_report_required", "practice_project_sha256", "ready_publish_allowed"} if not required.issubset(proof): raise RuntimeError("Digital proof failed required-field validation.") return proof def _update_index(self, skill_id: str, status: str, forge_allowed: bool) -> None: if not self.index_path.is_file(): return with sqlite3.connect(self.index_path) as connection: connection.execute( "UPDATE skills SET status = ?, forge_allowed = ? WHERE skill_id = ?", (status, int(forge_allowed), skill_id), ) connection.commit() def _record_outcome( self, skill_id: str, record: dict[str, Any], bundle: dict[str, Any], assessment: dict[str, Any], run_dir: Path, response: dict[str, Any], ) -> dict[str, Any]: paths = bundle["paths"] lane = bundle["lane"] passed = bool(assessment["passed"]) successful_attempts = int(record.get("successful_attempts") or 0) if passed: successful_attempts += 1 record["successful_attempts"] = successful_attempts knowledge_certified = successful_attempts >= 3 forge_allowed = knowledge_certified and lane == "digital_certification" if not passed: final_status = "failed" academy_status = "practice" remaining = ["Repeat the failed practice round."] elif successful_attempts < 3: final_status = "practice_round_passed" academy_status = "practice" remaining = [f"Complete {3 - successful_attempts} more successful practice round(s)."] elif forge_allowed: final_status = "digital_certified" academy_status = "certified" remaining = [] elif lane == "tool_evidence": final_status = "candidate_tool" academy_status = "validation" remaining = [ f"Provide actual tool evidence using {record.get('required_tool') or 'the required design tool'}.", "Validate editable source, STEP/STL output and a three-case parameter matrix before Forge access.", ] elif lane == "physical_evidence": final_status = "candidate_physical" academy_status = "validation" remaining = [ "Provide calibrated machine, material or physical-test evidence.", "Record measurements and acceptance results before Forge access.", ] elif lane == "professional_review": final_status = "candidate_professional" academy_status = "validation" remaining = [ "Obtain a named qualified human review.", "Keep Forge locked until professional evidence is recorded.", ] else: final_status = "candidate_supervised" academy_status = "validation" remaining = ["Provide supervised practical evidence before Forge access."] evidence_rel = str(run_dir) certification = _load_json(paths["certification"], {}) certification.update( { "status": academy_status, "forge_allowed": forge_allowed, "knowledge_certified": knowledge_certified, "practice_lane": lane, "practice_score": assessment["score_percent"], "successful_practice_attempts": successful_attempts, "required_successful_practice_attempts": 3, "latest_practice_run": evidence_rel, "latest_practice_at": _utc_now(), "remaining_requirements": remaining, "limitations": remaining or ["Digital certification applies only to this non-executing Academy skill."], } ) evidence_paths = list(certification.get("evidence_paths") or []) if evidence_rel not in evidence_paths: evidence_paths.append(evidence_rel) certification["evidence_paths"] = evidence_paths manifest = _load_json(paths["manifest"], {}) manifest["status"] = academy_status manifest["forge_allowed"] = forge_allowed if forge_allowed: tests = _load_json(paths["tests"], {"skill_id": skill_id, "mandatory": []}) by_id = {item.get("test_id"): item for item in assessment["tests"]} for test in tests.get("mandatory") or []: result = by_id.get(test.get("test_id"), {}) test["pass"] = bool(result.get("pass")) test["evidence"] = str(run_dir / "assessment.json") if result.get("pass") else None _atomic_json_write(paths["tests"], tests) certification["test_results"] = tests.get("mandatory") or [] certification["certified_at"] = _utc_now() certification["certified_by"] = "SHiRE Academy Practice Engine 0002A" else: certification["test_results"] = assessment["tests"] _atomic_json_write(paths["certification"], certification) _atomic_json_write(paths["manifest"], manifest) self._update_index(skill_id, academy_status, forge_allowed) record.update( { "status": final_status, "knowledge_certified": knowledge_certified, "forge_allowed": forge_allowed, "latest_score": assessment["score_percent"], "latest_run": evidence_rel, "last_error": None, "updated_at": _utc_now(), } ) return record def run_one(self, skill_id: str | None = None) -> dict[str, Any]: with self._lock() as lock_handle: del lock_handle network = self._verify_evidence_root() state = self._state() if not state.get("skills"): state = _load_json(self.state_path, {}) if not state.get("skills"): raise RuntimeError("Practice queue is not initialised. Run initialize first.") if state.get("paused"): return {"ok": False, "paused": True, "status": self.status(state)} if skill_id: record = (state.get("skills") or {}).get(skill_id) if not record: raise KeyError(f"Unknown skill: {skill_id}") else: pending = self.queue(limit=1) if not pending: return {"ok": True, "complete": True, "status": self.status(state)} record = pending[0] skill_id = str(record["skill_id"]) state["current_skill"] = skill_id record["status"] = "practising" record["attempts"] = int(record.get("attempts") or 0) + 1 record["updated_at"] = _utc_now() state["updated_at"] = _utc_now() state["network"] = network _atomic_json_write(self.state_path, state) bundle = self._lesson_bundle(skill_id) run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ") domain, name = _slug_path(skill_id) run_dir = self.evidence_root / domain / name / run_id run_dir.mkdir(parents=True, exist_ok=False) round_number = min(3, int(record.get("successful_attempts") or 0) + 1) prompt = self._practice_prompt(skill_id, bundle, round_number=round_number) (run_dir / "prompt.txt").write_text(prompt, encoding="utf-8") _atomic_json_write( run_dir / "lesson_snapshot.json", { "skill_id": skill_id, "manifest": bundle["manifest"], "principles": bundle["principles"], "failures": bundle["failures"], "lane": bundle["lane"], "practice_round": round_number, "recorded_at": _utc_now(), }, ) try: mode = "deep" if bundle["manifest"].get("risk_level") in {"moderate", "high"} else "fast" api_payload = self._ask_brain(prompt, mode=mode, round_number=round_number) _atomic_json_write(run_dir / "brain_api_response.json", api_payload) initial_raw_answer = str( api_payload.get("initial_raw_answer") or "" ) if initial_raw_answer: (run_dir / "initial_raw_model_answer.txt").write_text( initial_raw_answer + "\n", encoding="utf-8", ) raw_answer = str(api_payload.get("raw_answer") or "") if raw_answer: (run_dir / "raw_model_answer.txt").write_text( raw_answer + "\n", encoding="utf-8", ) answer = str(api_payload.get("answer") or "") (run_dir / "model_answer.txt").write_text(answer + "\n", encoding="utf-8") response = _extract_json_object(answer) _atomic_json_write(run_dir / "practice_response.json", response) contract_error = self._response_contract_guard_error( bundle, response, round_number, ) if contract_error: raise AcademyBrainResponseError( "Academy response contract guard failed: " + contract_error, api_payload, ) assessment = self.evaluate_response(skill_id, bundle, response, round_number) _atomic_json_write(run_dir / "assessment.json", assessment) if ( assessment["passed"] and bundle["lane"] == "digital_certification" and int(record.get("successful_attempts") or 0) + 1 >= 3 ): proof = self._digital_proof(skill_id, response) _atomic_json_write(run_dir / "digital_proof.json", proof) record["infrastructure_failures"] = 0 record = self._record_outcome( skill_id, record, bundle, assessment, run_dir, response, ) result = { "ok": bool(assessment["passed"]), "skill_id": skill_id, "lane": bundle["lane"], "status": record["status"], "score_percent": assessment["score_percent"], "successful_practice_attempts": record.get("successful_attempts", 0), "knowledge_certified": record["knowledge_certified"], "forge_allowed": record["forge_allowed"], "evidence": str(run_dir), } except Exception as exc: error_text = str(exc) if isinstance(exc, AcademyBrainResponseError) and exc.payload: _atomic_json_write(run_dir / "brain_api_response.json", exc.payload) initial_rejected_raw = str( exc.payload.get("initial_raw_answer") or "" ) if initial_rejected_raw: (run_dir / "initial_raw_model_answer.txt").write_text( initial_rejected_raw + "\n", encoding="utf-8", ) rejected_raw = str(exc.payload.get("raw_answer") or "") if rejected_raw: (run_dir / "raw_model_answer.txt").write_text( rejected_raw + "\n", encoding="utf-8", ) transient = self._transient_brain_failure(exc) infrastructure_failures = int( record.get("infrastructure_failures") or 0 ) if transient: ( infrastructure_failures, result_status, retryable, ) = self._infrastructure_failure_outcome( record ) else: record["status"] = "failed" result_status = "failed" retryable = False record["last_error"] = error_text record["updated_at"] = _utc_now() (run_dir / "ERROR.txt").write_text(error_text + "\n", encoding="utf-8") result = { "ok": False, "skill_id": skill_id, "lane": bundle["lane"], "status": result_status, "retryable": retryable, "infrastructure_failure": transient, "infrastructure_failures": infrastructure_failures, "error": error_text, "evidence": str(run_dir), } state["skills"][skill_id] = record state["current_skill"] = None state["updated_at"] = _utc_now() _atomic_json_write(self.state_path, state) _atomic_json_write(run_dir / "result.json", result) return {**result, "practice_status": self.status(state)} def run_tool_evidence(self, skill_id: str) -> dict[str, Any]: with self._lock() as lock_handle: del lock_handle network = self._verify_evidence_root() state = self._state() record = (state.get("skills") or {}).get(skill_id) if not record: raise KeyError(f"Unknown skill: {skill_id}") if record.get("lane") != "tool_evidence": raise RuntimeError("This skill is not in the tool-evidence lane.") if not record.get("knowledge_certified"): raise RuntimeError("Three successful knowledge rounds are required first.") if record.get("forge_allowed"): return { "ok": True, "already_certified": True, "skill_id": skill_id, "status": record.get("status"), "practice_status": self.status(state), } if not record.get("cadquery_supported"): raise RuntimeError( "This skill does not yet have a validated CadQuery evidence adapter." ) attempts = int(record.get("tool_evidence_attempts") or 0) if attempts >= 3: raise RuntimeError("CadQuery tool-evidence attempts are exhausted.") state["current_skill"] = skill_id record["status"] = "cadquery_practising" record["tool_evidence_attempts"] = attempts + 1 record["updated_at"] = _utc_now() state["network"] = network state["updated_at"] = _utc_now() _atomic_json_write(self.state_path, state) run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ") + "-cadquery" domain, name = _slug_path(skill_id) run_dir = self.evidence_root / domain / name / run_id run_dir.mkdir(parents=True, exist_ok=False) latest = Path(str(record.get("latest_run") or "")) response_path = latest / "practice_response.json" if latest else None snapshot_path = latest / "lesson_snapshot.json" if latest else None try: report = cadquery_engine_service.run_evidence( skill_id=skill_id, output_directory=run_dir / "cadquery-tool-evidence", practice_round=3, practice_response_path=response_path if response_path and response_path.is_file() else None, lesson_snapshot_path=snapshot_path if snapshot_path and snapshot_path.is_file() else None, ) _atomic_json_write(run_dir / "cadquery_bridge_result.json", report) paths = self._skill_paths(skill_id) certification = _load_json(paths["certification"], {}) manifest = _load_json(paths["manifest"], {}) tests = _load_json(paths["tests"], {"skill_id": skill_id, "mandatory": []}) latest_assessment = _load_json(latest / "assessment.json", {}) if latest else {} by_id = { item.get("test_id"): item for item in latest_assessment.get("tests") or [] } for test in tests.get("mandatory") or []: result = by_id.get(test.get("test_id"), {}) if result.get("pass"): test["pass"] = True test["evidence"] = str(latest / "assessment.json") tool_test = next( ( item for item in tests.get("mandatory") or [] if item.get("test_id") == "cadquery_tool_evidence" ), None, ) if tool_test is None: tool_test = { "test_id": "cadquery_tool_evidence", "pass": True, "evidence": str(run_dir / "cadquery-tool-evidence" / "cadquery_evidence_report.json"), } tests.setdefault("mandatory", []).append(tool_test) else: tool_test.update( { "pass": True, "evidence": str(run_dir / "cadquery-tool-evidence" / "cadquery_evidence_report.json"), } ) evidence_paths = list(certification.get("evidence_paths") or []) if str(run_dir) not in evidence_paths: evidence_paths.append(str(run_dir)) certification.update( { "status": "certified", "forge_allowed": True, "knowledge_certified": True, "tool_engine": "CadQuery", "tool_engine_version": report.get("engine_version"), "tool_bridge_version": report.get("bridge_version"), "tool_evidence_attempts": attempts + 1, "tool_evidence_report": report.get("report_path"), "certified_at": _utc_now(), "certified_by": "SHiRE Academy CadQuery Bridge 0002C", "evidence_paths": evidence_paths, "test_results": tests.get("mandatory") or [], "remaining_requirements": [], "limitations": [ "Certification covers digital CadQuery modelling and validation only.", "Printer, material, physical, medical and professional performance are not implied.", ], } ) manifest["status"] = "certified" manifest["forge_allowed"] = True _atomic_json_write(paths["tests"], tests) _atomic_json_write(paths["certification"], certification) _atomic_json_write(paths["manifest"], manifest) self._update_index(skill_id, "certified", True) record.update( { "status": "cadquery_certified", "forge_allowed": True, "latest_tool_run": str(run_dir), "last_error": None, "updated_at": _utc_now(), } ) result = { "ok": True, "skill_id": skill_id, "status": record["status"], "forge_allowed": True, "engine": "CadQuery", "evidence": str(run_dir), } except Exception as exc: record.update( { "status": "tool_failed", "forge_allowed": False, "last_error": str(exc), "latest_tool_run": str(run_dir), "updated_at": _utc_now(), } ) (run_dir / "ERROR.txt").write_text(str(exc) + "\n", encoding="utf-8") result = { "ok": False, "skill_id": skill_id, "status": "tool_failed", "forge_allowed": False, "error": str(exc), "evidence": str(run_dir), } state["skills"][skill_id] = record state["current_skill"] = None state["updated_at"] = _utc_now() _atomic_json_write(self.state_path, state) _atomic_json_write(run_dir / "result.json", result) return {**result, "practice_status": self.status(state)} def reconcile_index(self) -> dict[str, Any]: with self._lock() as lock_handle: del lock_handle if not self.index_path.is_file(): raise RuntimeError("Academy index is missing.") updated = 0 with sqlite3.connect(self.index_path) as connection: for item in self._catalog(): skill_id = str(item["skill_id"]) paths = self._skill_paths(skill_id) manifest = _load_json(paths["manifest"], {}) certification = _load_json(paths["certification"], {}) status = str(certification.get("status") or manifest.get("status") or "reading") forge_allowed = bool( certification.get("forge_allowed", manifest.get("forge_allowed", False)) ) connection.execute( "UPDATE skills SET status = ?, forge_allowed = ? WHERE skill_id = ?", (status, int(forge_allowed), skill_id), ) updated += 1 connection.commit() return {"ok": True, "updated_skills": updated} practice_service = DesignAcademyPracticeService()