--- /home/ray/ARMOR/apps/shire-mobile/limbforge-workspace/server.py	2026-08-06 16:33:50.696347212 +0800
+++ /tmp/shire-limbforge-edit-api.FUZG8C/server.py	2026-08-06 17:11:50.865731647 +0800
@@ -409,22 +409,27 @@
     return load_project(project_id)
 
 
-def save_measurement(
-    project_id: str,
-    measurement_key: str,
-    payload: dict,
-) -> dict:
-    project = load_project(project_id)
+# SHIRE-LIMBFORGE-EDIT-ANY-API-0001
+
+EDIT_AUDIT_ACTOR = "ray"
 
-    if project is None:
-        raise LookupError("Project not found.")
 
-    definition = MEASUREMENT_BY_KEY.get(measurement_key)
+def validate_measurement_payload(
+    project: dict,
+    measurement_key: str,
+    payload: dict,
+) -> tuple[float, str, str]:
+    definition = MEASUREMENT_BY_KEY.get(
+        measurement_key
+    )
 
     if definition is None:
         raise ValueError("Unknown measurement.")
 
-    if definition["oi_only"] and not project["osseointegration"]:
+    if (
+        definition["oi_only"]
+        and not project["osseointegration"]
+    ):
         raise ValueError(
             "This measurement is available only for "
             "osseointegration projects."
@@ -433,23 +438,467 @@
     try:
         value = float(payload.get("value_mm"))
     except (TypeError, ValueError):
-        raise ValueError("Enter a valid measurement in millimetres.")
+        raise ValueError(
+            "Enter a valid measurement in millimetres."
+        )
 
     if value <= 0 or value > 2000:
         raise ValueError(
-            "Measurement must be greater than zero and below 2000 mm."
+            "Measurement must be greater than zero "
+            "and below 2000 mm."
+        )
+
+    status = str(
+        payload.get("status", "completed")
+    ).strip()
+
+    if status not in {
+        "completed",
+        "specialist_review",
+    }:
+        raise ValueError(
+            "Invalid measurement status."
+        )
+
+    notes = str(
+        payload.get("notes", "")
+    ).strip()
+
+    return value, notes, status
+
+
+def require_change_reason(payload: dict) -> str:
+    reason = str(
+        payload.get("change_reason", "")
+    ).strip()
+
+    if not reason:
+        raise ValueError(
+            "Enter a reason for changing this measurement."
+        )
+
+    if len(reason) > 1000:
+        raise ValueError(
+            "The change reason must not exceed "
+            "1000 characters."
+        )
+
+    return reason
+
+
+def insert_snapshot_supersessions(
+    database: sqlite3.Connection,
+    project_id: str,
+    revision_id: str,
+    reason: str,
+    timestamp: str,
+    session_id: str | None = None,
+) -> int:
+    if session_id is None:
+        snapshots = database.execute(
+            """
+            SELECT id
+            FROM design_forge_measurement_snapshots
+            WHERE project_id = ?
+            """,
+            (project_id,),
+        ).fetchall()
+    else:
+        snapshots = database.execute(
+            """
+            SELECT id
+            FROM design_forge_measurement_snapshots
+            WHERE project_id = ?
+              AND (
+                    session_id = ?
+                    OR session_id IS NULL
+              )
+            """,
+            (
+                project_id,
+                session_id,
+            ),
+        ).fetchall()
+
+    inserted = 0
+
+    for snapshot in snapshots:
+        database.execute(
+            """
+            INSERT INTO
+                design_forge_snapshot_supersessions (
+                    id,
+                    snapshot_id,
+                    project_id,
+                    revision_id,
+                    reason,
+                    created_at
+                )
+            VALUES (?, ?, ?, ?, ?, ?)
+            """,
+            (
+                uuid.uuid4().hex,
+                snapshot["id"],
+                project_id,
+                revision_id,
+                reason,
+                timestamp,
+            ),
         )
 
-    status = str(payload.get("status", "completed"))
+        inserted += 1
+
+    return inserted
+
+
+def revision_to_dict(row: sqlite3.Row) -> dict:
+    revision = dict(row)
+
+    definition = MEASUREMENT_BY_KEY.get(
+        revision["measurement_key"]
+    )
+
+    revision["measurement_name"] = (
+        definition["name"]
+        if definition
+        else revision["measurement_key"]
+    )
+
+    revision["unit"] = "mm"
+
+    return revision
+
+
+def list_project_revisions(
+    project_id: str,
+) -> list[dict]:
+    if load_project(project_id) is None:
+        raise LookupError("Project not found.")
+
+    with DATABASE_LOCK, connect() as database:
+        rows = database.execute(
+            """
+            SELECT *
+            FROM measurement_revisions
+            WHERE project_id = ?
+            ORDER BY changed_at DESC, id DESC
+            """,
+            (project_id,),
+        ).fetchall()
+
+    return [
+        revision_to_dict(row)
+        for row in rows
+    ]
 
-    if status not in {"completed", "specialist_review"}:
-        raise ValueError("Invalid measurement status.")
 
+def list_session_revisions(
+    project_id: str,
+    session_id: str,
+) -> list[dict]:
+    session = load_measurement_session(
+        project_id,
+        session_id,
+    )
+
+    if session is None:
+        raise LookupError(
+            "Measurement session not found."
+        )
+
+    with DATABASE_LOCK, connect() as database:
+        rows = database.execute(
+            """
+            SELECT *
+            FROM measurement_revisions
+            WHERE project_id = ?
+              AND session_id = ?
+            ORDER BY changed_at DESC, id DESC
+            """,
+            (
+                project_id,
+                session_id,
+            ),
+        ).fetchall()
+
+    return [
+        revision_to_dict(row)
+        for row in rows
+    ]
+
+
+def session_to_dict(
+    row: sqlite3.Row,
+    measurements: list[sqlite3.Row] | None = None,
+) -> dict:
+    session = dict(row)
+
+    if measurements is not None:
+        session["measurements"] = {
+            item["measurement_key"]: {
+                "value_mm": item["value_mm"],
+                "notes": item["notes"],
+                "status": item["status"],
+                "updated_at": item["updated_at"],
+            }
+            for item in measurements
+        }
+
+    return session
+
+
+def list_measurement_sessions(
+    project_id: str,
+) -> list[dict]:
+    if load_project(project_id) is None:
+        raise LookupError("Project not found.")
+
+    with DATABASE_LOCK, connect() as database:
+        rows = database.execute(
+            """
+            SELECT
+                measurement_sessions.*,
+                (
+                    SELECT COUNT(*)
+                    FROM measurement_session_records
+                    WHERE
+                        measurement_session_records.session_id
+                        = measurement_sessions.id
+                ) AS measurement_count
+            FROM measurement_sessions
+            WHERE project_id = ?
+            ORDER BY created_at DESC, id DESC
+            """,
+            (project_id,),
+        ).fetchall()
+
+    return [
+        dict(row)
+        for row in rows
+    ]
+
+
+def load_measurement_session(
+    project_id: str,
+    session_id: str,
+) -> dict | None:
+    with DATABASE_LOCK, connect() as database:
+        session = database.execute(
+            """
+            SELECT *
+            FROM measurement_sessions
+            WHERE id = ?
+              AND project_id = ?
+            """,
+            (
+                session_id,
+                project_id,
+            ),
+        ).fetchone()
+
+        if session is None:
+            return None
+
+        measurements = database.execute(
+            """
+            SELECT *
+            FROM measurement_session_records
+            WHERE session_id = ?
+              AND project_id = ?
+            ORDER BY updated_at, measurement_key
+            """,
+            (
+                session_id,
+                project_id,
+            ),
+        ).fetchall()
+
+    return session_to_dict(
+        session,
+        measurements,
+    )
+
+
+def create_measurement_session(
+    project_id: str,
+    payload: dict,
+) -> dict:
+    if load_project(project_id) is None:
+        raise LookupError("Project not found.")
+
+    status = str(
+        payload.get("status", "active")
+    ).strip()
+
+    if status not in {
+        "active",
+        "completed",
+        "cancelled",
+    }:
+        raise ValueError(
+            "Invalid measurement-session status."
+        )
+
+    guide_version = str(
+        payload.get(
+            "guide_version",
+            "limbforge-anatomy-guides-v1",
+        )
+    ).strip()
+
+    if not guide_version:
+        raise ValueError(
+            "Measurement guide version is required."
+        )
+
+    if len(guide_version) > 200:
+        raise ValueError(
+            "Measurement guide version is too long."
+        )
+
+    session_id = uuid.uuid4().hex
     timestamp = utc_now()
 
+    completed_at = (
+        timestamp
+        if status == "completed"
+        else None
+    )
+
     with DATABASE_LOCK, connect() as database:
         database.execute(
             """
+            INSERT INTO measurement_sessions (
+                id,
+                project_id,
+                status,
+                guide_version,
+                started_at,
+                completed_at,
+                created_at,
+                updated_at
+            )
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            (
+                session_id,
+                project_id,
+                status,
+                guide_version,
+                timestamp,
+                completed_at,
+                timestamp,
+                timestamp,
+            ),
+        )
+
+    session = load_measurement_session(
+        project_id,
+        session_id,
+    )
+
+    if session is None:
+        raise RuntimeError(
+            "Measurement session could not be loaded."
+        )
+
+    return session
+
+
+def save_measurement(
+    project_id: str,
+    measurement_key: str,
+    payload: dict,
+) -> dict:
+    project = load_project(project_id)
+
+    if project is None:
+        raise LookupError("Project not found.")
+
+    value, notes, status = (
+        validate_measurement_payload(
+            project,
+            measurement_key,
+            payload,
+        )
+    )
+
+    timestamp = utc_now()
+
+    with DATABASE_LOCK, connect() as database:
+        existing = database.execute(
+            """
+            SELECT *
+            FROM measurements
+            WHERE project_id = ?
+              AND measurement_key = ?
+            """,
+            (
+                project_id,
+                measurement_key,
+            ),
+        ).fetchone()
+
+        changed = (
+            existing is not None
+            and (
+                float(existing["value_mm"]) != value
+                or existing["notes"] != notes
+                or existing["status"] != status
+            )
+        )
+
+        if changed:
+            reason = require_change_reason(payload)
+            revision_id = uuid.uuid4().hex
+
+            database.execute(
+                """
+                INSERT INTO measurement_revisions (
+                    id,
+                    project_id,
+                    session_id,
+                    measurement_key,
+                    previous_value_mm,
+                    new_value_mm,
+                    previous_notes,
+                    new_notes,
+                    previous_status,
+                    new_status,
+                    change_reason,
+                    changed_by,
+                    changed_at
+                )
+                VALUES (
+                    ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
+                )
+                """,
+                (
+                    revision_id,
+                    project_id,
+                    measurement_key,
+                    existing["value_mm"],
+                    value,
+                    existing["notes"],
+                    notes,
+                    existing["status"],
+                    status,
+                    reason,
+                    EDIT_AUDIT_ACTOR,
+                    timestamp,
+                ),
+            )
+
+            insert_snapshot_supersessions(
+                database,
+                project_id,
+                revision_id,
+                reason,
+                timestamp,
+            )
+
+        database.execute(
+            """
             INSERT INTO measurements (
                 project_id,
                 measurement_key,
@@ -470,7 +919,7 @@
                 project_id,
                 measurement_key,
                 value,
-                str(payload.get("notes", "")).strip(),
+                notes,
                 status,
                 timestamp,
             ),
@@ -482,12 +931,197 @@
             SET updated_at = ?
             WHERE id = ?
             """,
-            (timestamp, project_id),
+            (
+                timestamp,
+                project_id,
+            ),
         )
 
     return load_project(project_id)
 
 
+def save_session_measurement(
+    project_id: str,
+    session_id: str,
+    measurement_key: str,
+    payload: dict,
+) -> dict:
+    project = load_project(project_id)
+
+    if project is None:
+        raise LookupError("Project not found.")
+
+    session = load_measurement_session(
+        project_id,
+        session_id,
+    )
+
+    if session is None:
+        raise LookupError(
+            "Measurement session not found."
+        )
+
+    if session["status"] == "cancelled":
+        raise ValueError(
+            "Cancelled measurement sessions "
+            "cannot be edited."
+        )
+
+    value, notes, status = (
+        validate_measurement_payload(
+            project,
+            measurement_key,
+            payload,
+        )
+    )
+
+    timestamp = utc_now()
+
+    with DATABASE_LOCK, connect() as database:
+        existing = database.execute(
+            """
+            SELECT *
+            FROM measurement_session_records
+            WHERE session_id = ?
+              AND project_id = ?
+              AND measurement_key = ?
+            """,
+            (
+                session_id,
+                project_id,
+                measurement_key,
+            ),
+        ).fetchone()
+
+        changed = (
+            existing is not None
+            and (
+                float(existing["value_mm"]) != value
+                or existing["notes"] != notes
+                or existing["status"] != status
+            )
+        )
+
+        if changed:
+            reason = require_change_reason(payload)
+            revision_id = uuid.uuid4().hex
+
+            database.execute(
+                """
+                INSERT INTO measurement_revisions (
+                    id,
+                    project_id,
+                    session_id,
+                    measurement_key,
+                    previous_value_mm,
+                    new_value_mm,
+                    previous_notes,
+                    new_notes,
+                    previous_status,
+                    new_status,
+                    change_reason,
+                    changed_by,
+                    changed_at
+                )
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    revision_id,
+                    project_id,
+                    session_id,
+                    measurement_key,
+                    existing["value_mm"],
+                    value,
+                    existing["notes"],
+                    notes,
+                    existing["status"],
+                    status,
+                    reason,
+                    EDIT_AUDIT_ACTOR,
+                    timestamp,
+                ),
+            )
+
+            insert_snapshot_supersessions(
+                database,
+                project_id,
+                revision_id,
+                reason,
+                timestamp,
+                session_id,
+            )
+
+        database.execute(
+            """
+            INSERT INTO measurement_session_records (
+                session_id,
+                project_id,
+                measurement_key,
+                value_mm,
+                notes,
+                status,
+                updated_at
+            )
+            VALUES (?, ?, ?, ?, ?, ?, ?)
+            ON CONFLICT(session_id, measurement_key)
+            DO UPDATE SET
+                value_mm = excluded.value_mm,
+                notes = excluded.notes,
+                status = excluded.status,
+                updated_at = excluded.updated_at
+            """,
+            (
+                session_id,
+                project_id,
+                measurement_key,
+                value,
+                notes,
+                status,
+                timestamp,
+            ),
+        )
+
+        database.execute(
+            """
+            UPDATE measurement_sessions
+            SET updated_at = ?
+            WHERE id = ?
+              AND project_id = ?
+            """,
+            (
+                timestamp,
+                session_id,
+                project_id,
+            ),
+        )
+
+        database.execute(
+            """
+            UPDATE projects
+            SET updated_at = ?
+            WHERE id = ?
+            """,
+            (
+                timestamp,
+                project_id,
+            ),
+        )
+
+    updated = load_measurement_session(
+        project_id,
+        session_id,
+    )
+
+    if updated is None:
+        raise RuntimeError(
+            "Updated measurement session "
+            "could not be loaded."
+        )
+
+    return updated
+
+
+
 def delete_measurement(
     project_id: str,
     measurement_key: str,
@@ -595,6 +1229,7 @@
                         "status": "ok",
                         "service": "limbforge-shared-workspace",
                         "database": str(DATABASE),
+                        "edit_any_measurement_api": True,
                     }
                 )
                 return
