#!/usr/bin/env python3 from __future__ import annotations import argparse import hashlib import json import sys import urllib.error import urllib.request from datetime import datetime, timezone from pathlib import Path from typing import Any REPO = Path("/home/shire3d/ARMOR") if str(REPO) not in sys.path: sys.path.insert(0, str(REPO)) from ai.brain_server import ( ACADEMY_ROUND_JSON_SCHEMAS, canonicalise_academy_answer, ) from services.design_academy_practice_service import ( AcademyBrainResponseError, practice_service, ) STATE = REPO / "data/academy/design_mastery/runtime/practice_state.json" AUTO = REPO / "data/academy/design_mastery/runtime/autopilot_state.json" SPRINT = REPO / "data/academy/design_mastery/runtime/commercial_design_sprint.json" EVIDENCE_ROOT = Path( "/SHiREVault/SHiREAcademy/Autopilot/Commercial" ) MODEL_URL = "http://127.0.0.1:11434/api/chat" MODEL_NAME = "shire-mini-fast:qwen3-1.7b" def load(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding="utf-8")) def sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def write_json(path: Path, payload: Any) -> None: path.write_text( json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) def extract_response( payload: dict[str, Any], round_number: int, ) -> dict[str, Any]: for key in ( "answer", "response", "canonical_answer", "canonical", "normalised_answer", ): value = payload.get(key) if isinstance(value, dict): return value if isinstance(value, str) and value.strip(): try: parsed = json.loads(value) if isinstance(parsed, dict): return parsed except json.JSONDecodeError: normalised = canonicalise_academy_answer( value, round_number, ) return json.loads(normalised["canonical"]) raw = payload.get("raw_answer") if isinstance(raw, str) and raw.strip(): normalised = canonicalise_academy_answer( raw, round_number, ) return json.loads(normalised["canonical"]) raise RuntimeError( "Brain response did not contain a usable Academy answer" ) def repair_round3_response( prompt: str, rejected_raw: str, evidence: Path, ) -> dict[str, Any]: repair_request = { "model": MODEL_NAME, "think": False, "stream": False, "messages": [ { "role": "system", "content": ( "Repair one rejected SHiRE Academy Round 3 answer. " "Return only one JSON object matching the supplied schema. " "Use exactly these root fields: round, transfer_project, " "boundary_cases, provenance_review, safety_gate and " "limitations. transfer_project must contain title, goal, " "inputs, exactly three steps, outputs and exactly two " "measurable acceptance_criteria. boundary_cases must " "contain minimum, nominal, maximum and invalid. " "provenance_review must use licence_status unknown, " "decision quarantine and commercial_use blocked, with a " "reason of at least 20 characters. safety_gate must use " "risk_class low, action continue, " "forge_execution_allowed false and a reason of at least " "20 characters. Give exactly two honest limitations. " "Preserve relevant subject content from the rejected " "draft. Do not claim tool execution, physical validation, " "Forge access, release authority or commercial readiness." ), }, { "role": "user", "content": json.dumps( { "academy_prompt": prompt, "rejected_raw_answer": rejected_raw, }, sort_keys=True, ), }, ], "options": { "temperature": 0.0, "num_predict": 720, "num_ctx": 1536, }, "format": ACADEMY_ROUND_JSON_SCHEMAS[3], } write_json( evidence / "round3-repair-request.json", repair_request, ) request = urllib.request.Request( MODEL_URL, data=json.dumps(repair_request).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen( request, timeout=420, ) as response: repair_payload = json.loads( response.read().decode("utf-8") ) except ( urllib.error.URLError, TimeoutError, json.JSONDecodeError, ) as exc: raise RuntimeError( f"Round 3 schema repair request failed: {exc}" ) from exc write_json( evidence / "round3-repair-response.json", repair_payload, ) raw = str( (repair_payload.get("message") or {}).get("content") or "" ).strip() if not raw: raise RuntimeError( "Round 3 schema repair returned an empty answer" ) normalised = canonicalise_academy_answer(raw, 3) repaired = json.loads(normalised["canonical"]) write_json( evidence / "round3-repaired-answer.json", repaired, ) return { "ok": True, "context_profile": "academy_round_specific_v5", "general_context_injected": False, "output_contract": "academy_round_json_v6", "exam_round": 3, "structured_output_schema": True, "canonical_json_gate": True, "answer": repaired, "raw_answer": raw, "repair_used": True, "repair_model": MODEL_NAME, } def generate_round3_direct( prompt: str, evidence: Path, ) -> dict[str, Any]: direct_request = { "model": MODEL_NAME, "think": False, "stream": False, "messages": [ { "role": "system", "content": ( "Complete SHiRE Academy Round 3 using only the supplied " "skill prompt. Return one JSON object matching the supplied " "schema exactly. Create a harmless transfer project with " "exactly three steps and exactly two measurable acceptance " "criteria. Include minimum, nominal, maximum and invalid " "boundary cases. Unknown source provenance must remain " "quarantined and commercial use blocked. " "forge_execution_allowed must be false. Give exactly two " "honest limitations. Do not claim tool execution, physical " "validation, Forge access or commercial readiness." ), }, { "role": "user", "content": prompt, }, ], "options": { "temperature": 0.0, "num_predict": 620, "num_ctx": 1536, }, "format": ACADEMY_ROUND_JSON_SCHEMAS[3], } write_json( evidence / "round3-direct-request.json", direct_request, ) request = urllib.request.Request( MODEL_URL, data=json.dumps(direct_request).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen( request, timeout=480, ) as response: direct_payload = json.loads( response.read().decode("utf-8") ) except ( urllib.error.URLError, TimeoutError, json.JSONDecodeError, ) as exc: write_json( evidence / "round3-direct-failure.json", { "classification": "infrastructure_failure", "retryable": True, "error": str(exc), }, ) raise RuntimeError( f"Direct Round 3 fallback failed: {exc}" ) from exc write_json( evidence / "round3-direct-response.json", direct_payload, ) raw = str( (direct_payload.get("message") or {}).get("content") or "" ).strip() if not raw: raise RuntimeError( "Direct Round 3 fallback returned an empty answer" ) normalised = canonicalise_academy_answer(raw, 3) answer = json.loads(normalised["canonical"]) write_json( evidence / "round3-direct-answer.json", answer, ) return { "ok": True, "context_profile": "academy_round_specific_v5", "general_context_injected": False, "output_contract": "academy_round_json_v6", "exam_round": 3, "structured_output_schema": True, "canonical_json_gate": True, "answer": answer, "raw_answer": raw, "repair_used": False, "direct_fallback_used": True, "direct_model": MODEL_NAME, } def main() -> int: parser = argparse.ArgumentParser( description=( "Generic uncounted SHiRE Academy qualification" ) ) parser.add_argument("skill_id") parser.add_argument( "round_number", type=int, choices=(1, 2, 3), ) parser.add_argument( "--direct-round3", action="store_true", help="Skip the Brain API and use one direct schema-constrained Round 3 request.", ) args = parser.parse_args() if args.direct_round3 and args.round_number != 3: parser.error("--direct-round3 is valid only for Round 3") state_before = sha256(STATE) practice = load(STATE) autopilot = load(AUTO) sprint = load(SPRINT) skill_id = args.skill_id round_number = args.round_number record = practice["skills"][skill_id] assert autopilot["current_skill"] == skill_id assert sprint["current_skill"] == skill_id assert record.get("forge_allowed") is False assert record.get("knowledge_certified") is False attempts_before = int(record.get("attempts") or 0) successes_before = int( record.get("successful_attempts") or 0 ) bundle = practice_service._lesson_bundle(skill_id) prompt = practice_service._practice_prompt( skill_id, bundle, round_number, ) timestamp = datetime.now(timezone.utc).strftime( "%Y%m%dT%H%M%S%fZ" ) domain, name = skill_id.split(".", 1) evidence = ( EVIDENCE_ROOT / sprint["sprint_id"] / domain / name / f"{timestamp}-UNCOUNTED-R{round_number}" ) evidence.mkdir(parents=True, exist_ok=False) (evidence / "prompt.txt").write_text( prompt, encoding="utf-8", ) write_json( evidence / "pre_state.json", { "skill_id": skill_id, "round": round_number, "attempts": attempts_before, "successful_attempts": successes_before, "knowledge_certified": record.get("knowledge_certified"), "forge_allowed": record.get("forge_allowed"), "practice_state_sha256": state_before, }, ) risk_mode = str( bundle["manifest"].get("risk_level") or "low" ).lower() repair_used = False if args.direct_round3: payload = generate_round3_direct( prompt, evidence, ) else: try: payload = practice_service._ask_brain( prompt, risk_mode, round_number, ) except AcademyBrainResponseError as exc: error_payload = dict(exc.payload or {}) write_json( evidence / "brain-error.json", { "error": str(exc), "payload": error_payload, }, ) rejected_raw = str( error_payload.get("raw_answer") or "" ).strip() retryable_round3_failure = ( round_number == 3 and error_payload.get("retryable") is True and error_payload.get("infrastructure_failure") is True ) if not retryable_round3_failure: raise if rejected_raw: payload = repair_round3_response( prompt, rejected_raw, evidence, ) repair_used = True else: payload = generate_round3_direct( prompt, evidence, ) write_json( evidence / "brain_payload.json", payload, ) response = extract_response( payload, round_number, ) guard_error = ( practice_service._response_contract_guard_error( bundle, response, round_number, ) ) if guard_error: raise RuntimeError(guard_error) assessment = practice_service.evaluate_response( skill_id, bundle, response, round_number, ) classification = ( "UNCOUNTED_QUALIFICATION_PASS" if assessment["passed"] else "UNCOUNTED_QUALIFICATION_FAIL" ) write_json( evidence / "response.json", response, ) write_json( evidence / "assessment.json", assessment, ) result = { "classification": classification, "skill_id": skill_id, "round": round_number, "passed": assessment["passed"], "score_percent": assessment["score_percent"], "evidence": str(evidence), "counted_attempt_consumed": False, "forge_allowed": False, "automatic_schema_repair_used": repair_used, } write_json( evidence / "result.json", result, ) practice_after = load(STATE) record_after = practice_after["skills"][skill_id] assert sha256(STATE) == state_before assert int( record_after.get("attempts") or 0 ) == attempts_before assert int( record_after.get("successful_attempts") or 0 ) == successes_before assert record_after.get("knowledge_certified") is False assert record_after.get("forge_allowed") is False print(json.dumps({ "ok": assessment["passed"], "classification": classification, "skill_id": skill_id, "round": round_number, "score_percent": assessment["score_percent"], "tests": assessment["tests"], "evidence": str(evidence), "automatic_schema_repair_used": repair_used, "direct_round3_fallback_used": bool(payload.get("direct_fallback_used")), "counted_attempt_consumed": False, "practice_state_unchanged": True, "forge_allowed": False, }, indent=2, sort_keys=True)) return 0 if assessment["passed"] else 2 if __name__ == "__main__": raise SystemExit(main())