"""SZL Vertical Services — six governed engines behind one operational fabric. The service combines deterministic vertical calculations, canonical formula bindings, Living Anatomy contracts, session-scoped Second-Brain memory, bounded official-source connectors, source-bound builds, model and kernel routing, hash-addressed receipts, and six domain-native command experiences. Killinchu is the single public cyber-physical resilience and maritime product. The legacy ``/vessels`` route remains only as a compatibility surface. Sentra is an independently testable capability plane behind Killinchu ``/defend``; Aegis is a portfolio label, IMMUNE remains migration-gated, and neither is another public Space. PURIQ resolves to Finance without duplicate execution authority. """ from __future__ import annotations import html import json from pathlib import Path from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse, JSONResponse from szl_verticals.core import ( ENGINES, RUNTIME_SOURCE_IDENTITY_FIELDS, SOURCE_REPOSITORY, VERSION, build_info as _build_info, ) from szl_verticals.counsel import counsel from szl_verticals.finance import finance from szl_verticals.frontier import frontier from szl_verticals.http_bounds import BodyBoundError, error_payload, maybe_parse_json, read_bounded_body from szl_verticals.intelligence import intelligence from szl_verticals.killinchu import killinchu from szl_verticals.killinchu_runtime_contract import ( architecture as _szl_killinchu_architecture, compatibility_headers as _szl_killinchu_headers, lobe as _szl_killinchu_lobe, ) from szl_verticals.lyte import lyte from szl_verticals.operational import STORE, operational, vertical_readiness from szl_verticals.presentation import NATIVE_ICON_HTML, NATIVE_INTERFACE_CSS, NATIVE_TOKENS_CSS from szl_verticals.sentra import SENTRA_KEY_SOURCE, sentra from szl_verticals.showcase import showcase from szl_verticals.terra import terra from szl_verticals.vessels import vessels app = FastAPI( title="SZL Vertical Services", version=VERSION, description=( "Six governed Python engines with Living Anatomy, formula bindings, " "Second-Brain memory, bounded official-source connectors, source-bound " "model and kernel routing, Hatun review, and differentiated command rooms." ), ) @app.middleware("http") async def inbound_body_bounds(request: Request, call_next): """Cap inbound bytes and reject illegal JSON before route parsers run.""" if request.method in {"POST", "PUT", "PATCH"}: try: body = await read_bounded_body(request) maybe_parse_json(request.headers, body) except BodyBoundError as exc: return JSONResponse(error_payload(exc), status_code=exc.status_code) request._body = body return await call_next(request) @app.middleware("http") async def response_hardening(request: Request, call_next): response = await call_next(request) response.headers["X-Content-Type-Options"] = "nosniff" response.headers["Referrer-Policy"] = "no-referrer" response.headers["Permissions-Policy"] = "camera=(), microphone=(), geolocation=()" response.headers["Cross-Origin-Resource-Policy"] = "same-site" if request.url.path != "/": response.headers["Cache-Control"] = "no-store" return response @app.middleware("http") async def killinchu_product_identity(request: Request, call_next): """Label compatibility responses without changing domain payloads.""" response = await call_next(request) for key, value in _szl_killinchu_headers(request.url.path).items(): response.headers[key] = value return response for router in ( sentra, lyte, vessels, killinchu, finance, terra, counsel, operational, frontier, intelligence, showcase, ): app.include_router(router) CATALOG = { "sentra": { "purpose": ( "Killinchu Defend capability: evidence-linked attack paths, " "deny-by-default gates, bounded response review, and signed verdicts" ), "public_home": "SZLHOLDINGS/killinchu", "public_route": "https://szlholdings-killinchu.hf.space/defend", "product_state": "CAPABILITY_PLANE", "experience": "/experience/defend", "intelligence": "/intelligence/defend", "aliases": ["aegis", "defend"], }, "lyte": { "purpose": ( "business-observability metrics, delivery health, percentiles, " "drift, and economic-outcome context" ), "public_home": "SZLHOLDINGS/lyte", "experience": "/experience/lyte", "intelligence": "/intelligence/lyte", "aliases": ["business-observability"], }, "killinchu": { "purpose": "defense policy and maritime track-risk command", "public_home": "SZLHOLDINGS/killinchu", "experience": "/experience/killinchu", "intelligence": "/intelligence/killinchu", "status": "CANONICAL", "vessels": "CONSOLIDATED", "aliases": ["vessels"], }, "finance": { "purpose": ( "PURIQ market-series analytics, SEC evidence, public rates, crypto " "spot references, and read-only prediction-market intelligence" ), "public_home": "SZLHOLDINGS/finance", "experience": "/experience/puriq", "intelligence": "/intelligence/puriq", "aliases": ["puriq", "markets"], }, "terra": { "purpose": ( "property calculations, parcel evidence, and public building-" "condition research" ), "public_home": "SZLHOLDINGS/terra", "experience": "/experience/terra", "intelligence": "/intelligence/terra", "aliases": ["real-estate"], }, "counsel": { "purpose": "matters, obligations, public legal authority, and receipt chains", "public_home": "SZLHOLDINGS/counsel", "experience": "/experience/prism", "intelligence": "/intelligence/prism", "aliases": ["prism"], }, } def _source_ready(build: dict) -> bool: return ( build["build"]["state"] == "OBSERVED" and build["source_binding"]["bindings_agree"] is True ) def _health_truth_label(source_ready: bool) -> str: return "MEASURED" if source_ready else "UNAVAILABLE" @app.get("/healthz") def root_health() -> dict: build = _build_info() source_ready = _source_ready(build) return { "ok": source_ready, "status": "ok" if source_ready else "source-unavailable", "service": "szl-vertical-services", "version": VERSION, "engines": list(ENGINES), "routes": {engine: f"/{engine}/healthz" for engine in ENGINES}, "compatibility_routes": { "/vessels": { "status": "DEPRECATED_COMPATIBILITY", "canonical": "/killinchu", }, "/api/verticals/aegis": { "status": "PORTFOLIO_ALIAS", "canonical": "/api/verticals/sentra", "public_route": "https://szlholdings-killinchu.hf.space/defend", }, "/api/verticals/defend": { "status": "CAPABILITY_ALIAS", "canonical": "/api/verticals/sentra", "public_route": "https://szlholdings-killinchu.hf.space/defend", }, "/api/verticals/puriq": { "status": "PRODUCT_ALIAS", "canonical": "/api/verticals/finance", }, }, "build": build["build"], "sentra_signing_key_source": SENTRA_KEY_SOURCE, "state": { "business_working_sets": "SESSION_ISOLATED_PROCESS_MEMORY", "connector_observations": STORE.status(), }, "session_header": "X-SZL-Session", "official_source_connectors_wired": True, "vertical_intelligence_wired": True, "model_provider_invocation_fail_closed": True, "caller_supplied_model_endpoints_allowed": False, "hatun_can_authorize": False, "effectors_enabled": False, "truth_label": _health_truth_label(source_ready), "source": SOURCE_REPOSITORY, **{field: build[field] for field in RUNTIME_SOURCE_IDENTITY_FIELDS}, } @app.get("/readyz") def readiness() -> JSONResponse: verticals = {engine: vertical_readiness(engine) for engine in ENGINES} build = _build_info() ready = all(item["ready"] for item in verticals.values()) and _source_ready(build) payload = { "ready": ready, "service": "szl-vertical-services", "version": VERSION, "verticals": { engine: { "ready": item["ready"], "status": item["status"], "requirements": item["requirements"], "live_data": item["live_data"], } for engine, item in verticals.items() }, "build": build["build"], **{field: build[field] for field in RUNTIME_SOURCE_IDENTITY_FIELDS}, "store": STORE.status(), "intelligence_plan_ready": True, "inference_requires_operator_model_binding": True, "truth_label": _health_truth_label(ready), } return JSONResponse(payload, status_code=200 if ready else 503) @app.get("/api/build-info") def build_info() -> dict: return _build_info() @app.get("/api/source") @app.get("/.well-known/szl-source.json") def source_document() -> dict: return _build_info() @app.get("/api/catalog") def catalog() -> dict: return { "service": "szl-vertical-services", "version": VERSION, "engines": CATALOG, "product_authority": _szl_killinchu_architecture(), "vessels_independent_vertical": False, "vessels_canonical_home": "SZLHOLDINGS/killinchu", "aegis_canonical_runtime": "killinchu:defend", "sentra_independent_public_vertical": False, "sentra_public_route": "https://szlholdings-killinchu.hf.space/defend", "immune_canonical_runtime": "MIGRATION_REQUIRED", "puriq_canonical_runtime": "finance", "operational_fabric": { "catalog": "/api/verticals", "frontier": "/api/verticals/{vertical}/frontier", "experience": "/experience/{vertical}", "intelligence_room": "/intelligence/{vertical}", "intelligence_profile": "/api/verticals/{vertical}/intelligence", "intelligence_plan": "/api/verticals/{vertical}/intelligence/plan", "intelligence_invoke": "/api/verticals/{vertical}/intelligence/invoke", "anatomy": "/api/verticals/{vertical}/anatomy", "formulas": "/api/verticals/{vertical}/formulas", "connectors": "/api/verticals/{vertical}/connectors", "second_brain": "/api/verticals/{vertical}/second-brain", "hatun_review": "/api/verticals/{vertical}/hatun/evaluate", "readiness": "/api/verticals/{vertical}/readyz", "fetch": "/api/verticals/{vertical}/connectors/{connector_id}/fetch", }, "state": { "business_working_sets": "SESSION_ISOLATED_PROCESS_MEMORY", "connector_observations": STORE.status(), }, "session_header": "X-SZL-Session", "official_source_connectors_wired": True, "live_observations_require_explicit_fetch": True, "caller_supplied_urls_allowed": False, "caller_supplied_model_endpoints_allowed": False, "effectors_enabled": False, "truth_label": "MEASURED", "fashion_rule": "take the job, never proprietary code", "fashion": "/api/fashion", } def _fashion_contract_path() -> Path: here = Path(__file__).resolve() candidates = ( here.parent / "contracts" / "fashion-lineage.v1.json", here.parents[1] / "contracts" / "fashion-lineage.v1.json", ) for candidate in candidates: if candidate.is_file(): return candidate raise FileNotFoundError("fashion-lineage.v1.json missing from image and repo") _FASHION_PATH = _fashion_contract_path() @app.get("/api/fashion") def fashion_lineage() -> dict: payload = json.loads(_FASHION_PATH.read_text(encoding="utf-8")) payload["served_from"] = "contracts/fashion-lineage.v1.json" return payload def _landing_page() -> str: cards = [] names = { "sentra": "Killinchu Defend", "lyte": "Lyte", "killinchu": "Killinchu", "finance": "PURIQ Finance", "terra": "Terra", "counsel": "Prism Counsel", } for engine in ENGINES: info = CATALOG[engine] badge = ( " · VESSELS CONSOLIDATED HERE" if engine == "killinchu" else " · CAPABILITY PLANE INSIDE KILLINCHU" if engine == "sentra" else "" ) aliases = " · ".join(info.get("aliases", [])) cards.append( f"""
{html.escape(engine.upper())}{badge}

{html.escape(names[engine])}

{html.escape(info['purpose'])}

{html.escape(aliases)}
""" ) build = _build_info() revision = build["build"]["revision"] revision_short = revision[:12] if revision != "UNAVAILABLE" else revision store = STORE.status() source_ready = _source_ready(build) contract_chip = "REACHABLE" if source_ready else "UNAVAILABLE" return f""" SZL Vertical Services
{NATIVE_ICON_HTML}SZL / VERTICAL SERVICES V{VERSION}

Six engines. One second brain.

Explore domain calculations, public-source evidence and governed research. Open a workspace, inspect its contract, and check the source and readiness behind each result.

{contract_chip} Python runtime contractsource {html.escape(revision_short)}store {html.escape(store['durability'])}3 model routes6 kernel contractseffectors disabled
{''.join(cards)}
Operational boundary: official-source connectors are fixed and bounded. Connector observations are hash-addressed and stored under a hashed session scope. Model invocation remains unavailable until an operator binds a fixed allowlisted endpoint, credential, protocol, and exact declared revision. Hatun can recommend review or abstention only. NOAA AIS is historical official planning data—not represented as a live vessel feed. Trading, legal advice, cyber effectors, person-level prospecting, and unattended consequential actions remain disabled.
""" @app.get("/killinchu/architecture", tags=["Killinchu"]) def killinchu_architecture() -> dict: """Read the canonical source contract; reachability is not model truth.""" return _szl_killinchu_architecture() @app.get("/killinchu/aegis/healthz", tags=["Killinchu"]) def killinchu_aegis_lobe() -> dict: return _szl_killinchu_lobe("aegis") @app.get("/killinchu/vessels/healthz", tags=["Killinchu"]) def killinchu_vessels_lobe() -> dict: return _szl_killinchu_lobe("vessels") @app.get("/", response_class=HTMLResponse) def root() -> HTMLResponse: return HTMLResponse(_landing_page())