@@ -614,6 +1249,66 @@
             ]
 
             if (
+                len(parts) == 6
+                and parts[0] == "api"
+                and parts[1] == "projects"
+                and parts[3] == "sessions"
+                and parts[5] == "revisions"
+            ):
+                self.send_json(
+                    list_session_revisions(
+                        parts[2],
+                        parts[4],
+                    )
+                )
+                return
+
+            if (
+                len(parts) == 5
+                and parts[0] == "api"
+                and parts[1] == "projects"
+                and parts[3] == "sessions"
+            ):
+                session = load_measurement_session(
+                    parts[2],
+                    parts[4],
+                )
+
+                if session is None:
+                    raise LookupError(
+                        "Measurement session not found."
+                    )
+
+                self.send_json(session)
+                return
+
+            if (
+                len(parts) == 4
+                and parts[0] == "api"
+                and parts[1] == "projects"
+                and parts[3] == "sessions"
+            ):
+                self.send_json(
+                    list_measurement_sessions(
+                        parts[2]
+                    )
+                )
+                return
+
+            if (
+                len(parts) == 4
+                and parts[0] == "api"
+                and parts[1] == "projects"
+                and parts[3] == "revisions"
+            ):
+                self.send_json(
+                    list_project_revisions(
+                        parts[2]
+                    )
+                )
+                return
+
+            if (
                 len(parts) == 3
                 and parts[0] == "api"
                 and parts[1] == "projects"
@@ -621,7 +1316,9 @@
                 project = load_project(parts[2])
 
                 if project is None:
-                    raise LookupError("Project not found.")
+                    raise LookupError(
+                        "Project not found."
+                    )
 
                 self.send_json(project)
                 return
@@ -631,9 +1328,16 @@
         except Exception as error:
             self.handle_api_error(error)
 
+
     def do_POST(self) -> None:
         path = urlparse(self.path).path
 
+        parts = [
+            unquote(part)
+            for part in path.strip("/").split("/")
+            if part
+        ]
+
         try:
             if path == "/api/projects":
                 self.send_json(
@@ -642,6 +1346,21 @@
                 )
                 return
 
+            if (
+                len(parts) == 4
+                and parts[0] == "api"
+                and parts[1] == "projects"
+                and parts[3] == "sessions"
+            ):
+                self.send_json(
+                    create_measurement_session(
+                        parts[2],
+                        self.read_json(),
+                    ),
+                    HTTPStatus.CREATED,
+                )
+                return
+
             self.send_json(
                 {"error": "Route not found."},
                 HTTPStatus.NOT_FOUND,
@@ -650,8 +1369,10 @@
         except Exception as error:
             self.handle_api_error(error)
 
+
     def do_PUT(self) -> None:
         path = urlparse(self.path).path
+
         parts = [
             unquote(part)
             for part in path.strip("/").split("/")
@@ -687,6 +1408,23 @@
                 )
                 return
 
+            if (
+                len(parts) == 7
+                and parts[0] == "api"
+                and parts[1] == "projects"
+                and parts[3] == "sessions"
+                and parts[5] == "measurements"
+            ):
+                self.send_json(
+                    save_session_measurement(
+                        parts[2],
+                        parts[4],
+                        parts[6],
+                        self.read_json(),
+                    )
+                )
+                return
+
             self.send_json(
                 {"error": "Route not found."},
                 HTTPStatus.NOT_FOUND,
@@ -695,6 +1433,7 @@
         except Exception as error:
             self.handle_api_error(error)
 
+
     def do_DELETE(self) -> None:
         path = urlparse(self.path).path
         parts = [
