feat(lab): separate evidence catalogs and record methods

src/k1link/sessions/store.py

@@ -11,7 +11,7 @@ import threading
 from collections.abc import Iterator
 from contextlib import contextmanager
 from pathlib import Path
-from typing import Any, cast
+from typing import Any, Literal, cast
 from uuid import uuid4
 
 from k1link.artifacts import utc_now_iso
@@ -45,6 +45,8 @@ LAB_ARCHIVE_ID = "missioncore.lab-instances"
 LAB_ORIGIN = "missioncore.lab-instance/v1"
 LAB_ID_PATTERN = re.compile(r"^LAB [A-Z][A-Z0-9._-]{0,31}$")
 SHA256_PATTERN = re.compile(r"^[a-f0-9]{64}$")
+LAB_METHOD_SCHEMA = "missioncore.laboratory-method/v1"
+SessionScope = Literal["all", "source", "laboratory"]
 
 SCHEMA_SQL = """
 CREATE TABLE IF NOT EXISTS observation_sessions (
@@ -205,27 +207,51 @@ class SessionStore:
             connection.commit()
         return tuple(imported)
 
-    def list_recent(self, *, limit: int = 20, cursor: str | None = None) -> SessionPage:
+    def list_recent(
+        self,
+        *,
+        limit: int = 20,
+        cursor: str | None = None,
+        scope: SessionScope = "all",
+    ) -> SessionPage:
         if not 1 <= limit <= 100:
             raise ValueError("limit must be within 1..100")
+        scope_clause = {
+            "all": "1 = 1",
+            "source": (
+                "NOT EXISTS (SELECT 1 FROM observation_lab_instances AS lab "
+                "WHERE lab.session_id = sessions.session_id)"
+            ),
+            "laboratory": (
+                "EXISTS (SELECT 1 FROM observation_lab_instances AS lab "
+                "WHERE lab.session_id = sessions.session_id)"
+            ),
+        }.get(scope)
+        if scope_clause is None:
+            raise ValueError("scope must be all, source, or laboratory")
         parameters: list[object] = []
-        where = ""
+        where = f"WHERE {scope_clause}"  # noqa: S608 - closed static scope clauses
         with self._connect() as connection:
             if cursor is not None:
                 _validate_identifier(cursor, "session cursor")
                 cursor_row = connection.execute(
-                    "SELECT started_at_utc, session_id FROM observation_sessions "
-                    "WHERE session_id = ?",
+                    "SELECT sessions.started_at_utc, sessions.session_id "
+                    "FROM observation_sessions AS sessions "
+                    f"WHERE {scope_clause} AND sessions.session_id = ?",  # noqa: S608
                     (cursor,),
                 ).fetchone()
                 if cursor_row is None:
                     raise SessionNotFoundError("observation session cursor was not found")
-                where = "WHERE (COALESCE(started_at_utc, ''), session_id) < (COALESCE(?, ''), ?)"
+                where += (
+                    " AND (COALESCE(sessions.started_at_utc, ''), sessions.session_id) "
+                    "< (COALESCE(?, ''), ?)"
+                )
                 parameters.extend((cursor_row["started_at_utc"], cursor_row["session_id"]))
             parameters.append(limit + 1)
             rows = connection.execute(
-                f"SELECT * FROM observation_sessions {where} "  # noqa: S608 - static clause
-                "ORDER BY COALESCE(started_at_utc, '') DESC, session_id DESC LIMIT ?",
+                f"SELECT sessions.* FROM observation_sessions AS sessions {where} "  # noqa: S608
+                "ORDER BY COALESCE(sessions.started_at_utc, '') DESC, "
+                "sessions.session_id DESC LIMIT ?",
                 parameters,
             ).fetchall()
             lab_rows = (
@@ -366,7 +392,9 @@ class SessionStore:
             or duration_seconds <= 0
         ):
             raise ValueError("LAB duration must be a positive finite value")
-        serialized_provenance = _serialize_provenance(provenance or {})
+        normalized_provenance = provenance or {}
+        _validate_lab_method(normalized_provenance)
+        serialized_provenance = _serialize_provenance(normalized_provenance)
         published_at = utc_now_iso()
 
         with self._lock, self._connect() as connection:
@@ -1078,6 +1106,55 @@ def _serialize_provenance(value: dict[str, Any]) -> str:
     return serialized
 
 
+def _validate_lab_method(provenance: dict[str, Any]) -> None:
+    method = provenance.get("method")
+    if not isinstance(method, dict):
+        raise ValueError("LAB provenance must include a method manifest")
+    if method.get("schema_version") != LAB_METHOD_SCHEMA:
+        raise ValueError("LAB method schema is invalid")
+    if method.get("completeness") not in {"complete", "legacy-partial"}:
+        raise ValueError("LAB method completeness is invalid")
+    if method.get("execution_class") not in {
+        "deterministic",
+        "ai-inference",
+        "hybrid",
+    }:
+        raise ValueError("LAB method execution class is invalid")
+    pipeline_id = method.get("pipeline_id")
+    if (
+        not isinstance(pipeline_id, str)
+        or not pipeline_id.strip()
+        or len(pipeline_id) > 160
+    ):
+        raise ValueError("LAB method pipeline id is invalid")
+    components = method.get("components")
+    if not isinstance(components, list) or not 1 <= len(components) <= 32:
+        raise ValueError("LAB method components are invalid")
+    identities = 0
+    for component in components:
+        if not isinstance(component, dict):
+            raise ValueError("LAB method component is invalid")
+        if component.get("kind") not in {"source", "tool", "model", "algorithm", "runtime"}:
+            raise ValueError("LAB method component kind is invalid")
+        for field in ("name", "version", "role"):
+            value = component.get(field)
+            if not isinstance(value, str) or not value.strip() or len(value) > 240:
+                raise ValueError(f"LAB method component {field} is invalid")
+        identity = component.get("identity_sha256")
+        if identity is not None:
+            if not isinstance(identity, str) or SHA256_PATTERN.fullmatch(identity) is None:
+                raise ValueError("LAB method component identity is invalid")
+            identities += 1
+    if identities == 0:
+        raise ValueError("LAB method must bind at least one component identity")
+    if method["completeness"] == "complete" and any(
+        component.get("identity_sha256") is None
+        for component in components
+        if component.get("kind") in {"model", "algorithm"}
+    ):
+        raise ValueError("complete LAB method must identify every model and algorithm")
+
+
 def _require_utc_timestamp(value: str, field: str) -> None:
     from datetime import datetime