#!/usr/bin/env python3 """Validate Booking Agent Skill V2 source and optional HOLD artifact using stdlib only.""" from __future__ import annotations import argparse import hashlib import json import re import sys import zipfile from pathlib import Path from typing import Any EXPECTED_SCENARIO_COUNT = 49 REQUIRED_REFERENCES = { "00-boundaries-and-two-gates.md", "01-input-output-evidence.md", "02-target-and-source-splitting.md", "10-new-booking.md", "11-update-booking.md", "12-cancel-booking.md", "13-trace.md", "14-rooming-list.md", "15-allotment.md", "90-general-risk-ignored.md", "91-reason-codes-and-invariants.md", } REQUIRED_SCENARIOS = { "inbound_always_calls_agent", "new_explicit_row_complete", "new_duplicate_physical_rows_not_merged", "new_same_source_multiple_stays", "update_before_after", "update_after_only", "update_missing_target_value", "row_action_body_conflict", "cancel_cxl_group", "cancel_fit_name_only", "cancel_duplicate_rows_not_merged", "trace_concrete_no_action", "trace_multi_tour_copies", "trace_cross_message_never_merges", "trace_fo_hsk_precedence", "trace_extra_bed_default_one", "rooming_valid_excel", "rooming_body_only", "rooming_missing_tour", "rooming_two_files_same_tour", "rooming_one_file_multiple_tours", "allotment_name_only", "allotment_direct_deduction", "allotment_before_after_no_deduction", "allotment_two_actual_same_source_not_aggregated", "allotment_whole_cancel", "allotment_multiple_sources_ambiguous", "payment_only", "payment_plus_new", "task_plus_local_risk_plus_text", "wrong_direction_guard", } FORBIDDEN_FILE_PATTERNS = ( re.compile(r"rate-room-directory", re.IGNORECASE), re.compile(r"room-type-mapping", re.IGNORECASE), re.compile(r"\.eml$", re.IGNORECASE), re.compile(r"\.xlsx?$", re.IGNORECASE), ) FORBIDDEN_SCHEMA_TERMS = { '"PAYMENT"', '"CREATE_PAYMENT_NOTICE"', '"PAYMENT_TO_BOOKING"', '"CANCEL_ALLOTMENT_CONTROL_BLOCK"', '"source_old_room_items"', '"source_remaining_room_items"', '"contributions"', '"message_notifications"', } def fail(message: str) -> None: raise ValueError(message) def load_json(path: Path) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: fail(f"Cannot read JSON {path}: {error}") def sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def require_file(path: Path) -> None: if not path.is_file(): fail(f"Required file is missing: {path}") def frontmatter(skill: str) -> dict[str, str]: match = re.match(r"\A---\n(.*?)\n---\n", skill, re.DOTALL) if not match: fail("SKILL.md frontmatter is missing") result: dict[str, str] = {} for line in match.group(1).splitlines(): if ":" not in line: fail("SKILL.md frontmatter must be simple key/value YAML") key, value = line.split(":", 1) result[key.strip()] = value.strip() return result def validate_source(skill_root: Path, canonical_schemas: Path) -> None: for path in ( skill_root / "SKILL.md", skill_root / "agents/openai.yaml", skill_root / "manifest.json", skill_root / "prompts/main-prompt.md", skill_root / "fixtures/scenario-matrix.json", ): require_file(path) files = [path for path in skill_root.rglob("*") if path.is_file()] for path in files: relative = path.relative_to(skill_root).as_posix() if any(pattern.search(relative) for pattern in FORBIDDEN_FILE_PATTERNS): fail(f"Forbidden raw/mapping file in Skill source: {relative}") if path.name == ".DS_Store" or "__pycache__" in path.parts: fail(f"OS/cache file in Skill source: {relative}") skill_text = (skill_root / "SKILL.md").read_text(encoding="utf-8") metadata = frontmatter(skill_text) if set(metadata) != {"name", "description"}: fail("SKILL.md frontmatter may contain only name and description") if metadata["name"] != "booking-desk-event": fail("Unexpected Skill name") if len(metadata["description"]) < 80: fail("Skill description does not explain scope and trigger") boundary_sources = "\n".join(( skill_text, (skill_root / "references/00-boundaries-and-two-gates.md").read_text(encoding="utf-8"), (skill_root / "references/02-target-and-source-splitting.md").read_text(encoding="utf-8"), )) for marker in ( "每封进入预订流程的白名单入站邮件都执行一次", "booking-business-agent-compact-input-v1", "booking-business-agent-compact-decision-v1", "完整读取", "输出 Schema 是字段和允许值的唯一权威", "信息系统在 Layer 5B", "不得把 Payment 识别为任务", "不同物理来源单元", ): if marker not in boundary_sources: fail(f"Skill is missing boundary marker: {marker}") reference_names = {path.name for path in (skill_root / "references").glob("*.md")} missing_references = REQUIRED_REFERENCES - reference_names if missing_references: fail(f"Missing references: {', '.join(sorted(missing_references))}") skill_links = set(re.findall(r"references/([0-9A-Za-z._-]+\.md)", skill_text)) if not skill_links.issubset(reference_names): fail("SKILL.md links to a missing reference") chinese_files = [skill_root / "SKILL.md", skill_root / "agents/openai.yaml"] + sorted( (skill_root / "references").glob("*.md") ) for path in chinese_files: if not re.search(r"[\u4e00-\u9fff]", path.read_text(encoding="utf-8")): fail(f"Agent-readable Skill material is not Chinese: {path.relative_to(skill_root)}") prompt = (skill_root / "prompts/main-prompt.md").read_text(encoding="utf-8") if ( "$booking-desk-event" not in prompt or "只返回" not in prompt or "必要规则" not in prompt or "不调用其他工具" not in prompt or "只判断一次" not in prompt ): fail("Independent Main Prompt is incomplete") for redundant in ("你是 Wyndham-RSVN", "原样回显 `source_input_hash`", "decision_origin"): if redundant in prompt: fail(f"Main Prompt contains contract detail that belongs outside the prompt: {redundant}") business_markers = ("NEW_BOOKING", "UPDATE_BOOKING", "ALLOTMENT_SOURCE_DEDUCTION", "FO+HSK") if any(marker in prompt for marker in business_markers): fail("Main Prompt duplicates detailed business rules") source_manifest = load_json(skill_root / "manifest.json") expected_manifest = { "skill_release_version": "Booking Agent Skill V2", "publication_status": "HOLD", "distribution": "INTERNAL_ONLY", "main_prompt_packaged": False, "business_version": "booking-business-agent-v1.0", "input_contract": "booking-business-agent-compact-input-v1", "output_contract": "booking-business-agent-compact-decision-v1", "instruction_language": "zh-CN", } for key, value in expected_manifest.items(): if source_manifest.get(key) != value: fail(f"Editable manifest has unexpected {key}") for name in ( "booking-business-agent-compact-input-v1.schema.json", "booking-business-agent-compact-decision-v1.schema.json", ): skill_schema = skill_root / "schemas" / name canonical = canonical_schemas / name require_file(skill_schema) require_file(canonical) load_json(skill_schema) if skill_schema.read_bytes() != canonical.read_bytes(): fail(f"Skill schema is not byte-identical to canonical contract: {name}") candidate_text = (skill_root / "schemas/booking-business-agent-compact-decision-v1.schema.json").read_text( encoding="utf-8" ) for forbidden in FORBIDDEN_SCHEMA_TERMS: if forbidden in candidate_text: fail(f"Candidate schema contains retired term: {forbidden}") for required in ( '"risk_items"', '"unclassified_texts"', '"ALLOTMENT_SOURCE_WHOLE_CANCEL"', '"rooming_ref"', ): if required not in candidate_text: fail(f"Candidate schema is missing current term: {required}") candidate_schema = load_json(skill_root / "schemas/booking-business-agent-compact-decision-v1.schema.json") if "source_input_hash" in (candidate_schema.get("required") or []): fail("Compact output must not require source_input_hash") if "source_input_hash" in (candidate_schema.get("properties") or {}): fail("Compact output must not expose source_input_hash") trace_item = candidate_schema.get("$defs", {}).get("traceItem", {}) expected_trace_fields = { "service_type", "service_text", "room_type", "quantity", "source_refs", } if set(trace_item.get("required") or []) != expected_trace_fields: fail("Trace service item required fields do not match the approved contract") if set((trace_item.get("properties") or {}).keys()) != expected_trace_fields: fail("Trace service item contains an undeclared or retired field") if trace_item.get("additionalProperties") is not False: fail("Trace service item must reject undeclared fields") validate_fixtures(skill_root) def validate_fixtures(skill_root: Path) -> None: fixture_root = skill_root / "fixtures" matrix = load_json(fixture_root / "scenario-matrix.json") scenarios = matrix.get("scenarios") or [] ids = [item.get("scenario_id") for item in scenarios] if matrix.get("privacy") != "SYNTHETIC_NO_PII": fail("Scenario matrix is not marked synthetic") if matrix.get("scenario_count") != EXPECTED_SCENARIO_COUNT or len(scenarios) != EXPECTED_SCENARIO_COUNT: fail(f"Expected exactly {EXPECTED_SCENARIO_COUNT} scenarios") if len(set(ids)) != len(ids) or any(not isinstance(value, str) or not value for value in ids): fail("Scenario IDs must be unique non-empty strings") missing = REQUIRED_SCENARIOS - set(ids) if missing: fail(f"Scenario matrix is missing: {', '.join(sorted(missing))}") for item in scenarios: for key in ("category", "signal", "expected", "target_count", "derived_count", "risk_count", "unclassified_count"): if key not in item: fail(f"Scenario {item.get('scenario_id')} is missing {key}") def validate_artifact(archive: Path, release_manifest: Path) -> None: require_file(archive) require_file(release_manifest) manifest = load_json(release_manifest) if manifest.get("publication_status") != "HOLD" or manifest.get("distribution") != "INTERNAL_ONLY": fail("Release artifact is not HOLD/INTERNAL_ONLY") if manifest.get("instruction_language") != "zh-CN": fail("Release artifact instruction_language is not zh-CN") if manifest.get("main_prompt_packaged") is not False: fail("Release manifest says Main Prompt is packaged") prompt_hash = manifest.get("main_prompt_sha256") if not isinstance(prompt_hash, str) or not re.fullmatch(r"[0-9a-f]{64}", prompt_hash): fail("Release manifest is missing the independent Main Prompt hash") if manifest.get("sha256") != sha256(archive): fail("Release manifest artifact SHA-256 is stale") with zipfile.ZipFile(archive) as bundle: members = bundle.namelist() if members != manifest.get("members"): fail("Archive member list differs from release manifest") if any("/prompts/" in member or member.endswith("/main-prompt.md") for member in members): fail("Archive contains the independent Main Prompt") for member in members: expected = (manifest.get("member_sha256") or {}).get(member) if hashlib.sha256(bundle.read(member)).hexdigest() != expected: fail(f"Archive member hash mismatch: {member}") def main() -> int: parser = argparse.ArgumentParser(description=__doc__) repository_root = Path(__file__).resolve().parents[1] parser.add_argument("--skill-root", type=Path, required=True) parser.add_argument( "--canonical-schemas", type=Path, default=repository_root / "docs/project/requirements/schemas", ) parser.add_argument("--archive", type=Path) parser.add_argument("--release-manifest", type=Path) args = parser.parse_args() if (args.archive is None) != (args.release_manifest is None): fail("--archive and --release-manifest must be supplied together") validate_source(args.skill_root.resolve(), args.canonical_schemas.resolve()) if args.archive is not None and args.release_manifest is not None: validate_artifact(args.archive.resolve(), args.release_manifest.resolve()) print(f"Validated Booking Business Agent Skill source: {args.skill_root.resolve()}") return 0 if __name__ == "__main__": try: raise SystemExit(main()) except (OSError, ValueError, zipfile.BadZipFile) as error: print(f"Validation failed: {error}", file=sys.stderr) raise SystemExit(1)