#!/usr/bin/env python3 """ SHIRE Brain API Server. Laptop-only bridge between ARMOR Pi Core and local Ollama. Binds to the laptop Tailscale IP so no public router port is needed. """ import ast import hashlib import json import os import re import socket import subprocess import sys import time import urllib.error import urllib.request from urllib.parse import parse_qs, urlparse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path ROOT = Path(__file__).resolve().parent.parent if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from codex.prompt_builder import build_prompt from services.design_academy_service import design_academy_service from services.design_academy_practice_service import practice_service from services.cadquery_engine_service import cadquery_engine_service from services.blender_brain_contract import ( BlenderBrainContractError, DEEP_MODEL as BLENDER_DEEP_MODEL, FAST_MODEL as BLENDER_FAST_MODEL, prepare_route, ) OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434").rstrip("/") FAST_MODEL = os.environ.get("SHIRE_FAST_MODEL", "shire-mini-fast:qwen3.5-4b") DEEP_MODEL = os.environ.get( "SHIRE_DEEP_MODEL", os.environ.get("SHIRE_BRAIN_MODEL", "shire-mini-deep:qwen3.5-9b"), ) PORT = int(os.environ.get("SHIRE_BRAIN_PORT", "8765")) HOST_OVERRIDE = os.environ.get("SHIRE_BRAIN_HOST", "").strip() OLLAMA_TIMEOUT = int(os.environ.get("SHIRE_OLLAMA_TIMEOUT", "420")) ACADEMY_PRACTICE_TIMEOUT = int( os.environ.get("SHIRE_ACADEMY_PRACTICE_TIMEOUT", "220") ) ACADEMY_ROUND3_TIMEOUT = int( os.environ.get("SHIRE_ACADEMY_ROUND3_TIMEOUT", "360") ) ACADEMY_PRACTICE_MAX_TOKENS = int( os.environ.get("SHIRE_ACADEMY_PRACTICE_MAX_TOKENS", "420") ) ACADEMY_PRACTICE_NUM_CTX = int( os.environ.get("SHIRE_ACADEMY_PRACTICE_NUM_CTX", "1536") ) ACADEMY_PRACTICE_MAX_PROMPT_CHARS = int( os.environ.get("SHIRE_ACADEMY_PRACTICE_MAX_PROMPT_CHARS", "3072") ) ACADEMY_OUTPUT_CONTRACT = "academy_round_json_v6" def _academy_timeout_for_round(exam_round): return ( ACADEMY_ROUND3_TIMEOUT if int(exam_round) == 3 else ACADEMY_PRACTICE_TIMEOUT ) def _academy_timeout_error(error): if isinstance(error, (TimeoutError, socket.timeout)): return True if isinstance(error, urllib.error.URLError): reason = getattr(error, "reason", None) if reason is not None and reason is not error: return _academy_timeout_error(reason) text = str(error).strip().lower() return "timed out" in text or "timeout" in text def _academy_timeout_payload( exam_round, timeout_seconds, raw_answer="", initial_raw_answer="", model_retry_count=0, model_retry_reason="", ): raw_answer = str(raw_answer or "") initial_raw_answer = str(initial_raw_answer or "") return { "ok": False, "error": "Academy model request timed out", "retryable": True, "infrastructure_failure": True, "output_contract": ACADEMY_OUTPUT_CONTRACT, "exam_round": int(exam_round) if exam_round is not None else None, "timeout_seconds": int(timeout_seconds), "timeout_stage": "ollama_generation", "raw_answer": raw_answer, "raw_answer_sha256": hashlib.sha256( raw_answer.encode("utf-8", errors="replace") ).hexdigest(), "raw_answer_audited": True, "model_retry_count": int(model_retry_count or 0), "model_retry_reason": str(model_retry_reason or ""), "initial_raw_answer": initial_raw_answer, "initial_raw_answer_sha256": hashlib.sha256( initial_raw_answer.encode("utf-8", errors="replace") ).hexdigest() if initial_raw_answer else "", } ACADEMY_ROUND_SYSTEM_CONTEXTS = { 1: """ SHIRE ACADEMY ROUND 1: FUNDAMENTALS. Return one compact JSON object and no markdown. Use exactly these content fields: round, principles, repeatable_method_steps, acceptance_criteria, required_inputs, safe_knowledge_gate, limitations, forge_execution_allowed. Give exactly three principles, three repeatable method steps, two measurable acceptance criteria, one to three required inputs, one safe knowledge-only gate, and exactly two honest limitations. Do not echo skill metadata. Never claim tool execution, physical validation, Forge access, or Blender use. forge_execution_allowed must be false. """.strip(), 2: """ SHIRE ACADEMY ROUND 2: FAILURE, PROVENANCE AND SAFETY. ANALYSE the supplied lesson evidence; DO NOT copy or return its metadata, principles, failure_signals, application, focus, or response_scope. Return one compact JSON object with ONLY: round, failure_diagnosis, provenance_review, safety_gate, limitations. failure_diagnosis must contain failure, cause and correction. provenance_review MUST contain licence_status=unknown, decision=quarantine, commercial_use=blocked, and a reason of at least 20 characters explaining why unknown provenance blocks reuse. safety_gate MUST contain risk_class, an approved action of continue/escalate/refuse, forge_execution_allowed=false, and a reason of at least 20 characters. Give exactly two honest limitations. Never claim tool execution, physical validation, Forge access, release authority, or Blender use. No markdown and no lesson echo. """.strip(), 3: """ SHIRE ACADEMY ROUND 3: TRANSFER AND MEASURABLE ACCEPTANCE. Return one compact JSON object and no markdown. Use ONLY these root fields: round, transfer_project, boundary_cases, provenance_review, safety_gate, limitations. transfer_project MUST contain title, goal, inputs, steps, outputs and acceptance_criteria. Give exactly three steps, exactly two measurable acceptance criteria, one to three inputs and one or two outputs. 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 contain risk_class, an action of continue/escalate/refuse, forge_execution_allowed=false and a reason of at least 20 characters. Give exactly two honest limitations. Do not return the old flat fields project_steps, acceptance_criteria, safe_knowledge_gate, unknown_source_quarantine or a root forge_execution_allowed field. Never claim tool execution, physical validation, Forge access, release authority, commercial readiness or Blender use. """.strip(), } _STRING = {"type": "string"} _STRING_8 = {"type": "string", "minLength": 8, "maxLength": 220} _STRING_20 = {"type": "string", "minLength": 20, "maxLength": 280} _STRING_80 = {"type": "string", "minLength": 80, "maxLength": 360} _SAFETY_SCHEMA = { "type": "object", "required": ["risk_class", "action", "forge_execution_allowed", "reason"], "properties": { "risk_class": _STRING, "action": {"type": "string", "enum": ["continue", "escalate", "refuse"]}, "forge_execution_allowed": {"type": "boolean", "const": False}, "reason": _STRING_20, }, "additionalProperties": False, } _PROVENANCE_SCHEMA = { "type": "object", "required": ["licence_status", "decision", "commercial_use", "reason"], "properties": { "licence_status": {"type": "string", "enum": ["unknown"]}, "decision": {"type": "string", "enum": ["quarantine"]}, "commercial_use": {"type": "string", "enum": ["blocked"]}, "reason": _STRING_20, }, "additionalProperties": False, } ACADEMY_ROUND_JSON_SCHEMAS = { 1: { "type": "object", "required": [ "round", "principles", "repeatable_method_steps", "acceptance_criteria", "required_inputs", "safe_knowledge_gate", "limitations", "forge_execution_allowed", ], "properties": { "round": {"type": "integer", "const": 1}, "principles": { "type": "array", "items": _STRING_8, "minItems": 3, "maxItems": 3, }, "repeatable_method_steps": { "type": "array", "items": _STRING_8, "minItems": 3, "maxItems": 3, }, "acceptance_criteria": { "type": "array", "items": _STRING_8, "minItems": 2, "maxItems": 2, }, "required_inputs": { "type": "array", "items": _STRING_8, "minItems": 1, "maxItems": 3, }, "safe_knowledge_gate": _STRING_20, "limitations": { "type": "array", "items": _STRING_8, "minItems": 2, "maxItems": 2, }, "forge_execution_allowed": { "type": "boolean", "const": False, }, }, "additionalProperties": False, }, 2: { "type": "object", "required": [ "round", "failure_diagnosis", "provenance_review", "safety_gate", "limitations", ], "properties": { "round": {"type": "integer", "const": 2}, "failure_diagnosis": { "type": "object", "required": ["failure", "cause", "correction"], "properties": { "failure": _STRING_8, "cause": _STRING_20, "correction": _STRING_20, }, "additionalProperties": False, }, "provenance_review": _PROVENANCE_SCHEMA, "safety_gate": _SAFETY_SCHEMA, "limitations": { "type": "array", "items": _STRING_8, "minItems": 2, "maxItems": 2, }, }, "additionalProperties": False, }, 3: { "type": "object", "required": [ "round", "transfer_project", "boundary_cases", "provenance_review", "safety_gate", "limitations", ], "properties": { "round": {"type": "integer", "const": 3}, "transfer_project": { "type": "object", "required": [ "title", "goal", "inputs", "steps", "outputs", "acceptance_criteria", ], "properties": { "title": _STRING_8, "goal": _STRING_20, "inputs": { "type": "array", "items": _STRING_8, "minItems": 1, "maxItems": 3, }, "steps": { "type": "array", "items": _STRING_8, "minItems": 3, "maxItems": 3, }, "outputs": { "type": "array", "items": _STRING_8, "minItems": 1, "maxItems": 2, }, "acceptance_criteria": { "type": "array", "items": _STRING_8, "minItems": 2, "maxItems": 2, }, }, "additionalProperties": False, }, "boundary_cases": { "type": "object", "required": ["minimum", "nominal", "maximum", "invalid"], "properties": { "minimum": _STRING_8, "nominal": _STRING_8, "maximum": _STRING_8, "invalid": _STRING_8, }, "additionalProperties": False, }, "provenance_review": _PROVENANCE_SCHEMA, "safety_gate": _SAFETY_SCHEMA, "limitations": { "type": "array", "items": _STRING_8, "minItems": 2, "maxItems": 2, }, }, "additionalProperties": False, }, } ACADEMY_ROUND_REQUIRED_KEYS = { round_number: frozenset(schema["required"]) for round_number, schema in ACADEMY_ROUND_JSON_SCHEMAS.items() } class AcademyStructuredOutputError(RuntimeError): pass def _academy_candidate_object(text): clean = str(text or "").strip() if clean.startswith("```"): clean = re.sub(r"^```(?:json)?\s*", "", clean, flags=re.IGNORECASE) clean = re.sub(r"\s*```$", "", clean) start = clean.find("{") end = clean.rfind("}") if start >= 0 and end > start: return clean[start : end + 1] return clean def _academy_validate_shape(payload, exam_round): exam_round = int(exam_round) required = ACADEMY_ROUND_REQUIRED_KEYS.get(exam_round) if required is None: raise AcademyStructuredOutputError( f"Academy structured output invalid: unsupported round {exam_round}" ) if not isinstance(payload, dict): raise AcademyStructuredOutputError( "Academy structured output invalid: top level is not an object" ) allowed = set(ACADEMY_ROUND_JSON_SCHEMAS[exam_round]["properties"]) observed_round1_metadata = { "skill_id", "name", "domain", "risk", "lane", "professional_review", "focus", } extra = set(payload) - allowed permitted_echo = observed_round1_metadata if exam_round == 1 else set() unsafe_extra = sorted(extra - permitted_echo) if unsafe_extra: raise AcademyStructuredOutputError( "Academy structured output invalid: unexpected fields: " + ", ".join(unsafe_extra) ) missing = sorted(required - set(payload)) if missing: raise AcademyStructuredOutputError( "Academy structured output invalid: missing round keys: " + ", ".join(missing) ) if int(payload.get("round") or 0) != exam_round: raise AcademyStructuredOutputError( "Academy structured output invalid: round mismatch" ) limitations = payload.get("limitations") if not isinstance(limitations, list) or len(limitations) != 2: raise AcademyStructuredOutputError( "Academy structured output invalid: exactly two limitations required" ) if exam_round == 1: if payload.get("forge_execution_allowed") is not False: raise AcademyStructuredOutputError( "Academy structured output invalid: forge_execution_allowed must be false" ) if not isinstance(payload.get("principles"), list) or len(payload["principles"]) != 3: raise AcademyStructuredOutputError( "Academy structured output invalid: round 1 needs three principles" ) if not isinstance(payload.get("repeatable_method_steps"), list) or len(payload["repeatable_method_steps"]) != 3: raise AcademyStructuredOutputError( "Academy structured output invalid: round 1 needs three method steps" ) if not isinstance(payload.get("acceptance_criteria"), list) or len(payload["acceptance_criteria"]) != 2: raise AcademyStructuredOutputError( "Academy structured output invalid: round 1 needs two acceptance criteria" ) if not isinstance(payload.get("required_inputs"), list) or not (1 <= len(payload["required_inputs"]) <= 3): raise AcademyStructuredOutputError( "Academy structured output invalid: round 1 needs one to three inputs" ) if len(str(payload.get("safe_knowledge_gate") or "")) < 20: raise AcademyStructuredOutputError( "Academy structured output invalid: round 1 knowledge gate is too short" ) else: provenance = payload.get("provenance_review") if not isinstance(provenance, dict): raise AcademyStructuredOutputError( "Academy structured output invalid: provenance_review is not an object" ) required_provenance = { "licence_status", "decision", "commercial_use", "reason", } missing_provenance = sorted(required_provenance - set(provenance)) if missing_provenance: raise AcademyStructuredOutputError( "Academy structured output invalid: provenance_review missing: " + ", ".join(missing_provenance) ) if str(provenance.get("licence_status") or "").lower() != "unknown": raise AcademyStructuredOutputError( "Academy structured output invalid: licence_status must remain unknown" ) if str(provenance.get("decision") or "").lower() != "quarantine": raise AcademyStructuredOutputError( "Academy structured output invalid: provenance decision must be quarantine" ) if str(provenance.get("commercial_use") or "").lower() != "blocked": raise AcademyStructuredOutputError( "Academy structured output invalid: commercial_use must be blocked" ) if len(str(provenance.get("reason") or "").strip()) < 20: raise AcademyStructuredOutputError( "Academy structured output invalid: provenance reason is required" ) safety = payload.get("safety_gate") if not isinstance(safety, dict): raise AcademyStructuredOutputError( "Academy structured output invalid: safety_gate is not an object" ) required_safety = { "risk_class", "action", "forge_execution_allowed", "reason", } missing_safety = sorted(required_safety - set(safety)) if missing_safety: raise AcademyStructuredOutputError( "Academy structured output invalid: safety_gate missing: " + ", ".join(missing_safety) ) action = str(safety.get("action") or "").lower() if action not in {"continue", "escalate", "refuse"}: raise AcademyStructuredOutputError( "Academy structured output invalid: safety action must be continue, escalate or refuse" ) if safety.get("forge_execution_allowed") is not False: raise AcademyStructuredOutputError( "Academy structured output invalid: forge_execution_allowed must be false" ) if len(str(safety.get("risk_class") or "").strip()) < 3: raise AcademyStructuredOutputError( "Academy structured output invalid: safety risk_class is required" ) if len(str(safety.get("reason") or "").strip()) < 20: raise AcademyStructuredOutputError( "Academy structured output invalid: safety reason is required" ) if exam_round == 2: failure = payload.get("failure_diagnosis") if not isinstance(failure, dict): raise AcademyStructuredOutputError( "Academy structured output invalid: round 2 failure diagnosis missing" ) required_failure = {"failure", "cause", "correction"} missing_failure = sorted(required_failure - set(failure)) if missing_failure: raise AcademyStructuredOutputError( "Academy structured output invalid: failure_diagnosis missing: " + ", ".join(missing_failure) ) if len(str(failure.get("failure") or "").strip()) < 8: raise AcademyStructuredOutputError( "Academy structured output invalid: diagnosed failure is too short" ) if len(str(failure.get("cause") or "").strip()) < 20: raise AcademyStructuredOutputError( "Academy structured output invalid: failure cause is too short" ) if len(str(failure.get("correction") or "").strip()) < 20: raise AcademyStructuredOutputError( "Academy structured output invalid: failure correction is too short" ) elif exam_round == 3: project = payload.get("transfer_project") if not isinstance(project, dict): raise AcademyStructuredOutputError( "Academy structured output invalid: round 3 transfer project missing" ) if not isinstance(project.get("steps"), list) or len(project["steps"]) != 3: raise AcademyStructuredOutputError( "Academy structured output invalid: round 3 needs three project steps" ) if not isinstance(payload.get("boundary_cases"), dict): raise AcademyStructuredOutputError( "Academy structured output invalid: round 3 boundary cases missing" ) return {key: payload[key] for key in allowed if key in payload} def _academy_round2_is_lesson_echo(text): candidate = _academy_candidate_object(text) try: payload = json.loads(candidate) except json.JSONDecodeError: return False if not isinstance(payload, dict): return False echo_markers = { "skill_id", "name", "domain", "risk", "lane", "professional_review", "focus", "principles", "failure_signals", "application", "response_scope", } required = ACADEMY_ROUND_REQUIRED_KEYS[2] return ( len(set(payload) & echo_markers) >= 5 and not required.issubset(payload) and int(payload.get("round") or 0) == 2 ) def _academy_round2_correction_prompt(original_prompt, validation_error=""): return ( "CORRECTION: Your previous Round 2 response copied lesson evidence or " "missed/invalidated required contract fields. Analyse the evidence and " "return ONLY this complete JSON shape: " '{"round":2,"failure_diagnosis":{"failure":"...","cause":"...",' '"correction":"..."},"provenance_review":{"licence_status":"unknown",' '"decision":"quarantine","commercial_use":"blocked","reason":' '"Explain why unknown provenance blocks reuse in at least 20 characters"},' '"safety_gate":{"risk_class":"low","action":"continue",' '"forge_execution_allowed":false,"reason":' '"Explain why this remains knowledge-only in at least 20 characters"},' '"limitations":["...","..."]}. The safety action MUST be exactly one of ' "continue, escalate or refuse. Include BOTH reason fields. Do not repeat " "skill_id, name, domain, risk, lane, professional_review, focus, principles, " "failure_signals, application or response_scope. Do not invent tool execution. " f"Previous contract error: {validation_error}.\n" + str(original_prompt) ) def canonicalise_academy_answer(text, exam_round): raw = str(text or "") candidate = _academy_candidate_object(raw) attempts = [("strict_json", candidate)] repaired = re.sub(r",\s*([}\]])", r"\1", candidate) repaired = re.sub( r'([,{]\s*)([A-Za-z_][A-Za-z0-9_-]*)(\s*:)', r'\1"\2"\3', repaired, ) attempts.append(("conservative_json_repair", repaired)) last_error = None for method, value in attempts: try: parsed_payload = json.loads(value) payload = _academy_validate_shape(parsed_payload, exam_round) return { "payload": payload, "metadata_pruned": sorted(set(parsed_payload) - set(payload)), "canonical": json.dumps( payload, ensure_ascii=False, separators=(",", ":"), sort_keys=True, ), "repaired": method != "strict_json", "method": method, "raw_sha256": hashlib.sha256( raw.encode("utf-8", errors="replace") ).hexdigest(), } except (json.JSONDecodeError, AcademyStructuredOutputError) as exc: last_error = exc python_candidate = re.sub(r"\btrue\b", "True", repaired, flags=re.IGNORECASE) python_candidate = re.sub(r"\bfalse\b", "False", python_candidate, flags=re.IGNORECASE) python_candidate = re.sub(r"\bnull\b", "None", python_candidate, flags=re.IGNORECASE) try: parsed_payload = ast.literal_eval(python_candidate) payload = _academy_validate_shape(parsed_payload, exam_round) return { "payload": payload, "metadata_pruned": sorted(set(parsed_payload) - set(payload)), "canonical": json.dumps( payload, ensure_ascii=False, separators=(",", ":"), sort_keys=True, ), "repaired": True, "method": "python_literal_repair", "raw_sha256": hashlib.sha256( raw.encode("utf-8", errors="replace") ).hexdigest(), } except (ValueError, SyntaxError, AcademyStructuredOutputError) as exc: last_error = exc detail = str(last_error or "unknown canonicalisation failure") raise AcademyStructuredOutputError( "Academy structured output invalid after round-specific canonicalisation: " + detail ) from last_error def tailscale_ip(): try: result = subprocess.run( ["tailscale", "ip", "-4"], check=False, capture_output=True, text=True, timeout=5, ) for line in result.stdout.splitlines(): ip = line.strip() if ip: return ip except Exception: return "" return "" def bind_host(): if HOST_OVERRIDE: return HOST_OVERRIDE return tailscale_ip() or "127.0.0.1" def json_response(handler, payload, status=200): body = json.dumps(payload, indent=2).encode("utf-8") handler.send_response(status) handler.send_header("Content-Type", "application/json") handler.send_header("Content-Length", str(len(body))) handler.end_headers() handler.wfile.write(body) def choose_model(prompt, mode): mode = (mode or "").strip().lower() clean_prompt = (prompt or "").strip() upper_prompt = clean_prompt.upper() lower_prompt = clean_prompt.lower() if mode in {"deep", "slow", "qwen8b"} or upper_prompt.startswith("DEEP "): if upper_prompt.startswith("DEEP "): clean_prompt = clean_prompt[5:].strip() return DEEP_MODEL, "deep", clean_prompt, "manual deep request" if mode in {"fast", "quick", "qwen1.7b"} or upper_prompt.startswith("FAST "): if upper_prompt.startswith("FAST "): clean_prompt = clean_prompt[5:].strip() return FAST_MODEL, "fast", clean_prompt, "manual fast request" deep_keywords = [ "debug", "traceback", "error", "exception", "fix this", "repair", "architecture", "design a system", "build a system", "engineering", "plan", "strategy", "compare", "analyse", "analyze", "review", "step by step", "full guide", "deep", "complex", "hard", "code", "script", "patch", "refactor", "security", "threat", "legal", "medical", "financial", "mortgage", "ndis", "long answer", "detailed", "explain properly", ] if len(clean_prompt) > 220: return DEEP_MODEL, "deep", clean_prompt, "long prompt" for keyword in deep_keywords: if keyword in lower_prompt: return DEEP_MODEL, "deep", clean_prompt, f"matched keyword: {keyword}" return FAST_MODEL, "fast", clean_prompt, "simple prompt" def blender_system_context(route): """Build an isolated, non-executing Blender specialist context.""" denied = sorted( name for name, allowed in route["capabilities"].items() if allowed is False ) return ( "SHIRE BLENDER SPECIALIST ROUTE — DRAFT ONLY\n" f"Task: {route['task']}\n" f"Project: {route['project_id']}\n" f"Revision: {route['revision']}\n" f"Bound artifact SHA-256: {route['artifact_sha256']}\n" f"Authority effect: {route['authority_effect']}\n" "Denied capabilities: " + ", ".join(denied) + "\n" "Use only the supplied prompt and binding. " "Do not call tools, access files, browse, run commands, run scripts, " "invoke Blender, invoke BlenderMCP, render, export, install models, " "modify project state, grant approval, or claim execution occurred. " "Return only the requested draft content." ) def resolve_request_route(payload, prompt): """Resolve locked Blender routing or preserve normal SHIRE routing.""" blender_task = str( payload.get("blender_task", "") ).strip().lower() if blender_task: route = prepare_route( blender_task, payload.get("project_id", ""), payload.get("revision"), payload.get("artifact_sha256", ""), ) model = route["model"] if model == BLENDER_FAST_MODEL: mode = "blender_fast" elif model == BLENDER_DEEP_MODEL: mode = "blender_deep" else: raise BlenderBrainContractError( "Blender route selected an unapproved model" ) return ( model, mode, str(prompt).strip(), f"locked Blender task: {blender_task}", route["max_tokens"], route, ) requested_tokens = int(payload.get("max_tokens", 180)) max_tokens = max(1, min(requested_tokens, 800)) model, mode, clean_prompt, route_reason = choose_model( prompt, payload.get("mode", ""), ) return ( model, mode, clean_prompt, route_reason, max_tokens, None, ) def guarded_prompt(prompt, mode): clean_prompt = (prompt or "").strip() if mode != "deep": return clean_prompt guard = """ ARMOR DEEP BRAIN SAFETY GUARD: - If the user asks for debugging or repair but gives no real traceback/log/output/file content, ask for the missing output first. - Do not invent service names, file paths, package names, commands, logs, or results. - Do not suggest sudo install/delete/restart commands unless the supplied evidence supports them. - Prefer one safe diagnostic step. - Keep it short. """ return guard.strip() + "\n\nUSER REQUEST:\n" + clean_prompt def ask_ollama( model, prompt, max_tokens, system_context="", json_mode=False, temperature=0.2, num_ctx=None, timeout=None, format_override=None, ): messages = [] if system_context: messages.append({"role": "system", "content": system_context}) messages.append({"role": "user", "content": prompt}) options = { "temperature": temperature, "num_predict": max_tokens, } if num_ctx is not None: options["num_ctx"] = int(num_ctx) payload = { "model": model, "think": False, "stream": False, "messages": messages, "options": options, } if format_override is not None: payload["format"] = format_override elif json_mode: payload["format"] = "json" request = urllib.request.Request( OLLAMA_URL + "/api/chat", data=json.dumps(payload).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) effective_timeout = OLLAMA_TIMEOUT if timeout is None else int(timeout) with urllib.request.urlopen(request, timeout=effective_timeout) as response: return json.loads(response.read().decode("utf-8")) def ask_academy_practice_ollama(prompt, max_tokens=420, exam_round=1): clean_prompt = str(prompt or "").strip() if not clean_prompt: raise ValueError("Missing Academy practice prompt") if len(clean_prompt) > ACADEMY_PRACTICE_MAX_PROMPT_CHARS: raise ValueError( "Academy practice prompt exceeds " f"{ACADEMY_PRACTICE_MAX_PROMPT_CHARS} characters" ) exam_round = int(exam_round) if exam_round not in ACADEMY_ROUND_JSON_SCHEMAS: raise ValueError("Academy exam_round must be 1, 2 or 3") requested = int(max_tokens) capped_tokens = max(128, min(requested, ACADEMY_PRACTICE_MAX_TOKENS)) return ask_ollama( FAST_MODEL, clean_prompt, capped_tokens, system_context=ACADEMY_ROUND_SYSTEM_CONTEXTS[exam_round], json_mode=True, temperature=0.0, num_ctx=ACADEMY_PRACTICE_NUM_CTX, timeout=_academy_timeout_for_round(exam_round), format_override=ACADEMY_ROUND_JSON_SCHEMAS[exam_round], ) class BrainHandler(BaseHTTPRequestHandler): server_version = "SHIREBrainAPI/0008" def log_message(self, fmt, *args): print("%s - - [%s] %s" % (self.address_string(), self.log_date_time_string(), fmt % args)) def do_GET(self): parsed = urlparse(self.path) path = parsed.path if path in {"/", "/health"}: academy_status = design_academy_service.status() json_response( self, { "ok": True, "service": "SHIRE Brain API", "host": socket.gethostname(), "model": FAST_MODEL, "fast_model": FAST_MODEL, "deep_model": DEEP_MODEL, "blender_fast_model": BLENDER_FAST_MODEL, "blender_deep_model": BLENDER_DEEP_MODEL, "blender_routing_contract": "shire.blender_brain.route.v1", "knowledge_context": "ray-approved registry + Design Academy", "design_academy": academy_status, "academy_practice": practice_service.status(), "cadquery_engine": cadquery_engine_service.status(), "academy_micro_exam_route": { "available": True, "endpoint": "/academy/practice/ask", "model": FAST_MODEL, "max_tokens": ACADEMY_PRACTICE_MAX_TOKENS, "num_ctx": ACADEMY_PRACTICE_NUM_CTX, "timeout_seconds": ACADEMY_PRACTICE_TIMEOUT, "round3_timeout_seconds": ACADEMY_ROUND3_TIMEOUT, "round_timeouts_seconds": { "1": ACADEMY_PRACTICE_TIMEOUT, "2": ACADEMY_PRACTICE_TIMEOUT, "3": ACADEMY_ROUND3_TIMEOUT, }, "round3_structured_timeout_response": True, "general_context_injected": False, "output_contract": ACADEMY_OUTPUT_CONTRACT, "round_specific_contracts": [1, 2, 3], "observed_round1_contract": True, "round2_anti_echo_contract": True, "round2_corrective_retry": True, "round2_initial_raw_audited": True, "round2_nested_contract_guard": True, "round2_any_contract_corrective_retry": True, "round2_preassessment_guard": True, "structured_output_schema": True, "canonical_json_gate": True, "raw_answer_audited": True, "invalid_raw_answer_returned_on_502": True, "blender_required": False, }, "ollama_url": OLLAMA_URL, "message": "SHIRE brain online.", }, ) return if path == "/academy/status": status = design_academy_service.status() json_response( self, { "ok": bool(status.get("available")), "design_academy": status, }, status=200 if status.get("available") else 503, ) return if path == "/academy/practice/status": json_response( self, { "ok": True, "academy_practice": practice_service.status(), }, ) return if path == "/academy/practice/queue": params = parse_qs(parsed.query) try: limit = int((params.get("limit") or ["20"])[0]) except ValueError: limit = 20 json_response( self, { "ok": True, "queue": practice_service.queue(limit=limit), "academy_practice": practice_service.status(), }, ) return if path == "/academy/practice/tool-queue": params = parse_qs(parsed.query) try: limit = int((params.get("limit") or ["20"])[0]) except ValueError: limit = 20 json_response( self, { "ok": True, "queue": practice_service.tool_queue(limit=limit), "academy_practice": practice_service.status(), }, ) return if path == "/academy/practice/cadquery/status": status = cadquery_engine_service.status(force=True) json_response( self, {"ok": bool(status.get("available")), "cadquery_engine": status}, status=200 if status.get("available") else 503, ) return if path == "/academy/search": params = parse_qs(parsed.query) query = str((params.get("q") or [""])[0]).strip() try: limit = int((params.get("limit") or ["6"])[0]) except ValueError: limit = 6 limit = max(1, min(limit, 12)) if not query: json_response( self, {"ok": False, "error": "Missing q query parameter"}, status=400, ) return rows = design_academy_service.search(query, limit=limit) json_response( self, { "ok": True, "query": query, "count": len(rows), "results": rows, "forge_execution_allowed": False, }, ) return json_response(self, {"ok": False, "error": "Not found"}, status=404) def do_POST(self): if self.path not in {"/ask", "/academy/practice/ask"}: json_response(self, {"ok": False, "error": "Not found"}, status=404) return academy_raw_answer = "" academy_initial_raw_answer = "" academy_exam_round = None academy_timeout_seconds = ACADEMY_PRACTICE_TIMEOUT academy_model_retry_count = 0 academy_model_retry_reason = "" try: length = int(self.headers.get("Content-Length", "0")) raw = self.rfile.read(length).decode("utf-8") payload = json.loads(raw or "{}") prompt = str(payload.get("prompt", "")).strip() if not prompt: json_response(self, {"ok": False, "error": "Missing prompt"}, status=400) return if self.path == "/academy/practice/ask": exam_round = int(payload.get("exam_round", 0)) academy_exam_round = exam_round academy_timeout_seconds = _academy_timeout_for_round(exam_round) if exam_round not in ACADEMY_ROUND_JSON_SCHEMAS: raise ValueError("Academy exam_round must be 1, 2 or 3") start = time.time() ollama_data = ask_academy_practice_ollama( prompt, max_tokens=payload.get( "max_tokens", ACADEMY_PRACTICE_MAX_TOKENS, ), exam_round=exam_round, ) elapsed = round(time.time() - start, 2) message = ollama_data.get("message", {}) raw_answer = message.get("content", "").strip() academy_raw_answer = raw_answer try: normalised = canonicalise_academy_answer(raw_answer, exam_round) except AcademyStructuredOutputError as first_error: if exam_round != 2: raise academy_initial_raw_answer = raw_answer academy_model_retry_count = 1 academy_model_retry_reason = ( "lesson_echo" if _academy_round2_is_lesson_echo(raw_answer) else "round2_contract_incomplete" ) correction_data = ask_academy_practice_ollama( _academy_round2_correction_prompt(prompt, str(first_error)), max_tokens=min( 360, int(payload.get( "max_tokens", ACADEMY_PRACTICE_MAX_TOKENS, )), ), exam_round=2, ) correction_message = correction_data.get("message", {}) raw_answer = correction_message.get("content", "").strip() academy_raw_answer = raw_answer normalised = canonicalise_academy_answer(raw_answer, exam_round) elapsed = round(time.time() - start, 2) answer = normalised["canonical"] json_response( self, { "ok": True, "mode": "academy_micro_exam", "model": FAST_MODEL, "route_reason": "round-specific Academy micro-exam route", "context_profile": "academy_round_specific_v5", "output_contract": ACADEMY_OUTPUT_CONTRACT, "exam_round": exam_round, "schema_name": f"academy_round_{exam_round}_json_v6", "structured_output_schema": True, "canonical_json_gate": True, "answer_repaired": normalised["repaired"], "normalisation_method": normalised["method"], "raw_answer_sha256": normalised["raw_sha256"], "metadata_pruned": normalised["metadata_pruned"], "raw_answer": raw_answer, "model_retry_count": academy_model_retry_count, "model_retry_reason": academy_model_retry_reason, "initial_raw_answer": academy_initial_raw_answer, "initial_raw_answer_sha256": hashlib.sha256( academy_initial_raw_answer.encode( "utf-8", errors="replace" ) ).hexdigest() if academy_initial_raw_answer else "", "general_context_injected": False, "json_mode": True, "max_tokens": max( 128, min( int(payload.get("max_tokens", ACADEMY_PRACTICE_MAX_TOKENS)), ACADEMY_PRACTICE_MAX_TOKENS, ), ), "prompt_characters": len(prompt), "system_context_characters": len( ACADEMY_ROUND_SYSTEM_CONTEXTS[exam_round] ), "elapsed_seconds": elapsed, "blender_used": False, "answer": answer, }, ) return ( model, mode, clean_prompt, route_reason, max_tokens, blender_route, ) = resolve_request_route(payload, prompt) if blender_route is not None: guarded = clean_prompt system_context = blender_system_context(blender_route) else: guarded = guarded_prompt(clean_prompt, mode) system_context = build_prompt(clean_prompt) json_mode = bool(payload.get("json_mode", False)) start = time.time() ollama_data = ask_ollama( model, guarded, max_tokens, system_context=system_context, json_mode=json_mode, ) elapsed = round(time.time() - start, 2) message = ollama_data.get("message", {}) answer = message.get("content", "").strip() json_response( self, { "ok": True, "mode": mode, "model": model, "route_reason": route_reason, "blender_route": blender_route, "json_mode": json_mode, "elapsed_seconds": elapsed, "answer": answer, }, ) except AcademyStructuredOutputError as exc: json_response( self, { "ok": False, "error": str(exc), "retryable": True, "infrastructure_failure": True, "output_contract": ACADEMY_OUTPUT_CONTRACT, "exam_round": academy_exam_round, "raw_answer": academy_raw_answer, "raw_answer_sha256": hashlib.sha256( academy_raw_answer.encode("utf-8", errors="replace") ).hexdigest(), "raw_answer_audited": True, "model_retry_count": academy_model_retry_count, "model_retry_reason": academy_model_retry_reason, "initial_raw_answer": academy_initial_raw_answer, "initial_raw_answer_sha256": hashlib.sha256( academy_initial_raw_answer.encode( "utf-8", errors="replace" ) ).hexdigest() if academy_initial_raw_answer else "", }, status=502, ) except ValueError as exc: json_response(self, {"ok": False, "error": str(exc)}, status=400) except BlenderBrainContractError as exc: json_response( self, {"ok": False, "error": f"Blender routing blocked: {exc}"}, status=400, ) except urllib.error.HTTPError as exc: try: detail = exc.read().decode("utf-8", errors="replace") except Exception: detail = str(exc) json_response( self, { "ok": False, "error": f"Ollama HTTP error: {exc.code}", "detail": detail, }, status=500, ) except Exception as exc: if ( self.path == "/academy/practice/ask" and _academy_timeout_error(exc) ): json_response( self, _academy_timeout_payload( academy_exam_round, academy_timeout_seconds, raw_answer=academy_raw_answer, initial_raw_answer=academy_initial_raw_answer, model_retry_count=academy_model_retry_count, model_retry_reason=academy_model_retry_reason, ), status=504, ) return json_response(self, {"ok": False, "error": str(exc)}, status=500) def main(): host = bind_host() server = ThreadingHTTPServer((host, PORT), BrainHandler) print(f"SHIRE Brain API listening on http://{host}:{PORT}") print(f"Fast model: {FAST_MODEL}") print(f"Deep model: {DEEP_MODEL}") print(f"Ollama URL: {OLLAMA_URL}") academy_status = design_academy_service.status() print( "Design Academy: " f"available={academy_status.get('available')} " f"skills={academy_status.get('skills')} " f"cards={academy_status.get('knowledge_cards')}" ) server.serve_forever() if __name__ == "__main__": main()