Buckets:
| #!/usr/bin/env python3 | |
| """Fail-closed release checks for the Chebyshev Trackio logbook. | |
| The validator deliberately reads Trackio's static files directly. This keeps the | |
| release gate independent of a particular installed Trackio version while matching | |
| the on-disk cell and pin formats emitted by Trackio 0.31.x. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import re | |
| from dataclasses import dataclass | |
| from datetime import datetime | |
| from pathlib import Path, PurePosixPath | |
| from typing import Any, Sequence | |
| from urllib.parse import unquote | |
| CELL_RE = re.compile( | |
| r"(^|\n)---\n<!-- trackio-cell\n(?P<meta>[\s\S]*?)\n-->\n" | |
| r"(?P<body>[\s\S]*?)(?=\n---\n<!-- trackio-cell\n|\Z)" | |
| ) | |
| CELL_MARKER_RE = re.compile(r"(?m)^<!--\s*trackio-cell\b") | |
| FENCE_RE = re.compile(r"(`{3,4}|~{3,4})([^\n]*)\n([\s\S]*?)\n\1") | |
| PLACEHOLDER_RE = re.compile( | |
| r"__[A-Z0-9][A-Z0-9_-]*__|\b(?:TBD|TODO|RUNNING)\b" | |
| ) | |
| LOCAL_ARTIFACT_PREFIX = "trackio-local-path://" | |
| TRACKIO_ARTIFACT_PREFIX = "trackio-artifact://" | |
| LOCAL_ARTIFACT_RE = re.compile( | |
| r"trackio-local-path://([^\s<>\[\]()\"'`]+)" | |
| ) | |
| HTTPS_RE = re.compile(r"https://[^\s<>)\]\"'`]+") | |
| ALLOWED_CELL_TYPES = {"markdown", "code", "figure", "artifact", "dashboard"} | |
| POSTER_TITLE = "Reproduction poster" | |
| REQUIRED_CONCLUSION_TYPES = { | |
| "Executive summary": "markdown", | |
| POSTER_TITLE: "figure", | |
| "Reproduction bundle": "artifact", | |
| } | |
| EXECUTIVE_SUMMARY_TOKENS = ( | |
| "## Scope & cost", | |
| "This reproduction", | |
| "Full replication", | |
| "Scope", | |
| "Hardware", | |
| "Compute time", | |
| "Cost", | |
| "Outcome", | |
| ) | |
| class DuplicateKeyError(ValueError): | |
| """Raised when release metadata contains an ambiguous duplicate JSON key.""" | |
| class Page: | |
| slug: str | |
| title: str | |
| file: str | |
| path: Path | |
| order: int | |
| class Cell: | |
| page: Page | |
| index: int | |
| encounter: int | |
| metadata: dict[str, Any] | |
| body: str | |
| def cell_id(self) -> str: | |
| value = self.metadata.get("id") | |
| return value if isinstance(value, str) else "" | |
| def cell_type(self) -> str: | |
| value = self.metadata.get("type") | |
| return value if isinstance(value, str) else "" | |
| def title(self) -> str: | |
| value = self.metadata.get("title") | |
| return value if isinstance(value, str) else "" | |
| def _strict_json(text: str) -> Any: | |
| def reject_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]: | |
| result: dict[str, Any] = {} | |
| for key, value in pairs: | |
| if key in result: | |
| raise DuplicateKeyError(f"duplicate JSON key {key!r}") | |
| result[key] = value | |
| return result | |
| return json.loads(text, object_pairs_hook=reject_duplicates) | |
| def _line_number(text: str, offset: int) -> int: | |
| return text.count("\n", 0, offset) + 1 | |
| def _is_iso_datetime(value: str) -> bool: | |
| try: | |
| datetime.fromisoformat(value.replace("Z", "+00:00")) | |
| except ValueError: | |
| return False | |
| return True | |
| def _fence_parts(body: str) -> dict[str, list[tuple[str, str]]]: | |
| parts: dict[str, list[tuple[str, str]]] = {} | |
| for match in FENCE_RE.finditer(body): | |
| info = match.group(2).strip() | |
| language = (info.split() or [""])[0].lower() | |
| parts.setdefault(language, []).append((info, match.group(3))) | |
| return parts | |
| def _release_scan_text(text: str) -> str: | |
| """Remove only opaque figure payloads from release-safety text scans. | |
| Plotly's vendored JavaScript can legitimately contain strings that look like | |
| release placeholders or local URLs. Figure metadata, prose outside the | |
| payload fences, and every non-figure cell remain visible to the scanners. | |
| Malformed metadata is left untouched and is rejected separately while cells | |
| are loaded. | |
| """ | |
| def replace_cell(match: re.Match[str]) -> str: | |
| try: | |
| metadata = _strict_json(match.group("meta")) | |
| except (json.JSONDecodeError, DuplicateKeyError): | |
| return match.group(0) | |
| if not isinstance(metadata, dict) or metadata.get("type") != "figure": | |
| return match.group(0) | |
| def replace_fence(fence: re.Match[str]) -> str: | |
| info = fence.group(2).strip() | |
| language = (info.split() or [""])[0].lower() | |
| return "" if language in {"html", "raw"} else fence.group(0) | |
| sanitized_body = FENCE_RE.sub(replace_fence, match.group("body")) | |
| body_offset = match.start("body") - match.start() | |
| return match.group(0)[:body_offset] + sanitized_body | |
| return CELL_RE.sub(replace_cell, text) | |
| class ReleaseValidator: | |
| def __init__(self, logbook_root: Path, project_root: Path, mode: str) -> None: | |
| self.logbook_root = logbook_root.expanduser().resolve() | |
| self.project_root = project_root.expanduser().resolve() | |
| self.mode = mode | |
| self.errors: list[dict[str, str]] = [] | |
| self.pages: list[Page] = [] | |
| self.cells: list[Cell] = [] | |
| self.page_text: dict[str, str] = {} | |
| self.manifest: dict[str, Any] = {} | |
| self.local_artifact_count = 0 | |
| def error(self, code: str, message: str, location: str | None = None) -> None: | |
| item = {"code": code, "message": message} | |
| if location: | |
| item["location"] = location | |
| self.errors.append(item) | |
| def run(self) -> dict[str, Any]: | |
| if self.mode not in {"prepublish", "postpublish"}: | |
| self.error("invalid_mode", f"unsupported validation mode: {self.mode!r}") | |
| if not self.project_root.is_dir(): | |
| self.error( | |
| "missing_project_root", | |
| "project root is not an existing directory", | |
| str(self.project_root), | |
| ) | |
| if not self.logbook_root.is_dir(): | |
| self.error( | |
| "missing_logbook_root", | |
| "logbook root is not an existing directory", | |
| str(self.logbook_root), | |
| ) | |
| return self.summary() | |
| manifest_text = self._load_manifest() | |
| if self.manifest: | |
| self._load_declared_pages() | |
| self._check_page_inventory() | |
| self._load_cells() | |
| self._check_placeholders(manifest_text) | |
| self._check_figures() | |
| self._check_conclusion() | |
| self._check_pins() | |
| self._check_artifact_uris() | |
| self._check_agent_view_tokens() | |
| return self.summary() | |
| def summary(self) -> dict[str, Any]: | |
| figures = [cell for cell in self.cells if cell.cell_type == "figure"] | |
| pinned = [cell for cell in self.cells if cell.metadata.get("pinned") is True] | |
| artifact_cells = [cell for cell in self.cells if cell.cell_type == "artifact"] | |
| valid = not self.errors | |
| return { | |
| "schema_version": "1.0.0", | |
| "status": "success" if valid else "failed", | |
| "valid": valid, | |
| "mode": self.mode, | |
| "logbook_root": str(self.logbook_root), | |
| "project_root": str(self.project_root), | |
| "counts": { | |
| "declared_pages": len(self.pages), | |
| "cells": len(self.cells), | |
| "figures": len(figures), | |
| "nonposter_figures": sum( | |
| cell.title != POSTER_TITLE for cell in figures | |
| ), | |
| "artifact_cells": len(artifact_cells), | |
| "local_artifact_uris": self.local_artifact_count, | |
| "pins": len(pinned), | |
| "errors": len(self.errors), | |
| }, | |
| "errors": self.errors, | |
| } | |
| def _load_manifest(self) -> str: | |
| path = self.logbook_root / "logbook.json" | |
| if not path.is_file(): | |
| self.error("missing_manifest", "logbook.json is missing", str(path)) | |
| return "" | |
| try: | |
| text = path.read_text(encoding="utf-8") | |
| except (OSError, UnicodeError) as exc: | |
| self.error("unreadable_manifest", str(exc), str(path)) | |
| return "" | |
| try: | |
| value = _strict_json(text) | |
| except (json.JSONDecodeError, DuplicateKeyError) as exc: | |
| self.error("invalid_manifest_json", str(exc), str(path)) | |
| return text | |
| if not isinstance(value, dict): | |
| self.error("invalid_manifest", "logbook.json must contain an object", str(path)) | |
| return text | |
| self.manifest = value | |
| return text | |
| def _load_declared_pages(self) -> None: | |
| root = self.manifest.get("root") | |
| if not isinstance(root, dict): | |
| self.error("invalid_page_tree", "manifest root must be an object") | |
| return | |
| seen_slugs: dict[str, str] = {} | |
| seen_files: dict[str, str] = {} | |
| def visit(node: Any, location: str) -> None: | |
| if not isinstance(node, dict): | |
| self.error("invalid_page_node", "page node must be an object", location) | |
| return | |
| slug = node.get("slug") | |
| title = node.get("title") | |
| file_value = node.get("file") | |
| if not isinstance(slug, str) or not slug.strip(): | |
| self.error("invalid_page_slug", "page slug must be a non-empty string", location) | |
| slug = "" | |
| if not isinstance(title, str) or not title.strip(): | |
| self.error( | |
| "invalid_page_title", "page title must be a non-empty string", location | |
| ) | |
| title = "" | |
| if slug: | |
| if slug in seen_slugs: | |
| self.error( | |
| "duplicate_page_slug", | |
| f"page slug {slug!r} is declared more than once", | |
| location, | |
| ) | |
| else: | |
| seen_slugs[slug] = location | |
| safe_file = self._safe_declared_file(file_value, location) | |
| if safe_file is not None: | |
| relative, path = safe_file | |
| if relative in seen_files: | |
| self.error( | |
| "duplicate_page_file", | |
| f"page file {relative!r} is declared more than once", | |
| location, | |
| ) | |
| else: | |
| seen_files[relative] = location | |
| self.pages.append( | |
| Page( | |
| slug=slug, | |
| title=title, | |
| file=relative, | |
| path=path, | |
| order=len(self.pages), | |
| ) | |
| ) | |
| children = node.get("children") | |
| if not isinstance(children, list): | |
| self.error("invalid_page_children", "page children must be a list", location) | |
| return | |
| for index, child in enumerate(children): | |
| visit(child, f"{location}.children[{index}]") | |
| visit(root, "logbook.json:root") | |
| if self.pages and self.pages[0].file != "pages/index.md": | |
| self.error( | |
| "invalid_index_page", | |
| "the manifest root must declare pages/index.md", | |
| self.pages[0].file, | |
| ) | |
| def _safe_declared_file( | |
| self, file_value: Any, location: str | |
| ) -> tuple[str, Path] | None: | |
| if not isinstance(file_value, str) or not file_value: | |
| self.error("invalid_page_file", "page file must be a non-empty string", location) | |
| return None | |
| if "\\" in file_value: | |
| self.error( | |
| "unsafe_page_file", "page files must use POSIX separators", location | |
| ) | |
| return None | |
| pure = PurePosixPath(file_value) | |
| if pure.is_absolute() or any(part in {"", ".", ".."} for part in pure.parts): | |
| self.error( | |
| "unsafe_page_file", | |
| f"page file escapes or is not normalized: {file_value!r}", | |
| location, | |
| ) | |
| return None | |
| relative = pure.as_posix() | |
| candidate = (self.logbook_root / Path(*pure.parts)).resolve() | |
| if not candidate.is_relative_to(self.logbook_root): | |
| self.error( | |
| "unsafe_page_file", | |
| f"page file resolves outside the logbook: {file_value!r}", | |
| location, | |
| ) | |
| return None | |
| if pure.suffix.lower() != ".md" or not pure.parts or pure.parts[0] != "pages": | |
| self.error( | |
| "invalid_page_file", | |
| f"declared page is not a Markdown file under pages/: {file_value!r}", | |
| location, | |
| ) | |
| return None | |
| return relative, candidate | |
| def _check_page_inventory(self) -> None: | |
| declared = {page.file for page in self.pages} | |
| pages_root = self.logbook_root / "pages" | |
| actual: set[str] = set() | |
| if pages_root.is_dir(): | |
| for path in pages_root.rglob("*.md"): | |
| actual.add(path.relative_to(self.logbook_root).as_posix()) | |
| else: | |
| self.error("missing_pages_directory", "pages/ directory is missing", str(pages_root)) | |
| for file in sorted(declared - actual): | |
| self.error("missing_declared_page", "declared page is missing", file) | |
| for file in sorted(actual - declared): | |
| self.error("undeclared_page", "Markdown page is not declared by logbook.json", file) | |
| for page in self.pages: | |
| if page.file not in actual: | |
| continue | |
| if not page.path.is_file(): | |
| self.error("missing_declared_page", "declared page is not a file", page.file) | |
| continue | |
| resolved = page.path.resolve() | |
| if not resolved.is_relative_to(self.logbook_root): | |
| self.error( | |
| "unsafe_page_file", | |
| "declared page resolves outside the logbook", | |
| page.file, | |
| ) | |
| def _load_cells(self) -> None: | |
| seen_ids: dict[str, str] = {} | |
| encounter = 0 | |
| for page in self.pages: | |
| if not page.path.is_file() or not page.path.resolve().is_relative_to( | |
| self.logbook_root | |
| ): | |
| continue | |
| try: | |
| text = page.path.read_text(encoding="utf-8") | |
| except (OSError, UnicodeError) as exc: | |
| self.error("unreadable_page", str(exc), page.file) | |
| continue | |
| self.page_text[page.file] = text | |
| matches = list(CELL_RE.finditer(text)) | |
| marker_count = len(CELL_MARKER_RE.findall(text)) | |
| if marker_count != len(matches): | |
| self.error( | |
| "malformed_cell_marker", | |
| f"found {marker_count} Trackio marker(s) but parsed {len(matches)} cell(s)", | |
| page.file, | |
| ) | |
| for index, match in enumerate(matches): | |
| location = f"{page.file}:{_line_number(text, match.start('meta'))}" | |
| try: | |
| metadata = _strict_json(match.group("meta")) | |
| except (json.JSONDecodeError, DuplicateKeyError) as exc: | |
| self.error("invalid_cell_metadata_json", str(exc), location) | |
| continue | |
| if not isinstance(metadata, dict): | |
| self.error( | |
| "invalid_cell_metadata", | |
| "Trackio cell metadata must be a JSON object", | |
| location, | |
| ) | |
| continue | |
| cell = Cell( | |
| page=page, | |
| index=index, | |
| encounter=encounter, | |
| metadata=metadata, | |
| body=match.group("body"), | |
| ) | |
| encounter += 1 | |
| self.cells.append(cell) | |
| self._validate_cell_metadata(cell, location, seen_ids) | |
| self._check_code_attachments(cell, location) | |
| def _validate_cell_metadata( | |
| self, cell: Cell, location: str, seen_ids: dict[str, str] | |
| ) -> None: | |
| if not cell.cell_id: | |
| self.error("invalid_cell_id", "cell id must be a non-empty string", location) | |
| elif cell.cell_id in seen_ids: | |
| self.error( | |
| "duplicate_cell_id", | |
| f"cell id {cell.cell_id!r} was already used at {seen_ids[cell.cell_id]}", | |
| location, | |
| ) | |
| else: | |
| seen_ids[cell.cell_id] = location | |
| if cell.cell_type not in ALLOWED_CELL_TYPES: | |
| self.error( | |
| "invalid_cell_type", | |
| f"unsupported Trackio cell type {cell.cell_type!r}", | |
| location, | |
| ) | |
| if not cell.title: | |
| self.error("invalid_cell_title", "cell title must be a non-empty string", location) | |
| pinned = cell.metadata.get("pinned") | |
| pinned_at = cell.metadata.get("pinned_at") | |
| if "pinned" in cell.metadata and pinned is not True: | |
| self.error( | |
| "invalid_pin_metadata", | |
| "Trackio pins must use boolean `pinned: true`; unpinned cells omit it", | |
| location, | |
| ) | |
| if pinned is True: | |
| if not isinstance(pinned_at, str) or not _is_iso_datetime(pinned_at): | |
| self.error( | |
| "invalid_pin_metadata", | |
| "pinned cells require an ISO-8601 pinned_at string", | |
| location, | |
| ) | |
| elif "pinned_at" in cell.metadata: | |
| self.error( | |
| "invalid_pin_metadata", | |
| "pinned_at is present without `pinned: true`", | |
| location, | |
| ) | |
| def _check_code_attachments(self, cell: Cell, location: str) -> None: | |
| if cell.cell_type != "code": | |
| return | |
| for language, entries in _fence_parts(cell.body).items(): | |
| if language not in {"python", "json"}: | |
| continue | |
| for info, _ in entries: | |
| if re.search(r"(?:^|\s)title\s*=", info, flags=re.IGNORECASE): | |
| self.error( | |
| "titled_code_attachment", | |
| "code cells may not retain titled Python/JSON attachments", | |
| location, | |
| ) | |
| def _check_placeholders(self, manifest_text: str) -> None: | |
| sources = [ | |
| ("logbook.json", manifest_text), | |
| *((location, _release_scan_text(text)) for location, text in self.page_text.items()), | |
| ] | |
| for location, text in sources: | |
| for match in PLACEHOLDER_RE.finditer(text): | |
| self.error( | |
| "placeholder_token", | |
| f"unresolved final-release token {match.group(0)!r}", | |
| f"{location}:{_line_number(text, match.start())}", | |
| ) | |
| def _check_figures(self) -> None: | |
| figures = [cell for cell in self.cells if cell.cell_type == "figure"] | |
| nonposter = [cell for cell in figures if cell.title != POSTER_TITLE] | |
| if len(figures) < 5: | |
| self.error( | |
| "figure_count", | |
| f"release requires at least 5 figure cells; found {len(figures)}", | |
| ) | |
| if len(nonposter) != 4: | |
| self.error( | |
| "nonposter_figure_count", | |
| f"release requires exactly 4 nonposter figure cells; found {len(nonposter)}", | |
| ) | |
| for cell in figures: | |
| parts = _fence_parts(cell.body) | |
| html_payloads = [payload for _, payload in parts.get("html", [])] | |
| if not any(payload.strip() for payload in html_payloads): | |
| self.error( | |
| "missing_figure_html", | |
| "figure cell has no non-empty HTML/image payload", | |
| f"{cell.page.file}:{cell.cell_id}", | |
| ) | |
| if cell.title == POSTER_TITLE: | |
| continue | |
| raw_payloads = [payload for _, payload in parts.get("raw", [])] | |
| if not any(payload.strip() for payload in raw_payloads): | |
| self.error( | |
| "missing_figure_raw", | |
| "nonposter figure has no non-empty raw/reference payload", | |
| f"{cell.page.file}:{cell.cell_id}", | |
| ) | |
| def _conclusion_page(self) -> Page | None: | |
| candidates = [ | |
| page | |
| for page in self.pages | |
| if page.slug == "conclusion" and page.title == "Conclusion" | |
| ] | |
| if len(candidates) != 1: | |
| self.error( | |
| "missing_conclusion_page", | |
| "manifest must contain exactly one page with slug/title Conclusion", | |
| ) | |
| return None | |
| return candidates[0] | |
| def _check_conclusion(self) -> None: | |
| page = self._conclusion_page() | |
| if page is None: | |
| return | |
| cells = [cell for cell in self.cells if cell.page.file == page.file] | |
| by_title: dict[str, list[Cell]] = {} | |
| for cell in cells: | |
| by_title.setdefault(cell.title, []).append(cell) | |
| required: dict[str, Cell] = {} | |
| for title, expected_type in REQUIRED_CONCLUSION_TYPES.items(): | |
| matches = by_title.get(title, []) | |
| if len(matches) != 1: | |
| self.error( | |
| "required_conclusion_cell", | |
| "Conclusion must contain exactly one cell titled " | |
| f"{title!r}; found {len(matches)}", | |
| page.file, | |
| ) | |
| continue | |
| required[title] = matches[0] | |
| if matches[0].cell_type != expected_type: | |
| self.error( | |
| "required_conclusion_cell_type", | |
| f"{title!r} must be a {expected_type} cell, not {matches[0].cell_type!r}", | |
| f"{page.file}:{matches[0].cell_id}", | |
| ) | |
| summary = required.get("Executive summary") | |
| if summary is not None: | |
| for token in EXECUTIVE_SUMMARY_TOKENS: | |
| if token not in summary.body: | |
| self.error( | |
| "executive_summary_content", | |
| f"Executive summary is missing required text {token!r}", | |
| f"{page.file}:{summary.cell_id}", | |
| ) | |
| def _check_pins(self) -> None: | |
| pinned = [cell for cell in self.cells if cell.metadata.get("pinned") is True] | |
| pinned.sort( | |
| key=lambda cell: ( | |
| str( | |
| cell.metadata.get("pinned_at") | |
| or cell.metadata.get("created_at") | |
| or "" | |
| ), | |
| cell.encounter, | |
| cell.index, | |
| ) | |
| ) | |
| titles = [cell.title for cell in pinned] | |
| expected = ["Executive summary", POSTER_TITLE] | |
| if len(pinned) != 2: | |
| self.error("pin_count", f"release requires exactly 2 pins; found {len(pinned)}") | |
| if titles != expected: | |
| self.error( | |
| "pin_order", | |
| f"Trackio pin order must be {expected!r}; found {titles!r}", | |
| ) | |
| conclusion = next( | |
| ( | |
| page | |
| for page in self.pages | |
| if page.slug == "conclusion" and page.title == "Conclusion" | |
| ), | |
| None, | |
| ) | |
| if conclusion is not None and len(pinned) == 2: | |
| if any(cell.page.file != conclusion.file for cell in pinned): | |
| self.error( | |
| "pin_location", | |
| "Executive summary and poster pins must be the Conclusion cells", | |
| ) | |
| def _check_artifact_uris(self) -> None: | |
| all_text = "\n".join( | |
| _release_scan_text(text) for text in self.page_text.values() | |
| ) | |
| local_matches = list(LOCAL_ARTIFACT_RE.finditer(all_text)) | |
| self.local_artifact_count = len(local_matches) | |
| prefix_count = all_text.count(LOCAL_ARTIFACT_PREFIX) | |
| if prefix_count != len(local_matches): | |
| self.error( | |
| "malformed_local_artifact_uri", | |
| "one or more trackio-local-path URIs have no parseable path", | |
| ) | |
| if "file://" in all_text: | |
| self.error("forbidden_file_uri", "file:// URLs are not release-safe") | |
| if self.mode == "postpublish": | |
| if LOCAL_ARTIFACT_PREFIX in all_text: | |
| self.error( | |
| "unpublished_local_artifact", | |
| "postpublish logbooks may not contain trackio-local-path://", | |
| ) | |
| if TRACKIO_ARTIFACT_PREFIX in all_text: | |
| self.error( | |
| "unpublished_trackio_artifact", | |
| "postpublish logbooks may not contain trackio-artifact://", | |
| ) | |
| else: | |
| for match in local_matches: | |
| self._check_local_artifact_path(match.group(1)) | |
| for cell in (cell for cell in self.cells if cell.cell_type == "artifact"): | |
| has_local = LOCAL_ARTIFACT_PREFIX in cell.body | |
| has_trackio = TRACKIO_ARTIFACT_PREFIX in cell.body | |
| has_https = HTTPS_RE.search(cell.body) is not None | |
| if self.mode == "postpublish": | |
| if not has_https: | |
| self.error( | |
| "artifact_without_public_url", | |
| "postpublish artifact cells require an HTTPS URL", | |
| f"{cell.page.file}:{cell.cell_id}", | |
| ) | |
| elif not (has_local or has_trackio or has_https): | |
| self.error( | |
| "artifact_without_target", | |
| "prepublish artifact cells require a local, Trackio, or HTTPS target", | |
| f"{cell.page.file}:{cell.cell_id}", | |
| ) | |
| def _check_local_artifact_path(self, encoded_path: str) -> None: | |
| decoded = unquote(encoded_path) | |
| if not decoded or "\x00" in decoded: | |
| self.error( | |
| "invalid_local_artifact_path", | |
| f"invalid local artifact path {encoded_path!r}", | |
| ) | |
| return | |
| local = Path(decoded).expanduser() | |
| candidate = ( | |
| local.resolve() | |
| if local.is_absolute() | |
| else (self.project_root / local).resolve() | |
| ) | |
| if not candidate.is_relative_to(self.project_root): | |
| self.error( | |
| "unsafe_local_artifact_path", | |
| "local artifact resolves outside project root", | |
| decoded, | |
| ) | |
| return | |
| if not candidate.is_file(): | |
| self.error( | |
| "missing_local_artifact", | |
| "local artifact URI does not resolve to an existing file", | |
| decoded, | |
| ) | |
| def _check_agent_view_tokens(self) -> None: | |
| value = self.manifest.get("agent_view_tokens") | |
| if isinstance(value, bool) or not isinstance(value, int): | |
| self.error( | |
| "invalid_agent_view_tokens", | |
| "manifest agent_view_tokens must be an integer", | |
| ) | |
| return | |
| if value < 0 or value >= 100_000: | |
| self.error( | |
| "agent_view_tokens_limit", | |
| f"agent_view_tokens must be in [0, 100000); found {value}", | |
| ) | |
| def validate_logbook_release( | |
| logbook_root: str | Path, project_root: str | Path, mode: str | |
| ) -> dict[str, Any]: | |
| """Validate a Trackio logbook and return its machine-readable summary.""" | |
| validator = ReleaseValidator(Path(logbook_root), Path(project_root), mode) | |
| try: | |
| return validator.run() | |
| except Exception as exc: # pragma: no cover - final fail-closed safety net | |
| validator.error("internal_validator_error", f"{type(exc).__name__}: {exc}") | |
| return validator.summary() | |
| def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: | |
| parser = argparse.ArgumentParser( | |
| description="Fail-closed Trackio logbook release validation" | |
| ) | |
| parser.add_argument("--logbook-root", type=Path, required=True) | |
| parser.add_argument("--project-root", type=Path, required=True) | |
| parser.add_argument( | |
| "--mode", choices=("prepublish", "postpublish"), required=True | |
| ) | |
| return parser.parse_args(argv) | |
| def main(argv: Sequence[str] | None = None) -> int: | |
| args = parse_args(argv) | |
| summary = validate_logbook_release( | |
| logbook_root=args.logbook_root, | |
| project_root=args.project_root, | |
| mode=args.mode, | |
| ) | |
| print(json.dumps(summary, indent=2, sort_keys=True)) | |
| return 0 if summary["valid"] else 2 | |
| if __name__ == "__main__": | |
| raise SystemExit(main()) | |
Xet Storage Details
- Size:
- 28.9 kB
- Xet hash:
- a156c608f54ca0731b3f44f6eac6e4b8f0b674ac613815a279756468f1c58d37
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.