| """Minimal canonical schema for the released pure-VLM semantic rows.""" |
|
|
| from __future__ import annotations |
|
|
| import math |
| import re |
| from collections.abc import Mapping, Sequence |
| from typing import Any |
|
|
| from viss_core.data_v0.canonical import validate_content_sha256 |
|
|
|
|
| SCHEMA_TRAIN_ROW = "streameva.sft_row.v1" |
| _SHA256_RE = re.compile(r"^[0-9a-f]{64}$") |
| _ID_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_.-]{2,127}$") |
|
|
|
|
| class SchemaError(ValueError): |
| """Raised when a semantic training row violates its release ABI.""" |
|
|
|
|
| def validate_row(row: Mapping[str, Any], *, verify_hash: bool = True) -> None: |
| """Validate one plain-answer video SFT row and reject legacy schemas.""" |
|
|
| if not isinstance(row, Mapping) or row.get("schema") != SCHEMA_TRAIN_ROW: |
| raise SchemaError("only the semantic SFT-row schema is supported") |
| _exact_keys( |
| row, |
| required={ |
| "schema", |
| "uid", |
| "component", |
| "group_id", |
| "split", |
| "messages", |
| "sample_loss_weight", |
| "parents", |
| "content_sha256", |
| }, |
| optional={"decision_loss_profile"}, |
| path="sft_row", |
| ) |
| _identifier(row["uid"], "sft_row.uid") |
| _identifier(row["group_id"], "sft_row.group_id") |
| component = row["component"] |
| if component != "preservation": |
| raise SchemaError("semantic SFT row uses a non-semantic component") |
| if row["split"] not in {"train", "policy_holdout"}: |
| raise SchemaError("semantic SFT split must be train or policy_holdout") |
| if row.get("decision_loss_profile", "full") != "full": |
| raise SchemaError("semantic SFT decision-loss profile diverged") |
| weight = row["sample_loss_weight"] |
| if ( |
| isinstance(weight, bool) |
| or not isinstance(weight, (int, float)) |
| or not math.isfinite(float(weight)) |
| or weight <= 0 |
| ): |
| raise SchemaError("sample_loss_weight must be finite and positive") |
|
|
| messages = _sequence(row["messages"], "sft_row.messages") |
| if len(messages) != 2 or [message.get("role") for message in messages] != [ |
| "user", |
| "assistant", |
| ]: |
| raise SchemaError("semantic SFT must contain user then assistant") |
| for index, message in enumerate(messages): |
| path = f"sft_row.messages[{index}]" |
| _exact_keys(message, required={"role", "content", "loss"}, path=path) |
| expected_loss = message["role"] == "assistant" |
| if message["loss"] is not expected_loss: |
| raise SchemaError("only the assistant semantic answer may carry loss") |
| content = _sequence(message["content"], f"{path}.content") |
| if not content: |
| raise SchemaError(f"{path}.content must not be empty") |
| for item_index, item in enumerate(content): |
| item_path = f"{path}.content[{item_index}]" |
| if not isinstance(item, Mapping): |
| raise SchemaError(f"{item_path} must be an object") |
| if item.get("type") == "text": |
| _exact_keys(item, required={"type", "text"}, path=item_path) |
| _nonempty_string(item["text"], f"{item_path}.text") |
| elif item.get("type") == "image": |
| if message["role"] != "user": |
| raise SchemaError("only the user may contain images") |
| _exact_keys( |
| item, |
| required={"type", "image", "frame_id"}, |
| path=item_path, |
| ) |
| _relative_uri(item["image"], f"{item_path}.image") |
| _identifier(item["frame_id"], f"{item_path}.frame_id") |
| else: |
| raise SchemaError(f"{item_path}.type must be text or image") |
| assistant_content = messages[1]["content"] |
| if len(assistant_content) != 1 or assistant_content[0].get("type") != "text": |
| raise SchemaError("assistant must contain exactly one semantic text answer") |
| _parent_list(row["parents"], "sft_row.parents") |
| _forbid_absolute_paths(row) |
| if verify_hash: |
| try: |
| validate_content_sha256(row) |
| except ValueError as error: |
| raise SchemaError(str(error)) from error |
|
|
|
|
| def _exact_keys( |
| value: Mapping[str, Any], |
| *, |
| required: set[str], |
| path: str, |
| optional: set[str] | None = None, |
| ) -> None: |
| if not isinstance(value, Mapping): |
| raise SchemaError(f"{path} must be an object") |
| optional = optional or set() |
| missing = required - set(value) |
| unknown = set(value) - required - optional |
| if missing or unknown: |
| raise SchemaError( |
| f"{path} fields diverged: missing={sorted(missing)} unknown={sorted(unknown)}" |
| ) |
|
|
|
|
| def _sequence(value: Any, path: str) -> Sequence[Any]: |
| if not isinstance(value, Sequence) or isinstance(value, (str, bytes, bytearray)): |
| raise SchemaError(f"{path} must be an array") |
| return value |
|
|
|
|
| def _nonempty_string(value: Any, path: str) -> str: |
| if not isinstance(value, str) or not value.strip(): |
| raise SchemaError(f"{path} must be non-empty text") |
| return value |
|
|
|
|
| def _identifier(value: Any, path: str) -> str: |
| value = _nonempty_string(value, path) |
| if _ID_RE.fullmatch(value) is None: |
| raise SchemaError(f"{path} is not a stable identifier") |
| return value |
|
|
|
|
| def _relative_uri(value: Any, path: str) -> str: |
| value = _nonempty_string(value, path) |
| if value.startswith(("/", "file:")) or ".." in value.split("/"): |
| raise SchemaError(f"{path} must be release-relative") |
| return value |
|
|
|
|
| def _parent_list(value: Any, path: str) -> None: |
| for index, parent in enumerate(_sequence(value, path)): |
| parent_path = f"{path}[{index}]" |
| _exact_keys( |
| parent, |
| required={"kind", "id", "sha256"}, |
| path=parent_path, |
| ) |
| _nonempty_string(parent["kind"], f"{parent_path}.kind") |
| _identifier(parent["id"], f"{parent_path}.id") |
| if not isinstance(parent["sha256"], str) or _SHA256_RE.fullmatch( |
| parent["sha256"] |
| ) is None: |
| raise SchemaError(f"{parent_path}.sha256 must be lowercase SHA256") |
|
|
|
|
| def _forbid_absolute_paths(value: Any, *, path: str = "$") -> None: |
| if isinstance(value, Mapping): |
| for key, child in value.items(): |
| _forbid_absolute_paths(child, path=f"{path}.{key}") |
| elif isinstance(value, Sequence) and not isinstance( |
| value, (str, bytes, bytearray) |
| ): |
| for index, child in enumerate(value): |
| _forbid_absolute_paths(child, path=f"{path}[{index}]") |
| elif isinstance(value, str) and value.startswith(("/", "file://")): |
| raise SchemaError(f"{path}: absolute filesystem paths are forbidden") |
|
|
