File size: 8,357 Bytes
78ac017 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 | """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)
|