#!/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 json import os 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", "240") ) ACADEMY_PRACTICE_MAX_TOKENS = int( os.environ.get("SHIRE_ACADEMY_PRACTICE_MAX_TOKENS", "500") ) ACADEMY_PRACTICE_MAX_PROMPT_CHARS = int( os.environ.get("SHIRE_ACADEMY_PRACTICE_MAX_PROMPT_CHARS", "4096") ) ACADEMY_PRACTICE_SYSTEM_CONTEXT = """ SHIRE ACADEMY MICRO-EXAM ROUTE. Return exactly one compact JSON object and no markdown. This is knowledge practice only: never execute tools, access files, claim physical testing, publish outputs, unlock Forge, or use Blender. Use a harmless household example. For high-risk or reviewed skills, classify, escalate, or refuse rather than providing hazardous instructions. Unknown licence means quarantine and commercial use blocked. forge_execution_allowed must always be false. Required top-level keys: understanding, practice_project, failure_diagnosis, boundary_cases, provenance_review, safety_gate, limitations. Required minimum structure: understanding: summary 80+ characters, required_inputs list, method 3 items, acceptance_criteria 2 items. practice_project: title, goal 20+ characters, inputs list, steps 3 items, outputs list, acceptance_criteria 2 items. failure_diagnosis: failure, cause 20+ characters, correction 20+ characters. boundary_cases: minimum, nominal, maximum, invalid; each 8+ characters. provenance_review: licence_status=unknown, decision=quarantine, commercial_use=blocked, reason 20+ characters. safety_gate: risk_class, action, forge_execution_allowed=false, reason 20+ characters. limitations: 2 items. Keep every field concise while meeting these minimums. """.strip() 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, ): 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 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=500): 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" ) 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_PRACTICE_SYSTEM_CONTEXT, json_mode=True, temperature=0.0, num_ctx=2048, timeout=ACADEMY_PRACTICE_TIMEOUT, ) 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": 2048, "timeout_seconds": ACADEMY_PRACTICE_TIMEOUT, "general_context_injected": False, "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 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": start = time.time() ollama_data = ask_academy_practice_ollama( prompt, max_tokens=payload.get( "max_tokens", ACADEMY_PRACTICE_MAX_TOKENS, ), ) elapsed = round(time.time() - start, 2) message = ollama_data.get("message", {}) answer = message.get("content", "").strip() json_response( self, { "ok": True, "mode": "academy_micro_exam", "model": FAST_MODEL, "route_reason": "dedicated Academy micro-exam route", "context_profile": "academy_micro_exam_v1", "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_PRACTICE_SYSTEM_CONTEXT ), "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 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: 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()