298 lines
12 KiB
Python
298 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Build and validate the deterministic fixed-channel field Recovery `.skill`."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
|
|
SKILL_NAME = "fixed-channel-field-recovery"
|
|
SKILL_VERSION = "fixed-channel-field-recovery-qbd-v1"
|
|
ARCHIVE_ROOT = SKILL_NAME
|
|
FIXED_TIMESTAMP = (2026, 8, 10, 0, 0, 0)
|
|
FIXED_TIMESTAMP_UTC = "2026-08-10T00:00:00Z"
|
|
PUBLICATION_STATUS = "INTERNAL_ONLY"
|
|
EXPECTED_RELATIVE_FILES = (
|
|
Path("SKILL.md"),
|
|
Path("agents/openai.yaml"),
|
|
Path("references/contract.md"),
|
|
Path("references/qbd-profile.md"),
|
|
Path("references/recovery-patch-set-v1.schema.json"),
|
|
Path("references/recovery-request-set-v1.schema.json"),
|
|
)
|
|
REQUIRED_MARKERS = {
|
|
Path("SKILL.md"): (
|
|
"fixed-channel-recovery-v1",
|
|
"RecoveryPatchSet",
|
|
"REFERENCE_ONLY",
|
|
"SCHEMA_CONSTRAINED",
|
|
"Never turn a missing",
|
|
),
|
|
Path("references/contract.md"): (
|
|
"Recovery Contract Reference",
|
|
"ROW_OPERATION_DATE",
|
|
"ROOM_SOURCE_PRICE",
|
|
"UNIQUE_ALLOWED_CANDIDATE",
|
|
"MULTIPLE_SCHEMA_VALID_CANDIDATES",
|
|
),
|
|
Path("references/qbd-profile.md"): (
|
|
"Ignore QBD F/Nights Override completely",
|
|
"QBD has no source Rate Code",
|
|
"departure minus arrival",
|
|
"missing price stays MISSING",
|
|
"GRPA1",
|
|
),
|
|
Path("references/recovery-patch-set-v1.schema.json"): (
|
|
"FixedChannelRecoveryPatchSetV1",
|
|
"selected_candidate_ref_ids",
|
|
"TASK_PROCESSING_FAILED",
|
|
"additionalProperties",
|
|
),
|
|
Path("references/recovery-request-set-v1.schema.json"): (
|
|
"FixedChannelRecoveryRequestSetV1",
|
|
"locked_context",
|
|
"PARSER_NORMALIZATION",
|
|
"TOKEN_FORMAT_UNRECOGNIZED",
|
|
),
|
|
}
|
|
FORBIDDEN_TEXT_PATTERNS = (
|
|
(re.compile(r"https?://", re.IGNORECASE), "network URL"),
|
|
(re.compile(r"\b[\w.+-]+@[\w.-]+\.[A-Za-z]{2,}\b"), "email address"),
|
|
(re.compile(r"rate-room-v20\d{6}", re.IGNORECASE), "Layer 4 directory version"),
|
|
(re.compile(r"rate-room-directory", re.IGNORECASE), "Layer 4 directory payload"),
|
|
(re.compile(r"\.xlsx\b", re.IGNORECASE), "workbook file name"),
|
|
(re.compile(r"\b(?:api[_-]?key|hmac[_-]?secret|bearer\s+[A-Za-z0-9._-]+)\b", re.IGNORECASE), "secret marker"),
|
|
)
|
|
|
|
|
|
def fail(message: str) -> None:
|
|
raise ValueError(message)
|
|
|
|
|
|
def sha256_bytes(value: bytes) -> str:
|
|
return hashlib.sha256(value).hexdigest()
|
|
|
|
|
|
def sha256(path: Path) -> str:
|
|
return sha256_bytes(path.read_bytes())
|
|
|
|
|
|
def _source_paths(source: Path) -> list[Path]:
|
|
if not source.is_dir():
|
|
fail(f"Skill source directory is missing: {source}")
|
|
actual = sorted(
|
|
(path.relative_to(source) for path in source.rglob("*") if path.is_file()),
|
|
key=Path.as_posix,
|
|
)
|
|
expected = list(EXPECTED_RELATIVE_FILES)
|
|
if actual != expected:
|
|
fail(
|
|
"Skill source members differ from the frozen set: "
|
|
f"expected={[path.as_posix() for path in expected]}, "
|
|
f"actual={[path.as_posix() for path in actual]}"
|
|
)
|
|
paths = [source / relative for relative in expected]
|
|
if any(path.is_symlink() for path in paths):
|
|
fail("Skill source cannot contain symbolic links")
|
|
return paths
|
|
|
|
|
|
def _validate_frontmatter(skill_text: str) -> None:
|
|
match = re.match(r"^---\n(?P<body>.*?)\n---\n", skill_text, re.DOTALL)
|
|
if not match:
|
|
fail("SKILL.md frontmatter is malformed")
|
|
entries: dict[str, str] = {}
|
|
for line in match.group("body").splitlines():
|
|
if ":" not in line:
|
|
fail("SKILL.md frontmatter contains a malformed line")
|
|
key, value = line.split(":", 1)
|
|
key = key.strip()
|
|
value = value.strip()
|
|
if not key or not value or key in entries:
|
|
fail("SKILL.md frontmatter contains an empty or duplicate field")
|
|
entries[key] = value
|
|
if set(entries) != {"name", "description"}:
|
|
fail("SKILL.md frontmatter must contain only name and description")
|
|
if entries["name"] != SKILL_NAME:
|
|
fail("SKILL.md name does not match the archive root")
|
|
description = entries["description"]
|
|
if not (1 <= len(description) <= 1024):
|
|
fail("SKILL.md description length is invalid")
|
|
for marker in ("RecoveryPatchSet", "fixed-channel-recovery-v1", "Never use"):
|
|
if marker not in description:
|
|
fail(f"SKILL.md description is missing trigger/boundary marker: {marker}")
|
|
|
|
|
|
def _validate_openai_yaml(text: str) -> None:
|
|
expected_keys = ("display_name", "short_description", "default_prompt")
|
|
lines = text.splitlines()
|
|
if not lines or lines[0] != "interface:":
|
|
fail("agents/openai.yaml must contain one interface mapping")
|
|
values: dict[str, str] = {}
|
|
for line in lines[1:]:
|
|
match = re.fullmatch(r' ([a-z_]+): "(.*)"', line)
|
|
if not match:
|
|
fail("agents/openai.yaml must use quoted scalar interface values")
|
|
values[match.group(1)] = match.group(2)
|
|
if tuple(values) != expected_keys:
|
|
fail("agents/openai.yaml contains unexpected or reordered interface fields")
|
|
if not 25 <= len(values["short_description"]) <= 64:
|
|
fail("agents/openai.yaml short_description must be 25-64 characters")
|
|
if f"${SKILL_NAME}" not in values["default_prompt"]:
|
|
fail("agents/openai.yaml default_prompt must explicitly invoke the Skill")
|
|
|
|
|
|
def validate_source(source: Path) -> list[Path]:
|
|
paths = _source_paths(source)
|
|
texts: dict[Path, str] = {}
|
|
for relative, path in zip(EXPECTED_RELATIVE_FILES, paths):
|
|
try:
|
|
text = path.read_text(encoding="utf-8")
|
|
except UnicodeDecodeError as error:
|
|
fail(f"Skill member is not UTF-8: {relative}: {error}")
|
|
if "TODO" in text or "[TODO" in text:
|
|
fail(f"Skill member contains an unfinished placeholder: {relative}")
|
|
texts[relative] = text
|
|
if relative.suffix == ".json":
|
|
try:
|
|
payload = json.loads(text)
|
|
except json.JSONDecodeError as error:
|
|
fail(f"Skill JSON member is invalid: {relative}: {error}")
|
|
if payload.get("type") != "object" or payload.get("additionalProperties") is not False:
|
|
fail(f"Skill JSON schema must be a closed top-level object: {relative}")
|
|
|
|
_validate_frontmatter(texts[Path("SKILL.md")])
|
|
_validate_openai_yaml(texts[Path("agents/openai.yaml")])
|
|
for relative, markers in REQUIRED_MARKERS.items():
|
|
missing = [marker for marker in markers if marker not in texts[relative]]
|
|
if missing:
|
|
fail(f"Skill member {relative} is missing markers: {', '.join(missing)}")
|
|
|
|
combined = "\n".join(texts.values())
|
|
for pattern, label in FORBIDDEN_TEXT_PATTERNS:
|
|
if pattern.search(combined):
|
|
fail(f"Skill source contains forbidden {label}")
|
|
return paths
|
|
|
|
|
|
def archive_member(source: Path, file_path: Path) -> str:
|
|
return f"{ARCHIVE_ROOT}/{file_path.relative_to(source).as_posix()}"
|
|
|
|
|
|
def write_archive(source: Path, output: Path) -> list[str]:
|
|
files = validate_source(source)
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as archive:
|
|
for file_path in files:
|
|
info = zipfile.ZipInfo(archive_member(source, file_path), FIXED_TIMESTAMP)
|
|
info.compress_type = zipfile.ZIP_DEFLATED
|
|
info.external_attr = 0o100644 << 16
|
|
archive.writestr(
|
|
info,
|
|
file_path.read_bytes(),
|
|
compress_type=zipfile.ZIP_DEFLATED,
|
|
compresslevel=9,
|
|
)
|
|
return [archive_member(source, file_path) for file_path in files]
|
|
|
|
|
|
def validate_archive(source: Path, archive_path: Path) -> list[str]:
|
|
files = validate_source(source)
|
|
expected_members = [archive_member(source, file_path) for file_path in files]
|
|
if not archive_path.is_file():
|
|
fail(f"Skill archive is missing: {archive_path}")
|
|
with zipfile.ZipFile(archive_path) as archive:
|
|
members = archive.namelist()
|
|
if members != expected_members:
|
|
fail("Skill archive member list or order differs from the frozen source")
|
|
for file_path, member in zip(files, members):
|
|
info = archive.getinfo(member)
|
|
if info.date_time != FIXED_TIMESTAMP:
|
|
fail(f"Skill archive member timestamp is not deterministic: {member}")
|
|
if info.compress_type != zipfile.ZIP_DEFLATED:
|
|
fail(f"Skill archive member compression is invalid: {member}")
|
|
if (info.external_attr >> 16) != 0o100644:
|
|
fail(f"Skill archive member permissions are invalid: {member}")
|
|
if archive.read(member) != file_path.read_bytes():
|
|
fail(f"Skill archive content differs from source: {member}")
|
|
return expected_members
|
|
|
|
|
|
def build_manifest(source: Path, archive: Path, members: list[str]) -> dict[str, object]:
|
|
files = validate_source(source)
|
|
return {
|
|
"manifest_schema_version": 1,
|
|
"artifact": archive.name,
|
|
"skill_name": SKILL_NAME,
|
|
"skill_version": SKILL_VERSION,
|
|
"publication_status": PUBLICATION_STATUS,
|
|
"publication_note": "Provider transport is not enabled; local validator remains authoritative.",
|
|
"sha256": sha256(archive),
|
|
"member_count": len(members),
|
|
"members": members,
|
|
"member_sha256": {
|
|
archive_member(source, file_path): sha256_bytes(file_path.read_bytes())
|
|
for file_path in files
|
|
},
|
|
"archive_timestamp_utc": FIXED_TIMESTAMP_UTC,
|
|
}
|
|
|
|
|
|
def render_manifest(manifest: dict[str, object]) -> bytes:
|
|
return (json.dumps(manifest, ensure_ascii=False, indent=2) + "\n").encode("utf-8")
|
|
|
|
|
|
def write_or_check_manifest(path: Path, manifest: dict[str, object], check: bool) -> None:
|
|
expected = render_manifest(manifest)
|
|
if check:
|
|
if not path.is_file() or path.read_bytes() != expected:
|
|
fail(f"Skill package manifest is stale or missing: {path}")
|
|
return
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(expected)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--source", required=True, type=Path, help="Editable Skill source directory")
|
|
parser.add_argument("--output", required=True, type=Path, help="Destination .skill archive")
|
|
parser.add_argument("--manifest", type=Path, help="Optional deterministic manifest")
|
|
parser.add_argument("--check", action="store_true", help="Check existing output without replacing it")
|
|
args = parser.parse_args()
|
|
|
|
source = args.source.resolve()
|
|
output = args.output.resolve()
|
|
if args.check:
|
|
with tempfile.TemporaryDirectory(prefix="field-recovery-skill-package-") as temporary_directory:
|
|
expected = Path(temporary_directory) / output.name
|
|
write_archive(source, expected)
|
|
if not output.is_file() or expected.read_bytes() != output.read_bytes():
|
|
fail("Skill archive is not reproducible from the editable source")
|
|
else:
|
|
write_archive(source, output)
|
|
|
|
members = validate_archive(source, output)
|
|
if args.manifest:
|
|
write_or_check_manifest(
|
|
args.manifest.resolve(),
|
|
build_manifest(source, output, members),
|
|
args.check,
|
|
)
|
|
print(f"Validated Recovery Skill archive: {output} ({len(members)} files, sha256={sha256(output)})")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except (OSError, ValueError, zipfile.BadZipFile) as error:
|
|
print(f"Packaging failed: {error}", file=sys.stderr)
|
|
raise SystemExit(1)
|