--- /home/ray/ARMOR/apps/shire-mobile/limbforge-workspace/server.py 2026-08-05 11:53:39.537552790 +0800 +++ /tmp/shire-limbforge-oi.f8UkEb/stage/server.py 2026-08-06 16:33:49.911847910 +0800 @@ -522,6 +522,376 @@ return load_project(project_id) + +# SHIRE-LIMBFORGE-OI-FOUNDATION-0001 +# +# Explicit OI hardware type validation and persistent +# measurement-session / immutable snapshot foundations. + +import sqlite3 as _shire_oi_sqlite3 +from pathlib import Path as _ShireOiPath + +_SHIRE_OI_DB = ( + _ShireOiPath(__file__).resolve().parent + / "data" + / "limbforge.sqlite3" +) + +_SHIRE_OI_TYPES = { + "none", + "dual_cone", + "legacy_abutment", +} + + +def _shire_oi_bool(value) -> bool: + if isinstance(value, bool): + return value + + if isinstance(value, (int, float)): + return value != 0 + + if isinstance(value, str): + return value.strip().lower() in { + "1", + "true", + "yes", + "on", + } + + return bool(value) + + +def _shire_validate_oi_type( + osseointegration, + raw_value, +) -> str: + enabled = _shire_oi_bool(osseointegration) + + value = ( + "none" + if raw_value is None + else str(raw_value).strip().lower() + ) + + if not value: + value = "none" + + if value not in _SHIRE_OI_TYPES: + raise ValueError( + "oi_type must be none, dual_cone, " + "or legacy_abutment." + ) + + if enabled: + if value not in { + "dual_cone", + "legacy_abutment", + }: + raise ValueError( + "Osseointegration projects must select " + "Dual Cone or Legacy Abutment." + ) + + return value + + if value != "none": + raise ValueError( + "Conventional prosthesis projects must use " + "oi_type none." + ) + + return "none" + + +def _shire_apply_oi_schema() -> None: + with _shire_oi_sqlite3.connect( + _SHIRE_OI_DB + ) as connection: + connection.execute( + "PRAGMA foreign_keys = ON" + ) + + columns = { + row[1] + for row in connection.execute( + "PRAGMA table_info(projects)" + ) + } + + if "oi_type" not in columns: + connection.execute( + """ + ALTER TABLE projects + ADD COLUMN oi_type TEXT + NOT NULL + DEFAULT 'none' + CHECK ( + oi_type IN ( + 'none', + 'dual_cone', + 'legacy_abutment' + ) + ) + """ + ) + + connection.execute( + """ + UPDATE projects + SET oi_type = 'none' + WHERE oi_type IS NULL + OR TRIM(oi_type) = '' + """ + ) + + connection.execute( + """ + CREATE TABLE IF NOT EXISTS + measurement_sessions ( + id TEXT PRIMARY KEY, + project_id TEXT NOT NULL, + status TEXT NOT NULL + CHECK ( + status IN ( + 'active', + 'completed', + 'cancelled' + ) + ), + guide_version TEXT NOT NULL, + started_at TEXT NOT NULL, + completed_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + FOREIGN KEY (project_id) + REFERENCES projects(id) + ON DELETE CASCADE + ) + """ + ) + + connection.execute( + """ + CREATE INDEX IF NOT EXISTS + measurement_sessions_project_idx + ON measurement_sessions(project_id) + """ + ) + + connection.execute( + """ + CREATE TABLE IF NOT EXISTS + design_forge_measurement_snapshots ( + id TEXT PRIMARY KEY, + project_id TEXT NOT NULL, + session_id TEXT, + snapshot_json TEXT NOT NULL, + snapshot_sha256 TEXT NOT NULL, + created_at TEXT NOT NULL, + immutable INTEGER NOT NULL + DEFAULT 1 + CHECK (immutable = 1), + FOREIGN KEY (project_id) + REFERENCES projects(id) + ON DELETE RESTRICT, + FOREIGN KEY (session_id) + REFERENCES measurement_sessions(id) + ON DELETE RESTRICT + ) + """ + ) + + connection.execute( + """ + CREATE INDEX IF NOT EXISTS + design_forge_snapshots_project_idx + ON design_forge_measurement_snapshots( + project_id + ) + """ + ) + + connection.execute( + """ + CREATE UNIQUE INDEX IF NOT EXISTS + design_forge_snapshots_sha_idx + ON design_forge_measurement_snapshots( + snapshot_sha256 + ) + """ + ) + + connection.execute( + """ + CREATE TRIGGER IF NOT EXISTS + design_forge_snapshots_no_update + BEFORE UPDATE + ON design_forge_measurement_snapshots + BEGIN + SELECT RAISE( + ABORT, + 'Design Forge measurement snapshots ' + 'are immutable' + ); + END + """ + ) + + connection.execute( + """ + CREATE TRIGGER IF NOT EXISTS + design_forge_snapshots_no_delete + BEFORE DELETE + ON design_forge_measurement_snapshots + BEGIN + SELECT RAISE( + ABORT, + 'Design Forge measurement snapshots ' + 'cannot be deleted' + ); + END + """ + ) + + +_SHIRE_ORIGINAL_INITIALISE_DATABASE = initialise_database +_SHIRE_ORIGINAL_CREATE_PROJECT = create_project +_SHIRE_ORIGINAL_UPDATE_PROJECT = update_project + + +def initialise_database() -> None: + _SHIRE_ORIGINAL_INITIALISE_DATABASE() + _shire_apply_oi_schema() + + +def _shire_save_project_oi_type( + project_id: str, + oi_type: str, +) -> dict: + with _shire_oi_sqlite3.connect( + _SHIRE_OI_DB + ) as connection: + result = connection.execute( + """ + UPDATE projects + SET oi_type = ? + WHERE id = ? + """, + ( + oi_type, + project_id, + ), + ) + + if result.rowcount != 1: + raise RuntimeError( + "The project OI type was not saved." + ) + + project = load_project(project_id) + + if project is None: + raise RuntimeError( + "The project could not be reloaded." + ) + + if "oi_type" not in project: + project["oi_type"] = oi_type + + return project + + +def create_project(payload: dict) -> dict: + safe_payload = dict(payload) + + enabled = _shire_oi_bool( + safe_payload.get( + "osseointegration", + False, + ) + ) + + oi_type = _shire_validate_oi_type( + enabled, + safe_payload.get("oi_type"), + ) + + safe_payload["osseointegration"] = enabled + + project = _SHIRE_ORIGINAL_CREATE_PROJECT( + safe_payload + ) + + try: + return _shire_save_project_oi_type( + project["id"], + oi_type, + ) + except Exception: + with _shire_oi_sqlite3.connect( + _SHIRE_OI_DB + ) as connection: + connection.execute( + "DELETE FROM projects WHERE id = ?", + (project["id"],), + ) + + raise + + +def update_project( + project_id: str, + payload: dict, +) -> dict: + existing = load_project(project_id) + + if existing is None: + raise ValueError("Project not found.") + + safe_payload = dict(payload) + + enabled = _shire_oi_bool( + safe_payload.get( + "osseointegration", + existing.get( + "osseointegration", + False, + ), + ) + ) + + current_type = existing.get( + "oi_type", + "none", + ) + + requested_type = safe_payload.get( + "oi_type", + current_type, + ) + + if not enabled: + requested_type = "none" + + oi_type = _shire_validate_oi_type( + enabled, + requested_type, + ) + + safe_payload["osseointegration"] = enabled + + project = _SHIRE_ORIGINAL_UPDATE_PROJECT( + project_id, + safe_payload, + ) + + return _shire_save_project_oi_type( + project["id"], + oi_type, + ) + + class LimbForgeHandler(BaseHTTPRequestHandler): server_version = "SHiRE-LimbForge/0.1"