| """Validation and loading for the frozen temporal-scope semantic SFT release.""" |
|
|
| from __future__ import annotations |
|
|
| import json |
| import os |
| import re |
| from pathlib import Path |
| from typing import Any |
|
|
| from viss_core.data_v0.canonical import file_sha256, read_jsonl |
| from viss_core.data_v0.schema import validate_row |
|
|
| FORMAT = "streameva.temporal_scope_semantic_sft.v0" |
| TRAINER_RELATIVE_PATH = "trainer/sft.jsonl" |
| MANIFEST_RELATIVE_PATH = "manifest.json" |
| EXPECTED_PUBLIC_UPSTREAM_MANIFEST_SHA256 = os.environ.get("VISS_UPSTREAM_MANIFEST_SHA256", "") |
| ANSWER_RE = re.compile(r"^(?:[A-D]|Yes|No|10|[0-9])$") |
| DENSE_TEMPORAL_PARENT_KIND = "teacher_generated_pixel_verified_dense_temporal" |
| PURE_VLM_SCOPE = "pure_vlm_semantic_sft_only" |
| def _valid_semantic_target(row: dict[str, Any], text: str) -> bool: |
| if ANSWER_RE.fullmatch(text) is not None: |
| return True |
| parent_kinds = { |
| str(parent.get("kind") or "") for parent in row.get("parents") or [] |
| } |
| words = text.split() |
| return ( |
| DENSE_TEMPORAL_PARENT_KIND in parent_kinds |
| and 20 <= len(words) <= 220 |
| and "<" not in text |
| and ">" not in text |
| and "```" not in text |
| ) |
|
|
|
|
| class SemanticReleaseError(ValueError): |
| """Raised when the semantic SFT artifact is not a sealed release.""" |
|
|
|
|
| def infer_semantic_release_root(path: Path) -> Path | None: |
| path = path.resolve() |
| for parent in path.parents: |
| manifest = parent / MANIFEST_RELATIVE_PATH |
| if not manifest.is_file(): |
| continue |
| try: |
| value = json.loads(manifest.read_text(encoding="utf-8")) |
| except (OSError, json.JSONDecodeError): |
| continue |
| if value.get("format") == FORMAT: |
| return parent |
| return None |
|
|
|
|
| def validate_semantic_release( |
| root: Path, *, verify_media_hashes: bool |
| ) -> dict[str, Any]: |
| root = root.resolve() |
| manifest_path = root / MANIFEST_RELATIVE_PATH |
| manifest = json.loads(manifest_path.read_text(encoding="utf-8")) |
| required = { |
| "artifacts", |
| "contracts", |
| "counts", |
| "distributions", |
| "format", |
| "inputs", |
| "license", |
| "model_lineage", |
| "public_release_allowed", |
| "release_id", |
| "status", |
| "training_allowed", |
| "training_scope", |
| } |
| if set(manifest) != required: |
| raise SemanticReleaseError("semantic release manifest fields are invalid") |
| training_scope = manifest.get("training_scope") |
| if ( |
| manifest.get("format") != FORMAT |
| or manifest.get("status") != "sealed_internal_training_release" |
| or manifest.get("training_allowed") is not True |
| or manifest.get("public_release_allowed") is not False |
| or training_scope != PURE_VLM_SCOPE |
| ): |
| raise SemanticReleaseError("semantic release state is invalid") |
| contracts = manifest.get("contracts", {}) |
| if ( |
| contracts.get("agent_action_targets") != 0 |
| or contracts.get("environment_state_actor_targets") != 0 |
| or contracts.get("private_verifier_actor_targets") != 0 |
| or contracts.get("assistant_targets_plain_semantic_only") is not True |
| or contracts.get("self_contained_media") is not True |
| ): |
| raise SemanticReleaseError("semantic release contracts diverged") |
| if not re.fullmatch(r"[0-9a-f]{64}", EXPECTED_PUBLIC_UPSTREAM_MANIFEST_SHA256): |
| raise SemanticReleaseError("Set VISS_UPSTREAM_MANIFEST_SHA256 to your trusted upstream identity") |
| lineage = manifest.get("model_lineage", {}) |
| if ( |
| lineage.get("required_sha256s_sha256") |
| != EXPECTED_PUBLIC_UPSTREAM_MANIFEST_SHA256 |
| ): |
| raise SemanticReleaseError("semantic release public-upstream lineage diverged") |
|
|
| trainer = root / TRAINER_RELATIVE_PATH |
| inventory_path = root / "manifests/media_inventory.jsonl" |
| artifacts = manifest.get("artifacts", {}) |
| if ( |
| set(artifacts) != {TRAINER_RELATIVE_PATH, "manifests/media_inventory.jsonl"} |
| or not trainer.is_file() |
| or trainer.is_symlink() |
| or file_sha256(trainer) != artifacts.get(TRAINER_RELATIVE_PATH) |
| or not inventory_path.is_file() |
| or inventory_path.is_symlink() |
| or file_sha256(inventory_path) |
| != artifacts.get("manifests/media_inventory.jsonl") |
| ): |
| raise SemanticReleaseError("semantic release artifacts diverged") |
| rows = read_jsonl(trainer, verify_content_hashes=True, require_canonical=True) |
| if len(rows) != manifest.get("counts", {}).get("trainer_rows"): |
| raise SemanticReleaseError("semantic release row count diverged") |
| uids = set() |
| content_hashes = set() |
| referenced_media = set() |
| for row in rows: |
| validate_row(row) |
| if row["uid"] in uids or row["content_sha256"] in content_hashes: |
| raise SemanticReleaseError("semantic release contains duplicate rows") |
| uids.add(row["uid"]) |
| content_hashes.add(row["content_sha256"]) |
| if [message["role"] for message in row["messages"]] != ["user", "assistant"]: |
| raise SemanticReleaseError("non-VLM dialogue role admitted") |
| assistant = row["messages"][1] |
| texts = [item["text"] for item in assistant["content"] if item["type"] == "text"] |
| if len(texts) != 1 or not _valid_semantic_target(row, texts[0]): |
| raise SemanticReleaseError("non-semantic assistant target admitted") |
| images = [ |
| item["image"] |
| for message in row["messages"] |
| for item in message["content"] |
| if item["type"] == "image" |
| ] |
| invalid_frame_count = len(images) < 4 or len(images) > 32 or len(images) % 4 |
| if invalid_frame_count: |
| raise SemanticReleaseError("unsupported semantic frame count") |
| referenced_media.update(images) |
|
|
| inventory = read_jsonl(inventory_path, require_canonical=True) |
| if len(inventory) != manifest.get("counts", {}).get("media_files"): |
| raise SemanticReleaseError("semantic media inventory count diverged") |
| declared = set() |
| total_bytes = 0 |
| for item in inventory: |
| if set(item) != {"bytes", "path", "sha256"}: |
| raise SemanticReleaseError("semantic media inventory row is malformed") |
| relative = str(item["path"]) |
| path = (root / relative).resolve() |
| try: |
| path.relative_to(root) |
| except ValueError as exc: |
| raise SemanticReleaseError("semantic media path escapes release") from exc |
| if ( |
| relative in declared |
| or not relative.startswith("media/sha256/") |
| or not path.is_file() |
| or path.is_symlink() |
| or path.stat().st_size != item["bytes"] |
| or (verify_media_hashes and file_sha256(path) != item["sha256"]) |
| ): |
| raise SemanticReleaseError("semantic media inventory diverged") |
| declared.add(relative) |
| total_bytes += int(item["bytes"]) |
| if referenced_media != declared: |
| raise SemanticReleaseError("semantic media reference set diverged") |
| counts = manifest["counts"] |
| if total_bytes != counts.get("media_bytes"): |
| raise SemanticReleaseError("semantic media byte count diverged") |
| return manifest |
|
|
|
|
| def load_semantic_sft_rows(path: Path) -> list[dict[str, Any]]: |
| path = path.resolve() |
| root = infer_semantic_release_root(path) |
| if root is None or path.relative_to(root).as_posix() != TRAINER_RELATIVE_PATH: |
| raise SemanticReleaseError( |
| f"semantic trainer must be exactly {TRAINER_RELATIVE_PATH}" |
| ) |
| manifest = validate_semantic_release(root, verify_media_hashes=False) |
| inputs = manifest.get("inputs") or {} |
| selection = inputs.get("selection") or {} |
| if ( |
| selection.get("blind_audit_required_before_training") is True |
| and "blind_visual_filter" not in inputs |
| ): |
| raise SemanticReleaseError( |
| "refresh semantic candidate requires a complete blind-audit filter" |
| ) |
| if ( |
| selection.get("option_consensus_required_before_training") is True |
| and "option_consensus_filter" not in inputs |
| ): |
| raise SemanticReleaseError( |
| "distilled semantic candidate requires a complete option-consensus filter" |
| ) |
| return read_jsonl(path, verify_content_hashes=True, require_canonical=True) |
|
|
