--- /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"
 
