#!/usr/bin/env python3 from __future__ import annotations import argparse import json import os import sys import tempfile import time from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] DEFAULT_REPORTS_ROOT = ROOT / "reports" / "1c-sql" SAVED_STATE_TABLES = {"ConfigSave", "ConfigCASSave"} SAVED_STATE_SOURCE_BY_TARGET = {"ConfigSave": "Config", "ConfigCASSave": "ConfigCAS"} def safe_path_segment(value: str) -> str: result = "".join(ch if ch.isalnum() or ch in "_.-" else "_" for ch in value) return result or "base" def duplicate_values(values: list[str]) -> list[str]: seen: set[str] = set() duplicates: set[str] = set() for value in values: if value in seen: duplicates.add(value) seen.add(value) return sorted(duplicates) def table_from_module_ref(value: Any) -> str | None: if not isinstance(value, str) or ":" not in value: return None table = value.split(":", 1)[0] return table if table in SAVED_STATE_TABLES else None def read_json(path: Path, failures: list[dict[str, Any]], label: str) -> dict[str, Any]: if not path.exists(): failures.append({"code": "report_missing", "label": label, "path": str(path)}) return {} try: data = json.loads(path.read_text(encoding="utf-8")) except Exception as exc: failures.append({"code": "report_invalid_json", "label": label, "path": str(path), "error": str(exc)}) return {} if not isinstance(data, dict): failures.append({"code": "report_not_object", "label": label, "path": str(path)}) return {} return data def expect_schema(report: dict[str, Any], failures: list[dict[str, Any]], label: str, path: Path, expected: str) -> None: actual = report.get("schema") if actual != expected: failures.append({ "code": "report_schema_unexpected", "label": label, "path": str(path), "expected": expected, "actual": actual, }) def expect_report_identity( report: dict[str, Any], failures: list[dict[str, Any]], label: str, path: Path, *, expected_base_id: str, expected_transport: str | None = None, expected_endpoint_url: str | None = None, ) -> None: if report.get("base_id") != expected_base_id: failures.append({ "code": "report_base_id_unexpected", "label": label, "path": str(path), "expected": expected_base_id, "actual": report.get("base_id"), }) if expected_transport is not None and report.get("transport") != expected_transport: failures.append({ "code": "report_transport_unexpected", "label": label, "path": str(path), "expected": expected_transport, "actual": report.get("transport"), }) if expected_endpoint_url is not None and report.get("endpoint_url") != expected_endpoint_url: failures.append({ "code": "report_endpoint_url_unexpected", "label": label, "path": str(path), "expected": expected_endpoint_url, "actual": report.get("endpoint_url"), }) def expect_report_fresh( failures: list[dict[str, Any]], label: str, path: Path, *, max_age_seconds: int | None, ) -> None: if max_age_seconds is None: return if max_age_seconds < 0: failures.append({"code": "report_max_age_invalid", "label": label, "path": str(path), "max_age_seconds": max_age_seconds}) return try: modified_at = path.stat().st_mtime except OSError as exc: failures.append({"code": "report_mtime_unavailable", "label": label, "path": str(path), "error": str(exc)}) return age_seconds = max(0.0, time.time() - modified_at) if age_seconds > max_age_seconds: failures.append({ "code": "report_stale", "label": label, "path": str(path), "max_age_seconds": max_age_seconds, "age_seconds": round(age_seconds, 3), }) def validate_selector_chain( path: Path, label: str, failures: list[dict[str, Any]], *, base_id: str, transport: str, endpoint_url: str | None, require_composition: bool, max_age_seconds: int | None, ) -> dict[str, Any]: report = read_json(path, failures, label) coverage = report.get("coverage") if isinstance(report.get("coverage"), dict) else {} resolve_overrides = coverage.get("resolve_overrides") if isinstance(coverage.get("resolve_overrides"), dict) else {} saved_state_resolution = coverage.get("saved_state_resolution") if isinstance(coverage.get("saved_state_resolution"), dict) else {} composition = coverage.get("write_plan_composition") if isinstance(coverage.get("write_plan_composition"), dict) else {} form_composition = coverage.get("form_write_plan_composition") if isinstance(coverage.get("form_write_plan_composition"), dict) else {} skips = coverage.get("skips") if isinstance(coverage.get("skips"), list) else None summary = { "path": str(path), "passed": report.get("passed"), "base_id": report.get("base_id"), "transport": report.get("transport"), "endpoint_url": report.get("endpoint_url"), "resolve_overrides_status": resolve_overrides.get("status"), "write_plan_evidence": resolve_overrides.get("write_plan_evidence"), "next_method": resolve_overrides.get("next_method"), "saved_state_status": saved_state_resolution.get("status"), "saved_state_modules": saved_state_resolution.get("modules"), "write_plan_target": saved_state_resolution.get("write_plan_target"), "composition_status": composition.get("status"), "composed": composition.get("composed"), "form_composition_status": form_composition.get("status"), "form_composed": form_composition.get("composed"), "form_name_first": form_composition.get("name_first"), } if not report: return summary expect_report_fresh(failures, label, path, max_age_seconds=max_age_seconds) expect_schema(report, failures, label, path, "onec_mcp_selector_chain_live_smoke.v1") expect_report_identity( report, failures, label, path, expected_base_id=base_id, expected_transport=transport, expected_endpoint_url=endpoint_url, ) if report.get("passed") is not True: failures.append({"code": "selector_chain_not_passed", "label": label, "path": str(path), "issues": report.get("issues")}) for section in ("resolve_overrides", "saved_state_resolution", "write_plan_composition", "form_write_plan_composition", "skips"): if section not in coverage: failures.append({"code": "selector_chain_coverage_section_missing", "label": label, "section": section, "path": str(path)}) if resolve_overrides.get("attempted") is not True: failures.append({"code": "selector_chain_resolve_overrides_not_attempted", "label": label, "path": str(path), "coverage": resolve_overrides}) if resolve_overrides.get("write_plan_evidence") is not True: failures.append({"code": "selector_chain_write_plan_evidence_missing", "label": label, "path": str(path), "coverage": resolve_overrides}) if resolve_overrides.get("next_method") != "metadata.saved_state.modules.search": failures.append({"code": "selector_chain_next_method_unexpected", "label": label, "path": str(path), "coverage": resolve_overrides}) if saved_state_resolution.get("attempted") is not True: failures.append({"code": "selector_chain_saved_state_resolution_not_attempted", "label": label, "path": str(path), "coverage": saved_state_resolution}) modules = saved_state_resolution.get("modules") if not isinstance(modules, int) or modules < 0: failures.append({"code": "selector_chain_saved_state_modules_invalid", "label": label, "path": str(path), "coverage": saved_state_resolution}) if not isinstance(saved_state_resolution.get("write_plan_target"), bool): failures.append({"code": "selector_chain_write_plan_target_not_boolean", "label": label, "path": str(path), "coverage": saved_state_resolution}) if composition.get("attempted") is not True: failures.append({"code": "selector_chain_write_plan_composition_not_attempted", "label": label, "path": str(path), "composition": composition}) if not isinstance(composition.get("composed"), bool): failures.append({"code": "selector_chain_composed_not_boolean", "label": label, "path": str(path), "composition": composition}) if saved_state_resolution.get("write_plan_target") is True and composition.get("composed") is not True: failures.append({ "code": "selector_chain_write_plan_target_not_composed", "label": label, "path": str(path), "saved_state_resolution": saved_state_resolution, "composition": composition, }) if composition.get("composed") is True and composition.get("status") in {None, "skipped_no_saved_state_target"}: failures.append({"code": "selector_chain_composed_status_unexpected", "label": label, "path": str(path), "composition": composition}) if not isinstance(form_composition.get("attempted"), bool): failures.append({"code": "selector_chain_form_composition_attempted_not_boolean", "label": label, "path": str(path), "composition": form_composition}) if not isinstance(form_composition.get("composed"), bool): failures.append({"code": "selector_chain_form_composed_not_boolean", "label": label, "path": str(path), "composition": form_composition}) if form_composition.get("candidate") is True and form_composition.get("composed") is not True: failures.append({"code": "selector_chain_form_target_not_composed", "label": label, "path": str(path), "composition": form_composition}) if form_composition.get("composed") is True and form_composition.get("name_first") is not True: failures.append({"code": "selector_chain_form_composition_not_name_first", "label": label, "path": str(path), "composition": form_composition}) if skips is None: failures.append({"code": "selector_chain_skips_not_list", "label": label, "path": str(path), "skips": coverage.get("skips")}) if require_composition and composition.get("composed") is not True: failures.append({"code": "selector_chain_composition_required", "label": label, "path": str(path), "composition": composition}) if require_composition and ( form_composition.get("composed") is not True or form_composition.get("name_first") is not True ): failures.append({"code": "selector_chain_form_composition_required", "label": label, "path": str(path), "composition": form_composition}) steps = report.get("steps") if isinstance(report.get("steps"), list) else [] for step in steps: if not isinstance(step, dict): continue if step.get("name") in {"metadata.resolve_overrides", "code.search"} and not str(step.get("status") or "").startswith("skipped"): if step.get("working_state") != "working": failures.append({"code": "selector_chain_working_state_unexpected", "label": label, "path": str(path), "step": step}) return summary def validate_write_plan_safety( path: Path, label: str, failures: list[dict[str, Any]], *, base_id: str, transport: str, endpoint_url: str | None, require_mcp_policy: bool = False, max_age_seconds: int | None = None, ) -> dict[str, Any]: report = read_json(path, failures, label) checks = report.get("checks") if isinstance(report.get("checks"), dict) else {} summary = { "path": str(path), "status": report.get("status"), "checks": len(checks), "base_id": report.get("base_id"), "transport": report.get("transport"), "endpoint_url": report.get("endpoint_url"), } if not report: return summary expect_report_fresh(failures, label, path, max_age_seconds=max_age_seconds) expect_schema(report, failures, label, path, "onec_write_plan_safety_smoke.v1") expect_report_identity( report, failures, label, path, expected_base_id=base_id, expected_transport=transport, expected_endpoint_url=endpoint_url, ) if report.get("status") != "ok": failures.append({"code": "write_plan_safety_not_ok", "label": label, "path": str(path), "status": report.get("status")}) if report.get("failures"): failures.append({"code": "write_plan_safety_failures_present", "label": label, "path": str(path), "failures": report.get("failures")}) for check in ( "blocked_effective_form_path", "replace_with_control_without_control_fragment", "replace_with_control_with_control_fragment", "replace_with_control_drift", ): if check not in checks: failures.append({"code": "write_plan_safety_check_missing", "label": label, "check": check, "path": str(path)}) expect_check( checks, failures, label, path, "blocked_effective_form_path", { "status": "blocked", "error": "write_plan_required", "routed_method": "metadata.write.plan", "target_kind": "form", }, ) expect_check( checks, failures, label, path, "replace_with_control_without_control_fragment", { "status": "blocked", "allowed": False, "problem_codes": ["missing_control_fragment"], }, ) expect_check( checks, failures, label, path, "replace_with_control_with_control_fragment", { "status": "planned", "allowed": True, "apply_method": "metadata.module.write_apply", "target_kind": "module", }, ) expect_check( checks, failures, label, path, "replace_with_control_drift", { "status": "blocked", "allowed": False, "problem_codes": ["control_fragment_drift"], }, ) if require_mcp_policy: for check in ("mcp.initialize", "mcp.blocks_missing_base_id", "mcp.blocks_diagnostic_fallback"): if check not in checks: failures.append({"code": "write_plan_safety_check_missing", "label": label, "check": check, "path": str(path)}) expect_check(checks, failures, label, path, "mcp.initialize", {"status": "ok"}) expect_check( checks, failures, label, path, "mcp.blocks_missing_base_id", {"status": "blocked", "reason": "base_id_required", "method": "metadata.write.plan"}, ) expect_check( checks, failures, label, path, "mcp.blocks_diagnostic_fallback", {"status": "blocked", "reason": "diagnostic_method", "method": "storage.files.list"}, ) return summary def validate_write_preflight( path: Path, label: str, failures: list[dict[str, Any]], *, base_id: str, transport: str, endpoint_url: str | None, require_mcp_initialize: bool = False, max_age_seconds: int | None = None, ) -> dict[str, Any]: report = read_json(path, failures, label) checks = report.get("checks") if isinstance(report.get("checks"), dict) else {} summary = { "path": str(path), "status": report.get("status"), "checks": len(checks), "base_id": report.get("base_id"), "transport": report.get("transport"), "endpoint_url": report.get("endpoint_url"), } if not report: return summary expect_report_fresh(failures, label, path, max_age_seconds=max_age_seconds) expect_schema(report, failures, label, path, "onec_write_preflight_smoke.v1") expect_report_identity( report, failures, label, path, expected_base_id=base_id, expected_transport=transport, expected_endpoint_url=endpoint_url, ) if report.get("status") != "ok": failures.append({"code": "write_preflight_not_ok", "label": label, "path": str(path), "status": report.get("status")}) if report.get("failures"): failures.append({"code": "write_preflight_failures_present", "label": label, "path": str(path), "failures": report.get("failures")}) for check in ("method_exposed", "effective_path_preflight", "concrete_saved_state_preflight"): if check not in checks: failures.append({"code": "write_preflight_check_missing", "label": label, "check": check, "path": str(path)}) expect_check(checks, failures, label, path, "method_exposed", {"status": "ok"}, failure_code="write_preflight_check_field_unexpected") expect_check( checks, failures, label, path, "effective_path_preflight", {"schema": "onec_metadata_write_preflight.v1", "allowed": False, "plan_allowed": False}, failure_code="write_preflight_check_field_unexpected", ) concrete = checks.get("concrete_saved_state_preflight") if isinstance(checks.get("concrete_saved_state_preflight"), dict) else {} if concrete.get("status") != "skipped_no_saved_module_target": expect_check( checks, failures, label, path, "concrete_saved_state_preflight", {"schema": "onec_metadata_write_preflight.v1", "writer": "metadata.module.write_apply"}, failure_code="write_preflight_check_field_unexpected", ) if require_mcp_initialize: if "mcp.initialize" not in checks: failures.append({"code": "write_preflight_check_missing", "label": label, "check": "mcp.initialize", "path": str(path)}) expect_check(checks, failures, label, path, "mcp.initialize", {"status": "ok"}, failure_code="write_preflight_check_field_unexpected") return summary def validate_write_rollback_safety( path: Path, label: str, failures: list[dict[str, Any]], *, base_id: str, transport: str, endpoint_url: str | None, require_mcp_initialize: bool = False, max_age_seconds: int | None = None, ) -> dict[str, Any]: report = read_json(path, failures, label) checks = report.get("checks") if isinstance(report.get("checks"), dict) else {} summary = { "path": str(path), "status": report.get("status"), "checks": len(checks), "base_id": report.get("base_id"), "transport": report.get("transport"), "endpoint_url": report.get("endpoint_url"), } if not report: return summary expect_report_fresh(failures, label, path, max_age_seconds=max_age_seconds) expect_schema(report, failures, label, path, "onec_write_rollback_safety_smoke.v1") expect_report_identity( report, failures, label, path, expected_base_id=base_id, expected_transport=transport, expected_endpoint_url=endpoint_url, ) if report.get("status") != "ok": failures.append({"code": "write_rollback_safety_not_ok", "label": label, "path": str(path), "status": report.get("status")}) if report.get("failures"): failures.append({"code": "write_rollback_safety_failures_present", "label": label, "path": str(path), "failures": report.get("failures")}) for check in ("method_exposed", "history_available", "rollback_without_gate_blocked"): if check not in checks: failures.append({"code": "write_rollback_safety_check_missing", "label": label, "check": check, "path": str(path)}) expect_check(checks, failures, label, path, "method_exposed", {"status": "ok", "method": "metadata.write.rollback"}, failure_code="write_rollback_safety_check_field_unexpected") expect_check(checks, failures, label, path, "history_available", {"status": "ok"}, failure_code="write_rollback_safety_check_field_unexpected") expect_check( checks, failures, label, path, "rollback_without_gate_blocked", { "status": "invalid_argument", "argument": "allow_sql_saved_state_rollback", "applied": None, "has_rollback_result": False, }, failure_code="write_rollback_safety_check_field_unexpected", ) if require_mcp_initialize: if "mcp.initialize" not in checks: failures.append({"code": "write_rollback_safety_check_missing", "label": label, "check": "mcp.initialize", "path": str(path)}) expect_check(checks, failures, label, path, "mcp.initialize", {"status": "ok"}, failure_code="write_rollback_safety_check_field_unexpected") return summary def validate_saved_state_diff_smoke( path: Path, label: str, failures: list[dict[str, Any]], *, base_id: str, transport: str, endpoint_url: str | None, saved_state_table: str, require_mcp_initialize: bool = False, max_age_seconds: int | None = None, ) -> dict[str, Any]: report = read_json(path, failures, label) checks = report.get("checks") if isinstance(report.get("checks"), dict) else {} existing = checks.get("diff_existing_saved_state") if isinstance(checks.get("diff_existing_saved_state"), dict) else {} missing = checks.get("diff_missing_saved_state") if isinstance(checks.get("diff_missing_saved_state"), dict) else {} summary = { "path": str(path), "status": report.get("status"), "checks": len(checks), "base_id": report.get("base_id"), "transport": report.get("transport"), "endpoint_url": report.get("endpoint_url"), "saved_state_table": report.get("saved_state_table"), "diff_status": existing.get("status") or missing.get("status"), "needs_prepare": existing.get("needs_prepare") if existing else missing.get("needs_prepare"), } if not report: return summary expect_report_fresh(failures, label, path, max_age_seconds=max_age_seconds) expect_schema(report, failures, label, path, "onec_saved_state_diff_smoke.v1") expect_report_identity( report, failures, label, path, expected_base_id=base_id, expected_transport=transport, expected_endpoint_url=endpoint_url, ) if report.get("saved_state_table") != saved_state_table: failures.append({"code": "saved_state_diff_table_unexpected", "label": label, "path": str(path), "expected": saved_state_table, "actual": report.get("saved_state_table")}) if report.get("status") != "ok": failures.append({"code": "saved_state_diff_not_ok", "label": label, "path": str(path), "status": report.get("status")}) if report.get("failures"): failures.append({"code": "saved_state_diff_failures_present", "label": label, "path": str(path), "failures": report.get("failures")}) for check in ("method_exposed",): if check not in checks: failures.append({"code": "saved_state_diff_check_missing", "label": label, "check": check, "path": str(path)}) expect_check(checks, failures, label, path, "method_exposed", {"status": "ok", "method": "metadata.saved_state.diff"}, failure_code="saved_state_diff_check_field_unexpected") if not existing and not missing: failures.append({"code": "saved_state_diff_target_check_missing", "label": label, "path": str(path)}) if existing: if existing.get("status") not in {"changed", "unchanged", "not_found"}: failures.append({"code": "saved_state_diff_status_unexpected", "label": label, "path": str(path), "check": "diff_existing_saved_state", "status": existing.get("status")}) if existing.get("status") in {"changed", "unchanged"}: expect_check(checks, failures, label, path, "diff_existing_saved_state", {"schema": "onec_saved_state_diff.v1", "needs_prepare": False, "freshness": "live_sql_verified"}, failure_code="saved_state_diff_check_field_unexpected") if missing: expect_check(checks, failures, label, path, "diff_missing_saved_state", {"schema": "onec_saved_state_diff.v1", "status": "not_found", "needs_prepare": True, "prepare_method": "metadata.saved_state.prepare"}, failure_code="saved_state_diff_check_field_unexpected") if require_mcp_initialize: if "mcp.initialize" not in checks: failures.append({"code": "saved_state_diff_check_missing", "label": label, "check": "mcp.initialize", "path": str(path)}) expect_check(checks, failures, label, path, "mcp.initialize", {"status": "ok"}, failure_code="saved_state_diff_check_field_unexpected") return summary def validate_saved_state_changes_smoke( path: Path, label: str, failures: list[dict[str, Any]], *, base_id: str, transport: str, endpoint_url: str | None, require_mcp_initialize: bool = False, max_age_seconds: int | None = None, ) -> dict[str, Any]: report = read_json(path, failures, label) checks = report.get("checks") if isinstance(report.get("checks"), dict) else {} changes = checks.get("changes_list") if isinstance(checks.get("changes_list"), dict) else {} summary = { "path": str(path), "status": report.get("status"), "checks": len(checks), "base_id": report.get("base_id"), "transport": report.get("transport"), "endpoint_url": report.get("endpoint_url"), "changes_status": changes.get("status"), "files": changes.get("files"), } if not report: return summary expect_report_fresh(failures, label, path, max_age_seconds=max_age_seconds) expect_schema(report, failures, label, path, "onec_saved_state_changes_smoke.v1") expect_report_identity( report, failures, label, path, expected_base_id=base_id, expected_transport=transport, expected_endpoint_url=endpoint_url, ) if report.get("status") != "ok": failures.append({"code": "saved_state_changes_not_ok", "label": label, "path": str(path), "status": report.get("status")}) if report.get("failures"): failures.append({"code": "saved_state_changes_failures_present", "label": label, "path": str(path), "failures": report.get("failures")}) for check in ("method_exposed", "changes_list", "changes_list_context"): if check not in checks: failures.append({"code": "saved_state_changes_check_missing", "label": label, "check": check, "path": str(path)}) expect_check(checks, failures, label, path, "method_exposed", {"status": "ok", "method": "metadata.saved_state.changes.list"}, failure_code="saved_state_changes_check_field_unexpected") expect_check( checks, failures, label, path, "changes_list", {"schema": "onec_saved_state_changes_list.v1", "freshness": "live_sql_verified", "verified_against_sql": True}, failure_code="saved_state_changes_check_field_unexpected", ) expect_check( checks, failures, label, path, "changes_list_context", {"schema": "onec_saved_state_changes_list.v1", "freshness": "live_sql_verified", "include_context": True, "group_by_context": True}, failure_code="saved_state_changes_check_field_unexpected", ) if require_mcp_initialize: if "mcp.initialize" not in checks: failures.append({"code": "saved_state_changes_check_missing", "label": label, "check": "mcp.initialize", "path": str(path)}) expect_check(checks, failures, label, path, "mcp.initialize", {"status": "ok"}, failure_code="saved_state_changes_check_field_unexpected") return summary def expect_check( checks: dict[str, Any], failures: list[dict[str, Any]], label: str, path: Path, check_name: str, expected: dict[str, Any], *, failure_code: str = "write_plan_safety_check_field_unexpected", ) -> None: check = checks.get(check_name) if not isinstance(check, dict): return for field, expected_value in expected.items(): actual_value = check.get(field) if isinstance(expected_value, list): if not isinstance(actual_value, list) or not all(item in actual_value for item in expected_value): failures.append({ "code": failure_code, "label": label, "path": str(path), "check": check_name, "field": field, "expected": expected_value, "actual": actual_value, }) elif actual_value != expected_value: failures.append({ "code": failure_code, "label": label, "path": str(path), "check": check_name, "field": field, "expected": expected_value, "actual": actual_value, }) def validate_saved_state_form( path: Path, failures: list[dict[str, Any]], *, base_id: str, require_write: bool, max_age_seconds: int | None, ) -> dict[str, Any]: label = "saved-state form write" report = read_json(path, failures, label) routes = report.get("routes") if isinstance(report.get("routes"), list) else [] preflight = report.get("saved_state_preflight") if isinstance(report.get("saved_state_preflight"), dict) else {} preflight_counts = preflight.get("counts") if isinstance(preflight.get("counts"), dict) else {} summary = { "path": str(path), "passed": report.get("passed"), "status": report.get("status"), "base_id": report.get("base_id"), "table": report.get("table"), "routes": len(routes), "preflight_status": preflight.get("status"), "preflight_counts": preflight_counts, } if not report: return summary expect_report_fresh(failures, label, path, max_age_seconds=max_age_seconds) expect_schema(report, failures, label, path, "onec_saved_state_write_routes_smoke.v1") expect_report_identity(report, failures, label, path, expected_base_id=base_id) if report.get("table") not in SAVED_STATE_TABLES: failures.append({"code": "saved_state_form_table_unexpected", "path": str(path), "table": report.get("table")}) if report.get("passed") is not True: failures.append({"code": "saved_state_form_not_passed", "path": str(path), "error": report.get("error")}) if require_write and report.get("status") == "skipped_no_saved_state": failures.append({"code": "saved_state_form_write_required", "path": str(path)}) status = report.get("status") if status not in {"skipped_no_saved_state", "verified_and_rolled_back"}: failures.append({"code": "saved_state_form_status_unexpected", "path": str(path), "status": status}) if status in {"skipped_no_saved_state", "verified_and_rolled_back"}: expected_count_keys = ("forms", "limit") if status == "skipped_no_saved_state" else ("forms", "scanned", "limit") validate_saved_state_preflight( preflight, failures, "saved_state_form", path, expected_count_keys=expected_count_keys, ) if status == "skipped_no_saved_state": if report.get("skipped") is not True: failures.append({"code": "saved_state_form_skip_flag_missing", "path": str(path), "skipped": report.get("skipped")}) if routes: failures.append({"code": "saved_state_form_skip_routes_present", "path": str(path), "routes": routes}) health = report.get("health") if isinstance(report.get("health"), dict) else {} live_sql = health.get("live_sql") if isinstance(health.get("live_sql"), dict) else {} if health.get("status") != "ok": failures.append({"code": "saved_state_form_health_not_ok", "path": str(path), "health": health}) if live_sql.get("configured") is not True: failures.append({"code": "saved_state_form_live_sql_not_configured", "path": str(path), "live_sql": live_sql}) if status == "verified_and_rolled_back" and not routes: failures.append({"code": "saved_state_form_routes_missing", "path": str(path)}) for index, route in enumerate(routes): if not isinstance(route, dict): failures.append({"code": "saved_state_form_route_not_object", "path": str(path), "index": index, "route": route}) continue if route.get("status") != "ok": failures.append({"code": "saved_state_form_route_status_unexpected", "path": str(path), "index": index, "route": route}) write_plan = route.get("write_plan") if isinstance(route.get("write_plan"), dict) else {} if write_plan.get("allowed") is not True: failures.append({"code": "saved_state_form_route_write_plan_not_allowed", "path": str(path), "index": index, "write_plan": write_plan}) if write_plan.get("apply_method") != "metadata.form.element.write_apply": failures.append({"code": "saved_state_form_route_apply_method_unexpected", "path": str(path), "index": index, "write_plan": write_plan}) if write_plan.get("target_kind") != "form": failures.append({"code": "saved_state_form_route_target_kind_unexpected", "path": str(path), "index": index, "write_plan": write_plan}) for field in ("write_path", "old", "post_rollback_old"): if field not in route: failures.append({"code": "saved_state_form_route_field_missing", "path": str(path), "index": index, "field": field, "route": route}) write_path = route.get("write_path") if isinstance(write_path, str) and ":" in write_path: route_table = write_path.split(":", 1)[0] if route_table in SAVED_STATE_TABLES and route_table != report.get("table"): failures.append({ "code": "saved_state_form_route_table_mismatch", "path": str(path), "index": index, "table": report.get("table"), "write_path": write_path, }) if "old" in route and "post_rollback_old" in route and route.get("old") != route.get("post_rollback_old"): failures.append({"code": "saved_state_form_route_rollback_value_unexpected", "path": str(path), "index": index, "route": route}) return summary def validate_saved_state_module( path: Path, failures: list[dict[str, Any]], *, base_id: str, require_write: bool, max_age_seconds: int | None, ) -> dict[str, Any]: label = "saved-state module write" report = read_json(path, failures, label) write_plan = report.get("write_plan") if isinstance(report.get("write_plan"), dict) else {} preflight = report.get("saved_state_preflight") if isinstance(report.get("saved_state_preflight"), dict) else {} preflight_counts = preflight.get("counts") if isinstance(preflight.get("counts"), dict) else {} table = report.get("table") or report.get("saved_state_table") or table_from_module_ref(report.get("module_ref")) skip_statuses = {"skipped_no_saved_state", "skipped_no_saved_state_candidate"} summary = { "path": str(path), "status": report.get("status"), "base_id": report.get("base_id"), "table": table, "module_ref": report.get("module_ref"), "write_plan_allowed": write_plan.get("allowed"), "preflight_status": preflight.get("status"), "preflight_counts": preflight_counts, } if not report: return summary expect_report_fresh(failures, label, path, max_age_seconds=max_age_seconds) expect_schema(report, failures, label, path, "onec_module_stream_write_smoke.v1") expect_report_identity(report, failures, label, path, expected_base_id=base_id) if table not in SAVED_STATE_TABLES: failures.append({"code": "saved_state_module_table_unexpected", "path": str(path), "module_ref": report.get("module_ref")}) if require_write and report.get("status") in skip_statuses: failures.append({"code": "saved_state_module_write_required", "path": str(path)}) if report.get("status") not in skip_statuses | {"verified_and_rolled_back"}: failures.append({"code": "saved_state_module_status_unexpected", "path": str(path), "status": report.get("status")}) if report.get("status") in {"skipped_no_saved_state", "verified_and_rolled_back"}: expected_count_keys = ("modules", "limit") if report.get("status") == "skipped_no_saved_state" else ("modules", "scanned", "limit") validate_saved_state_preflight( preflight, failures, "saved_state_module", path, expected_count_keys=expected_count_keys, ) if report.get("status") in skip_statuses: if report.get("skipped") is not True: failures.append({"code": "saved_state_module_skip_flag_missing", "path": str(path), "skipped": report.get("skipped")}) if report.get("status") == "skipped_no_saved_state_candidate": diagnostics = report.get("diagnostics") if isinstance(report.get("diagnostics"), dict) else {} if not str(diagnostics.get("message") or "").strip(): failures.append({"code": "saved_state_module_skip_diagnostics_missing", "path": str(path)}) if report.get("status") == "verified_and_rolled_back": if write_plan.get("allowed") is not True: failures.append({"code": "saved_state_module_write_plan_not_allowed", "path": str(path), "write_plan": write_plan}) if write_plan.get("apply_method") != "metadata.module.write_apply": failures.append({"code": "saved_state_module_apply_method_unexpected", "path": str(path), "write_plan": write_plan}) if write_plan.get("target_kind") != "module": failures.append({"code": "saved_state_module_target_kind_unexpected", "path": str(path), "write_plan": write_plan}) metadata_write = report.get("metadata_write") if isinstance(report.get("metadata_write"), dict) else {} if metadata_write.get("status") != "verified_and_rolled_back": failures.append({"code": "saved_state_module_metadata_write_status_unexpected", "path": str(path), "metadata_write": metadata_write}) if metadata_write.get("rolled_back") is not True: failures.append({"code": "saved_state_module_rollback_missing", "path": str(path), "metadata_write": metadata_write}) return summary def validate_code_write_saved_state( path: Path, failures: list[dict[str, Any]], *, base_id: str, require_write: bool, expected_transport: str | None, expected_endpoint_url: str | None, max_age_seconds: int | None, ) -> dict[str, Any]: label = "code.write saved-state" report = read_json(path, failures, label) steps = report.get("steps") if isinstance(report.get("steps"), list) else [] summary = { "path": str(path), "status": report.get("status"), "base_id": report.get("base_id"), "transport": report.get("transport"), "endpoint_url": report.get("endpoint_url"), "steps": len(steps), } if not report: return summary expect_report_fresh(failures, label, path, max_age_seconds=max_age_seconds) expect_schema(report, failures, label, path, "onec_code_write_saved_state_smoke.v1") expect_report_identity( report, failures, label, path, expected_base_id=base_id, expected_transport=expected_transport, expected_endpoint_url=expected_endpoint_url, ) if require_write and report.get("status") in {"skipped_missing_target", "skipped_write_gate"}: failures.append({"code": "code_write_saved_state_required", "path": str(path), "status": report.get("status")}) if report.get("status") not in {"ok", "skipped_missing_target", "skipped_write_gate"}: failures.append({"code": "code_write_saved_state_status_unexpected", "path": str(path), "status": report.get("status")}) if report.get("failures"): failures.append({"code": "code_write_saved_state_failures_present", "path": str(path), "failures": report.get("failures")}) if report.get("status") == "ok": write_steps = [step for step in steps if isinstance(step, dict) and step.get("name") == "code.write apply"] write_step = write_steps[0] if write_steps else {} write_mode = write_step.get("write_mode") if isinstance(write_step.get("write_mode"), dict) else {} if write_step.get("status") != "applied" or write_step.get("applied") is not True: failures.append({"code": "code_write_saved_state_apply_missing", "path": str(path), "step": write_step}) if write_mode.get("target") != "saved_state" or write_mode.get("activation_state") != "not_activated": failures.append({"code": "code_write_saved_state_write_mode_unexpected", "path": str(path), "write_mode": write_mode}) return summary def validate_agent_working_view( path: Path, failures: list[dict[str, Any]], *, base_id: str, expected_adapter_url: str | None, require_target: bool, max_age_seconds: int | None, ) -> dict[str, Any]: label = "agent working view" report = read_json(path, failures, label) save_forms = report.get("save_forms_only_names") if isinstance(report.get("save_forms_only_names"), list) else [] code_read_working = report.get("code_read_working") if isinstance(report.get("code_read_working"), dict) else {} code_read_both = report.get("code_read_both") if isinstance(report.get("code_read_both"), dict) else {} working_state = code_read_working.get("current_state") if isinstance(code_read_working.get("current_state"), dict) else {} summary = { "path": str(path), "status": report.get("status"), "base_id": report.get("base_id"), "adapter_url": report.get("adapter_url"), "saved_forms": len(save_forms), "working_source": working_state.get("source"), "both_text_source": code_read_both.get("text_source"), } if not report: return summary expect_report_fresh(failures, label, path, max_age_seconds=max_age_seconds) expect_schema(report, failures, label, path, "onec_agent_working_view_report.v1") if report.get("base_id") != base_id: failures.append({"code": "agent_working_view_base_id_unexpected", "path": str(path), "expected": base_id, "actual": report.get("base_id")}) if expected_adapter_url is not None and report.get("adapter_url") != expected_adapter_url: failures.append({"code": "agent_working_view_adapter_url_unexpected", "path": str(path), "expected": expected_adapter_url, "actual": report.get("adapter_url")}) if require_target and report.get("status") == "skipped_missing_target": failures.append({"code": "agent_working_view_required", "path": str(path), "status": report.get("status")}) if report.get("status") not in {"ok", "skipped_missing_target"}: failures.append({"code": "agent_working_view_status_unexpected", "path": str(path), "status": report.get("status")}) if report.get("failures"): failures.append({"code": "agent_working_view_failures_present", "path": str(path), "failures": report.get("failures")}) if report.get("status") == "ok": if not save_forms: failures.append({"code": "agent_working_view_saved_forms_missing", "path": str(path)}) if code_read_working.get("status") != "ok" or working_state.get("source") != "saved_state": failures.append({"code": "agent_working_view_working_source_unexpected", "path": str(path), "code_read_working": code_read_working}) if code_read_both.get("status") != "ok" or code_read_both.get("text_source") != "saved_state": failures.append({"code": "agent_working_view_both_source_unexpected", "path": str(path), "code_read_both": code_read_both}) return summary def validate_saved_state_preflight( preflight: dict[str, Any], failures: list[dict[str, Any]], label: str, path: Path, *, expected_count_keys: tuple[str, ...], ) -> None: if preflight.get("status") != "ok": failures.append({"code": f"{label}_preflight_not_ok", "path": str(path), "preflight": preflight}) counts = preflight.get("counts") if isinstance(preflight.get("counts"), dict) else {} for key in expected_count_keys: value = counts.get(key) if not isinstance(value, int) or value < 0: failures.append({"code": f"{label}_preflight_count_invalid", "path": str(path), "key": key, "value": value, "counts": counts}) def validate_saved_state_strict_readiness( path: Path, failures: list[dict[str, Any]], *, base_id: str, saved_state_table: str | None, require_ready: bool, max_age_seconds: int | None, ) -> dict[str, Any]: label = "saved-state strict readiness" report = read_json(path, failures, label) summary = report.get("summary") if isinstance(report.get("summary"), dict) else {} tables = report.get("tables") if isinstance(report.get("tables"), list) else [] result = { "path": str(path), "status": report.get("status"), "ready": report.get("ready"), "base_id": report.get("base_id"), "table": report.get("saved_state_table"), "tables": tables, "saved_state_rows": summary.get("saved_state_rows"), "forms": summary.get("forms"), "modules": summary.get("modules"), } if not report: return result expect_report_fresh(failures, label, path, max_age_seconds=max_age_seconds) expect_schema(report, failures, label, path, "onec_saved_state_strict_readiness.v1") expect_report_identity(report, failures, label, path, expected_base_id=base_id) if report.get("status") not in {"ready", "blocked_no_saved_state_rows", "blocked_no_strict_candidates"}: failures.append({"code": "saved_state_strict_readiness_status_unexpected", "path": str(path), "status": report.get("status")}) if not isinstance(report.get("ready"), bool): failures.append({"code": "saved_state_strict_readiness_ready_not_boolean", "path": str(path), "ready": report.get("ready")}) if saved_state_table is not None: if report.get("saved_state_table") != saved_state_table: failures.append({ "code": "saved_state_strict_readiness_table_unexpected", "path": str(path), "expected_table": saved_state_table, "actual_table": report.get("saved_state_table"), }) if tables != [saved_state_table]: failures.append({ "code": "saved_state_strict_readiness_tables_unexpected", "path": str(path), "expected_tables": [saved_state_table], "actual_tables": tables, }) for key in ("saved_state_rows", "forms", "modules"): value = summary.get(key) if not isinstance(value, int) or value < 0: failures.append({"code": "saved_state_strict_readiness_summary_invalid", "path": str(path), "key": key, "value": value}) if require_ready and report.get("ready") is not True: failures.append({"code": "saved_state_strict_readiness_required", "path": str(path), "status": report.get("status")}) return result def validate_saved_state_copy_plan( path: Path, failures: list[dict[str, Any]], *, base_id: str, max_age_seconds: int | None, ) -> dict[str, Any]: label = "saved-state copy plan" report = read_json(path, failures, label) source_rows = report.get("source_rows") if isinstance(report.get("source_rows"), list) else [] source_details = report.get("source_row_details") if isinstance(report.get("source_row_details"), dict) else {} target = report.get("target") if isinstance(report.get("target"), dict) else {} source_family = report.get("source_family") if isinstance(report.get("source_family"), dict) else {} target_collisions = report.get("target_collisions") if isinstance(report.get("target_collisions"), dict) else {} collision_rows = target_collisions.get("rows") if isinstance(target_collisions.get("rows"), list) else [] summary = report.get("summary") if isinstance(report.get("summary"), dict) else {} obj = report.get("object") if isinstance(report.get("object"), dict) else {} result = { "path": str(path), "status": report.get("status"), "ready_to_copy": report.get("ready_to_copy"), "base_id": report.get("base_id"), "object": { "guid": obj.get("guid"), "kind": obj.get("kind"), "name": obj.get("name"), }, "target_table": target.get("table"), "source_family": { "expected_source_table": source_family.get("expected_source_table"), "source_tables": source_family.get("source_tables"), "valid": source_family.get("valid"), }, "source_rows": len(source_rows), "found_source_storage_rows": summary.get("found_source_storage_rows"), "target_collision_status": target_collisions.get("status"), "target_collision_rows": len(collision_rows), } if not report: return result expect_report_fresh(failures, label, path, max_age_seconds=max_age_seconds) expect_schema(report, failures, label, path, "onec_saved_state_copy_plan.v1") expect_report_identity(report, failures, label, path, expected_base_id=base_id) if report.get("status") != "plan_ready": failures.append({"code": "saved_state_copy_plan_status_unexpected", "path": str(path), "status": report.get("status")}) if report.get("ready_to_copy") is not True: failures.append({"code": "saved_state_copy_plan_not_ready", "path": str(path), "ready_to_copy": report.get("ready_to_copy")}) if target.get("table") not in {"ConfigSave", "ConfigCASSave"}: failures.append({"code": "saved_state_copy_plan_target_table_unexpected", "path": str(path), "target": target}) expected_source_table = SAVED_STATE_SOURCE_BY_TARGET.get(str(target.get("table") or "")) if expected_source_table: if source_family.get("expected_source_table") != expected_source_table: failures.append({ "code": "saved_state_copy_plan_source_family_expected_unexpected", "path": str(path), "expected_source_table": expected_source_table, "source_family": source_family, }) if source_family.get("valid") is not True: failures.append({"code": "saved_state_copy_plan_source_family_invalid", "path": str(path), "source_family": source_family}) source_tables = source_family.get("source_tables") if isinstance(source_family.get("source_tables"), list) else [] if any(table != expected_source_table for table in source_tables): failures.append({ "code": "saved_state_copy_plan_source_family_mismatch", "path": str(path), "expected_source_table": expected_source_table, "source_tables": source_tables, }) if not (obj.get("guid") and obj.get("kind") and obj.get("name")): failures.append({"code": "saved_state_copy_plan_object_identity_missing", "path": str(path), "object": obj}) if not source_rows: failures.append({"code": "saved_state_copy_plan_source_rows_missing", "path": str(path)}) for index, row in enumerate(source_rows): if not isinstance(row, dict): failures.append({"code": "saved_state_copy_plan_source_row_not_object", "path": str(path), "index": index, "row": row}) continue if row.get("table") not in {"Config", "ConfigCAS"}: failures.append({"code": "saved_state_copy_plan_source_row_table_unexpected", "path": str(path), "index": index, "row": row}) elif expected_source_table and row.get("table") != expected_source_table: failures.append({ "code": "saved_state_copy_plan_source_row_family_mismatch", "path": str(path), "index": index, "expected_source_table": expected_source_table, "row": row, }) if not row.get("file_name"): failures.append({"code": "saved_state_copy_plan_source_row_file_name_missing", "path": str(path), "index": index, "row": row}) if not row.get("role"): failures.append({"code": "saved_state_copy_plan_source_row_role_missing", "path": str(path), "index": index, "row": row}) planned_rows = summary.get("planned_source_rows") if planned_rows != len(source_rows): failures.append({ "code": "saved_state_copy_plan_summary_mismatch", "path": str(path), "field": "planned_source_rows", "expected": len(source_rows), "actual": planned_rows, }) found_rows = summary.get("found_source_storage_rows") if not isinstance(found_rows, int) or found_rows <= 0: failures.append({"code": "saved_state_copy_plan_found_rows_invalid", "path": str(path), "found_source_storage_rows": found_rows}) if not source_details: failures.append({"code": "saved_state_copy_plan_source_details_missing", "path": str(path)}) if target_collisions.get("status") != "clear": failures.append({"code": "saved_state_copy_plan_target_collision_status_unexpected", "path": str(path), "target_collisions": target_collisions}) if collision_rows: failures.append({"code": "saved_state_copy_plan_target_collisions_present", "path": str(path), "rows": collision_rows}) if summary.get("target_collision_rows") != 0: failures.append({ "code": "saved_state_copy_plan_target_collision_count_unexpected", "path": str(path), "target_collision_rows": summary.get("target_collision_rows"), }) return result def validate_saved_state_sql_artifact( path: Path, failures: list[dict[str, Any]], *, base_id: str, saved_state_table: str | None, expected_schema: str, expected_count_key: str, label: str, max_age_seconds: int | None, ) -> dict[str, Any]: report = read_json(path, failures, label) result = { "path": str(path), "status": report.get("status"), "base_id": report.get("base_id"), "table": report.get("target_table"), "source_table": report.get("source_table"), "read_only": report.get("read_only"), "sql_write_performed": report.get("sql_write_performed"), expected_count_key: report.get(expected_count_key), } if not report: return result expect_report_fresh(failures, label, path, max_age_seconds=max_age_seconds) expect_schema(report, failures, label, path, expected_schema) if report.get("base_id") != base_id: failures.append({"code": "saved_state_sql_artifact_base_id_unexpected", "label": label, "path": str(path), "expected": base_id, "actual": report.get("base_id")}) if report.get("status") != "ready": failures.append({"code": "saved_state_sql_artifact_status_unexpected", "label": label, "path": str(path), "status": report.get("status")}) if report.get("read_only") is not True: failures.append({"code": "saved_state_sql_artifact_not_read_only", "label": label, "path": str(path), "read_only": report.get("read_only")}) if report.get("sql_write_performed") is not False: failures.append({"code": "saved_state_sql_artifact_write_performed", "label": label, "path": str(path), "sql_write_performed": report.get("sql_write_performed")}) if saved_state_table is not None and report.get("target_table") != saved_state_table: failures.append({"code": "saved_state_sql_artifact_table_unexpected", "label": label, "path": str(path), "expected_table": saved_state_table, "actual_table": report.get("target_table")}) expected_source_table = SAVED_STATE_SOURCE_BY_TARGET.get(str(report.get("target_table") or "")) if expected_source_table and report.get("source_table") != expected_source_table: failures.append({ "code": "saved_state_sql_artifact_source_table_unexpected", "label": label, "path": str(path), "expected_source_table": expected_source_table, "actual_source_table": report.get("source_table"), }) count = report.get(expected_count_key) if not isinstance(count, int) or count <= 0: failures.append({"code": "saved_state_sql_artifact_count_invalid", "label": label, "path": str(path), "field": expected_count_key, "value": count}) if report.get("failures") not in ([], None): failures.append({"code": "saved_state_sql_artifact_failures_present", "label": label, "path": str(path), "failures": report.get("failures")}) return result def validate_saved_state_table_consistency( reports: dict[str, Any], failures: list[dict[str, Any]], report_dir: Path, *, expected_table: str | None, ) -> None: copy_plan = reports.get("saved_state_copy_plan") if isinstance(reports.get("saved_state_copy_plan"), dict) else {} readiness = reports.get("saved_state_strict_readiness") if isinstance(reports.get("saved_state_strict_readiness"), dict) else {} form = reports.get("saved_state_form_write") if isinstance(reports.get("saved_state_form_write"), dict) else {} module = reports.get("saved_state_module_write") if isinstance(reports.get("saved_state_module_write"), dict) else {} prepare_sql = reports.get("saved_state_prepare_sql") if isinstance(reports.get("saved_state_prepare_sql"), dict) else {} cleanup_sql = reports.get("saved_state_cleanup_sql") if isinstance(reports.get("saved_state_cleanup_sql"), dict) else {} copy_plan_table = copy_plan.get("target_table") if copy_plan_table not in SAVED_STATE_TABLES: return if expected_table is not None and expected_table != copy_plan_table: failures.append({ "code": "saved_state_table_unexpected", "label": "saved_state_copy_plan", "path": str(copy_plan.get("path") or report_dir), "expected_table": expected_table, "actual_table": copy_plan_table, }) for label, report in ( ("saved_state_strict_readiness", readiness), ("saved_state_form_write", form), ("saved_state_module_write", module), ("saved_state_prepare_sql", prepare_sql), ("saved_state_cleanup_sql", cleanup_sql), ): table = report.get("table") if table is None: failures.append({ "code": "saved_state_table_missing", "label": label, "path": str(report_dir), "expected_table": copy_plan_table, }) elif table != copy_plan_table: failures.append({ "code": "saved_state_table_mismatch", "label": label, "path": str(report.get("path") or report_dir), "expected_table": copy_plan_table, "actual_table": table, }) def validate_base(base_id: str, report_dir: Path, args: argparse.Namespace) -> dict[str, Any]: failures: list[dict[str, Any]] = [] reports: dict[str, Any] = {} if not args.skip_rest: reports["selector_chain_rest"] = validate_selector_chain( report_dir / "selector-chain-rest-smoke.json", "REST selector-chain", failures, base_id=base_id, transport="rest", endpoint_url=args.rest_adapter_url, require_composition=args.require_selector_chain_write_plan_composition, max_age_seconds=args.max_report_age_seconds, ) if not args.skip_write_plan_safety_smoke: reports["write_plan_safety_rest"] = validate_write_plan_safety( report_dir / "write-plan-safety-smoke.json", "REST write-plan safety", failures, base_id=base_id, transport="rest", endpoint_url=args.rest_adapter_url, max_age_seconds=args.max_report_age_seconds, ) reports["write_preflight_rest"] = validate_write_preflight( report_dir / "write-preflight-smoke.json", "REST write preflight", failures, base_id=base_id, transport="rest", endpoint_url=args.rest_adapter_url, max_age_seconds=args.max_report_age_seconds, ) if not args.skip_write_rollback_safety_smoke: reports["write_rollback_safety_rest"] = validate_write_rollback_safety( report_dir / "write-rollback-safety-smoke.json", "REST write-rollback safety", failures, base_id=base_id, transport="rest", endpoint_url=args.rest_adapter_url, max_age_seconds=args.max_report_age_seconds, ) if not args.skip_saved_state_diff_smoke: reports["saved_state_diff_rest"] = validate_saved_state_diff_smoke( report_dir / "saved-state-diff-smoke.json", "REST saved-state diff", failures, base_id=base_id, transport="rest", endpoint_url=args.rest_adapter_url, saved_state_table=args.saved_state_table or "ConfigCASSave", max_age_seconds=args.max_report_age_seconds, ) reports["saved_state_changes_rest"] = validate_saved_state_changes_smoke( report_dir / "saved-state-changes-smoke.json", "REST saved-state changes", failures, base_id=base_id, transport="rest", endpoint_url=args.rest_adapter_url, max_age_seconds=args.max_report_age_seconds, ) if not args.skip_saved_state_write_smoke: reports["saved_state_strict_readiness"] = validate_saved_state_strict_readiness( report_dir / "saved-state-strict-readiness.json", failures, base_id=base_id, saved_state_table=args.saved_state_table or "ConfigCASSave", require_ready=args.require_saved_state_write_smoke, max_age_seconds=args.max_report_age_seconds, ) reports["saved_state_form_write"] = validate_saved_state_form( report_dir / "saved-state-write-routes-smoke.json", failures, base_id=base_id, require_write=args.require_saved_state_write_smoke, max_age_seconds=args.max_report_age_seconds, ) reports["saved_state_module_write"] = validate_saved_state_module( report_dir / "module-stream-write-smoke-script.json", failures, base_id=base_id, require_write=args.require_saved_state_write_smoke, max_age_seconds=args.max_report_age_seconds, ) readiness_ready = reports["saved_state_strict_readiness"].get("ready") is True if not args.skip_saved_state_copy_plan and not readiness_ready: reports["saved_state_copy_plan"] = validate_saved_state_copy_plan( report_dir / "saved-state-copy-plan.json", failures, base_id=base_id, max_age_seconds=args.max_report_age_seconds, ) reports["saved_state_prepare_sql"] = validate_saved_state_sql_artifact( report_dir / "prepare-saved-state-copy-sql.json", failures, base_id=base_id, saved_state_table=args.saved_state_table, expected_schema="onec_saved_state_copy_sql_plan.v1", expected_count_key="expected_insert_rows", label="saved-state prepare SQL artifact", max_age_seconds=args.max_report_age_seconds, ) reports["saved_state_cleanup_sql"] = validate_saved_state_sql_artifact( report_dir / "cleanup-saved-state-copy-sql.json", failures, base_id=base_id, saved_state_table=args.saved_state_table, expected_schema="onec_saved_state_cleanup_sql_plan.v1", expected_count_key="expected_delete_rows", label="saved-state cleanup SQL artifact", max_age_seconds=args.max_report_age_seconds, ) validate_saved_state_table_consistency( reports, failures, report_dir, expected_table=args.saved_state_table, ) if not args.skip_code_write_saved_state_smoke: reports["code_write_saved_state_rest"] = validate_code_write_saved_state( report_dir / "code-write-saved-state-rest-smoke.json", failures, base_id=base_id, require_write=args.require_code_write_saved_state_smoke, expected_transport="rest", expected_endpoint_url=args.rest_adapter_url, max_age_seconds=args.max_report_age_seconds, ) reports["agent_working_view"] = validate_agent_working_view( report_dir / "agent-working-view.json", failures, base_id=base_id, expected_adapter_url=args.rest_adapter_url, require_target=args.require_code_write_saved_state_smoke, max_age_seconds=args.max_report_age_seconds, ) if not args.skip_mcp: reports["selector_chain_mcp"] = validate_selector_chain( report_dir / "selector-chain-mcp-smoke.json", "MCP selector-chain", failures, base_id=base_id, transport="mcp", endpoint_url=args.mcp_url, require_composition=args.require_selector_chain_write_plan_composition, max_age_seconds=args.max_report_age_seconds, ) if not args.skip_write_plan_safety_smoke: reports["write_plan_safety_mcp"] = validate_write_plan_safety( report_dir / "write-plan-safety-mcp-smoke.json", "MCP write-plan safety", failures, base_id=base_id, transport="mcp", endpoint_url=args.mcp_url, require_mcp_policy=True, max_age_seconds=args.max_report_age_seconds, ) reports["write_preflight_mcp"] = validate_write_preflight( report_dir / "write-preflight-mcp-smoke.json", "MCP write preflight", failures, base_id=base_id, transport="mcp", endpoint_url=args.mcp_url, require_mcp_initialize=True, max_age_seconds=args.max_report_age_seconds, ) if not args.skip_write_rollback_safety_smoke: reports["write_rollback_safety_mcp"] = validate_write_rollback_safety( report_dir / "write-rollback-safety-mcp-smoke.json", "MCP write-rollback safety", failures, base_id=base_id, transport="mcp", endpoint_url=args.mcp_url, require_mcp_initialize=True, max_age_seconds=args.max_report_age_seconds, ) if not args.skip_saved_state_diff_smoke: reports["saved_state_diff_mcp"] = validate_saved_state_diff_smoke( report_dir / "saved-state-diff-mcp-smoke.json", "MCP saved-state diff", failures, base_id=base_id, transport="mcp", endpoint_url=args.mcp_url, saved_state_table=args.saved_state_table or "ConfigCASSave", require_mcp_initialize=True, max_age_seconds=args.max_report_age_seconds, ) reports["saved_state_changes_mcp"] = validate_saved_state_changes_smoke( report_dir / "saved-state-changes-mcp-smoke.json", "MCP saved-state changes", failures, base_id=base_id, transport="mcp", endpoint_url=args.mcp_url, require_mcp_initialize=True, max_age_seconds=args.max_report_age_seconds, ) if not args.skip_code_write_saved_state_smoke: reports["code_write_saved_state_mcp"] = validate_code_write_saved_state( report_dir / "code-write-saved-state-mcp-smoke.json", failures, base_id=base_id, require_write=args.require_code_write_saved_state_smoke, expected_transport="mcp", expected_endpoint_url=args.mcp_url, max_age_seconds=args.max_report_age_seconds, ) return {"base_id": base_id, "report_dir": str(report_dir), "passed": not failures, "reports": reports, "failures": failures} def write_json(path: Path, data: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def write_self_test_reports(report_dir: Path, *, base_id: str, composed: bool, saved_state_written: bool) -> None: endpoints = {"rest": "http://rest.self-test", "mcp": "http://mcp.self-test"} composition_status = "planned" if composed else "skipped_no_saved_state_target" selector_report = { "schema": "onec_mcp_selector_chain_live_smoke.v1", "passed": True, "base_id": base_id, "coverage": { "resolve_overrides": { "attempted": True, "status": "not_found", "write_plan_evidence": True, "next_method": "metadata.saved_state.modules.search", }, "saved_state_resolution": { "attempted": True, "status": "ok", "modules": 1 if composed else 0, "write_plan_target": composed, }, "write_plan_composition": { "attempted": True, "status": composition_status, "composed": composed, "from_write_plan_target": composed, }, "form_write_plan_composition": { "attempted": composed, "search_attempted": composed, "search_status": "ok" if composed else None, "candidate": composed, "status": "planned" if composed else None, "composed": composed, "name_first": composed, }, "skips": [] if composed else [{"step": "metadata.write.plan", "status": "skipped_no_saved_state_target"}], }, "steps": [ { "name": "metadata.resolve_overrides", "status": "not_found", "working_state": "working", }, { "name": "code.search", "status": "ok", "working_state": "working", }, ], "issues": [], } for name, transport in ( ("selector-chain-rest-smoke.json", "rest"), ("selector-chain-mcp-smoke.json", "mcp"), ): report = json.loads(json.dumps(selector_report)) report["transport"] = transport report["endpoint_url"] = endpoints[transport] write_json(report_dir / name, report) safety_report = { "schema": "onec_write_plan_safety_smoke.v1", "status": "ok", "base_id": base_id, "transport": "rest", "endpoint_url": endpoints["rest"], "checks": { "blocked_effective_form_path": { "status": "blocked", "error": "write_plan_required", "routed_method": "metadata.write.plan", "target_kind": "form", }, "replace_with_control_without_control_fragment": { "status": "blocked", "allowed": False, "problem_codes": ["missing_control_fragment"], }, "replace_with_control_with_control_fragment": { "status": "planned", "allowed": True, "apply_method": "metadata.module.write_apply", "target_kind": "module", }, "replace_with_control_drift": { "status": "blocked", "allowed": False, "problem_codes": ["control_fragment_drift"], }, }, "failures": [], } write_json(report_dir / "write-plan-safety-smoke.json", safety_report) mcp_safety_report = json.loads(json.dumps(safety_report)) mcp_safety_report["transport"] = "mcp" mcp_safety_report["endpoint_url"] = endpoints["mcp"] mcp_safety_report["checks"].update({ "mcp.initialize": {"status": "ok"}, "mcp.blocks_missing_base_id": { "status": "blocked", "reason": "base_id_required", "method": "metadata.write.plan", }, "mcp.blocks_diagnostic_fallback": { "status": "blocked", "reason": "diagnostic_method", "method": "storage.files.list", }, }) write_json(report_dir / "write-plan-safety-mcp-smoke.json", mcp_safety_report) preflight_report = { "schema": "onec_write_preflight_smoke.v1", "status": "ok", "base_id": base_id, "transport": "rest", "endpoint_url": endpoints["rest"], "checks": { "method_exposed": { "status": "ok", "method_count": 112, }, "effective_path_preflight": { "schema": "onec_metadata_write_preflight.v1", "status": "needs_resolution", "allowed": False, "plan_allowed": False, }, "concrete_saved_state_preflight": { "schema": "onec_metadata_write_preflight.v1", "status": "ready", "allowed": True, "saved_state_status": "changed", "freshness": "live_sql_verified", "writer": "metadata.module.write_apply", }, }, "failures": [], } write_json(report_dir / "write-preflight-smoke.json", preflight_report) mcp_preflight_report = json.loads(json.dumps(preflight_report)) mcp_preflight_report["transport"] = "mcp" mcp_preflight_report["endpoint_url"] = endpoints["mcp"] mcp_preflight_report["checks"]["mcp.initialize"] = {"status": "ok"} write_json(report_dir / "write-preflight-mcp-smoke.json", mcp_preflight_report) rollback_safety_report = { "schema": "onec_write_rollback_safety_smoke.v1", "status": "ok", "base_id": base_id, "transport": "rest", "endpoint_url": endpoints["rest"], "checks": { "method_exposed": { "status": "ok", "method_count": 109, "method": "metadata.write.rollback", }, "history_available": { "schema": "onec_metadata_write_history.v1", "status": "ok", "operations": 1, }, "rollback_without_gate_blocked": { "schema": "onec_adapter_request_error.v1", "status": "invalid_argument", "error": "invalid_argument", "argument": "allow_sql_saved_state_rollback", "applied": None, "has_rollback_result": False, }, }, "failures": [], } write_json(report_dir / "write-rollback-safety-smoke.json", rollback_safety_report) mcp_rollback_safety_report = json.loads(json.dumps(rollback_safety_report)) mcp_rollback_safety_report["transport"] = "mcp" mcp_rollback_safety_report["endpoint_url"] = endpoints["mcp"] mcp_rollback_safety_report["checks"]["mcp.initialize"] = {"status": "ok"} write_json(report_dir / "write-rollback-safety-mcp-smoke.json", mcp_rollback_safety_report) saved_state_diff_report = { "schema": "onec_saved_state_diff_smoke.v1", "status": "ok", "base_id": base_id, "transport": "rest", "endpoint_url": endpoints["rest"], "saved_state_table": "ConfigCASSave", "checks": { "method_exposed": { "status": "ok", "method_count": 110, "method": "metadata.saved_state.diff", }, }, "failures": [], } if saved_state_written: saved_state_diff_report["checks"]["diff_existing_saved_state"] = { "schema": "onec_saved_state_diff.v1", "status": "changed", "table": "ConfigCASSave", "file_name": "self-test-form", "needs_prepare": False, "freshness": "live_sql_verified", "current_source": "saved_state", } else: saved_state_diff_report["checks"]["diff_missing_saved_state"] = { "schema": "onec_saved_state_diff.v1", "status": "not_found", "table": "ConfigCASSave", "file_name": "self-test-missing", "needs_prepare": True, "prepare_method": "metadata.saved_state.prepare", } write_json(report_dir / "saved-state-diff-smoke.json", saved_state_diff_report) mcp_saved_state_diff_report = json.loads(json.dumps(saved_state_diff_report)) mcp_saved_state_diff_report["transport"] = "mcp" mcp_saved_state_diff_report["endpoint_url"] = endpoints["mcp"] mcp_saved_state_diff_report["checks"]["mcp.initialize"] = {"status": "ok"} write_json(report_dir / "saved-state-diff-mcp-smoke.json", mcp_saved_state_diff_report) saved_state_changes_report = { "schema": "onec_saved_state_changes_smoke.v1", "status": "ok", "base_id": base_id, "transport": "rest", "endpoint_url": endpoints["rest"], "checks": { "method_exposed": { "status": "ok", "method_count": 113, "method": "metadata.saved_state.changes.list", }, "changes_list": { "schema": "onec_saved_state_changes_list.v1", "status": "changed" if saved_state_written else "empty", "freshness": "live_sql_verified", "verified_against_sql": True, "tables": 2, "files": 1 if saved_state_written else 0, "changed_files": 1 if saved_state_written else 0, "saved_only_files": 0, "first_diff_method": "metadata.saved_state.diff" if saved_state_written else None, }, "changes_list_context": { "schema": "onec_saved_state_changes_list.v1", "status": "changed" if saved_state_written else "empty", "freshness": "live_sql_verified", "include_context": True, "group_by_context": True, "context_enrichment": True, "context_limit": 5, "files_with_context": 1 if saved_state_written else 0, "groups": 1 if saved_state_written else 0, "first_group_files": 1 if saved_state_written else None, "first_group_diff_selectors": 1 if saved_state_written else None, "first_group_next_actions": 1 if saved_state_written else None, "first_group_recommended_action": "inspect_diff" if saved_state_written else None, "first_group_action_total": 1 if saved_state_written else None, "root_recommended_action": "inspect_diff" if saved_state_written else None, "root_action_total": 1 if saved_state_written else None, }, }, "failures": [], } write_json(report_dir / "saved-state-changes-smoke.json", saved_state_changes_report) mcp_saved_state_changes_report = json.loads(json.dumps(saved_state_changes_report)) mcp_saved_state_changes_report["transport"] = "mcp" mcp_saved_state_changes_report["endpoint_url"] = endpoints["mcp"] mcp_saved_state_changes_report["checks"]["mcp.initialize"] = {"status": "ok"} write_json(report_dir / "saved-state-changes-mcp-smoke.json", mcp_saved_state_changes_report) form_report = { "schema": "onec_saved_state_write_routes_smoke.v1", "base_id": base_id, "table": "ConfigCASSave", "file_name": "self-test-form", "passed": True, "status": "verified_and_rolled_back" if saved_state_written else "skipped_no_saved_state", "skipped": not saved_state_written, "health": {"status": "ok", "live_sql": {"configured": True}}, "saved_state_preflight": { "status": "ok", "counts": {"forms": 1 if saved_state_written else 0, "scanned": 1 if saved_state_written else 0, "limit": 1}, }, "routes": [ { "name": "self-test", "status": "ok", "old": "before", "post_rollback_old": "before", "write_path": "ConfigCASSave:self-test", "write_plan": { "status": "planned", "allowed": True, "apply_method": "metadata.form.element.write_apply", "target_kind": "form", }, } ] if saved_state_written else [], } write_json(report_dir / "saved-state-write-routes-smoke.json", form_report) readiness_report = { "schema": "onec_saved_state_strict_readiness.v1", "base_url": "http://rest.self-test", "base_id": base_id, "saved_state_table": "ConfigCASSave", "tables": ["ConfigCASSave"], "ready": saved_state_written, "status": "ready" if saved_state_written else "blocked_no_saved_state_rows", "checks": { "health": {"status": "ok", "live_sql": {"configured": True}}, "row_counts": {"status": "ok", "counts": {"ConfigCASSave": 2 if saved_state_written else 0}}, "saved_state_search": { "forms": {"status": "ok", "counts": {"forms": 1 if saved_state_written else 0, "scanned": 1 if saved_state_written else 0, "limit": 3}}, "modules": {"status": "ok", "counts": {"modules": 1 if saved_state_written else 0, "scanned": 1 if saved_state_written else 0, "limit": 3}}, }, }, "recommendations": [], "summary": { "saved_state_rows": 2 if saved_state_written else 0, "forms": 1 if saved_state_written else 0, "modules": 1 if saved_state_written else 0, }, } write_json(report_dir / "saved-state-strict-readiness.json", readiness_report) module_report = { "schema": "onec_module_stream_write_smoke.v1", "status": "verified_and_rolled_back" if saved_state_written else "skipped_no_saved_state", "skipped": not saved_state_written, "base_id": base_id, "module_ref": "ConfigCASSave:self-test#stream:0", "saved_state_preflight": { "status": "ok", "counts": {"modules": 1 if saved_state_written else 0, "scanned": 1 if saved_state_written else 0, "limit": 1}, }, "write_plan": { "status": "planned", "allowed": True, "apply_method": "metadata.module.write_apply", "target_kind": "module", } if saved_state_written else {}, "metadata_write": { "status": "verified_and_rolled_back", "routed_method": "metadata.module.write_apply", "rolled_back": True, } if saved_state_written else {}, } write_json(report_dir / "module-stream-write-smoke-script.json", module_report) code_write_report = { "schema": "onec_code_write_saved_state_smoke.v1", "status": "ok" if saved_state_written else "skipped_missing_target", "skipped": not saved_state_written, "endpoint_url": "http://rest.self-test", "transport": "rest", "base_id": base_id, "target": {"object_type": "CommonForm", "object_name": "SelfTestForm", "routine_name": "SelfTest"}, "steps": [ {"name": "code.read working before", "status": "ok" if saved_state_written else "not_found"}, { "name": "code.write apply", "status": "applied", "applied": True, "write_mode": {"target": "saved_state", "activation_state": "not_activated", "production_apply": False}, }, ] if saved_state_written else [{"name": "code.read working before", "status": "not_found"}], "failures": [], } write_json(report_dir / "code-write-saved-state-rest-smoke.json", code_write_report) mcp_code_write_report = json.loads(json.dumps(code_write_report)) mcp_code_write_report["transport"] = "mcp" mcp_code_write_report["endpoint_url"] = "http://mcp.self-test" if saved_state_written: mcp_code_write_report["steps"] = [{"name": "mcp.initialize", "status": "ok", "contract_version": "onec-selector-contract.v1"}] + mcp_code_write_report["steps"] write_json(report_dir / "code-write-saved-state-mcp-smoke.json", mcp_code_write_report) agent_working_view_report = { "schema": "onec_agent_working_view_report.v1", "status": "ok" if saved_state_written else "skipped_missing_target", "skipped": not saved_state_written, "adapter_url": "http://rest.self-test", "base_id": base_id, "target": {"extension": "test2", "object_type": "CommonForm", "object_name": "SelfTestForm", "routine_name": "SelfTest"}, "save_forms_only_names": ["SelfTestForm"] if saved_state_written else [], "forms": { "working": [{"name": "SelfTestForm", "activation_state": "saved_only", "source": "extension_saved_state", "table": "ConfigCASSave"}] if saved_state_written else [], "save": [{"name": "SelfTestForm", "activation_state": "saved_only", "source": "extension_saved_state", "table": "ConfigCASSave"}] if saved_state_written else [], }, "code_read_working": { "status": "ok" if saved_state_written else "not_found", "current_state": {"source": "saved_state", "activation_state": "not_activated"} if saved_state_written else {}, "has_text": saved_state_written, }, "code_read_both": { "status": "ok" if saved_state_written else "not_found", "current_state": {"source": "both", "activation_state": "mixed"} if saved_state_written else {}, "text_source": "saved_state" if saved_state_written else None, "comparison": {"saved_status": "ok", "active_status": "not_found", "both_present": False, "differs": False} if saved_state_written else {}, "layers": [], }, "failures": [], } write_json(report_dir / "agent-working-view.json", agent_working_view_report) copy_plan = { "schema": "onec_saved_state_copy_plan.v1", "base_url": "http://rest.self-test", "base_id": base_id, "status": "plan_ready", "ready_to_copy": True, "selector": {"base_id": base_id, "include_storage": True}, "object": {"guid": "self-test-object", "kind": "Catalog", "name": "SelfTest"}, "target": {"table": "ConfigCASSave", "mode": "prepare_saved_state_working_copy"}, "source_family": { "target_table": "ConfigCASSave", "expected_source_table": "ConfigCAS", "source_tables": ["ConfigCAS"], "mismatched_source_tables": [], "valid": True, }, "source_rows": [ {"role": "object", "table": "ConfigCAS", "file_name": "self-test-object"}, {"role": "module", "table": "ConfigCAS", "file_name": "self-test-object.0"}, ], "source_row_details": { "ConfigCAS": { "status": "ok", "rows": [{"FileName": "self-test-object", "PartNo": 0}, {"FileName": "self-test-object.0", "PartNo": 0}], "counts": {"rows": 2}, } }, "target_collisions": {"status": "clear", "table": "ConfigCASSave", "rows": [], "counts": {"rows": 0}}, "summary": { "planned_source_rows": 2, "found_source_storage_rows": 2, "target_collision_rows": 0, "active_source_tables": ["ConfigCAS"], }, "recommendations": [], } write_json(report_dir / "saved-state-copy-plan.json", copy_plan) prepare_sql_report = { "schema": "onec_saved_state_copy_sql_plan.v1", "plan_path": str(report_dir / "saved-state-copy-plan.json"), "sql_path": str(report_dir / "prepare-saved-state-copy.sql"), "read_only": True, "sql_write_performed": False, "status": "ready", "failures": [], "base_id": base_id, "target_table": "ConfigCASSave", "source_table": "ConfigCAS", "expected_insert_rows": 2, } cleanup_sql_report = { "schema": "onec_saved_state_cleanup_sql_plan.v1", "plan_path": str(report_dir / "saved-state-copy-plan.json"), "sql_path": str(report_dir / "cleanup-saved-state-copy.sql"), "read_only": True, "sql_write_performed": False, "status": "ready", "failures": [], "base_id": base_id, "target_table": "ConfigCASSave", "source_table": "ConfigCAS", "expected_delete_rows": 2, } write_json(report_dir / "prepare-saved-state-copy-sql.json", prepare_sql_report) write_json(report_dir / "cleanup-saved-state-copy-sql.json", cleanup_sql_report) def validator_args(**overrides: Any) -> argparse.Namespace: defaults = { "skip_rest": False, "skip_mcp": False, "skip_write_plan_safety_smoke": False, "skip_write_rollback_safety_smoke": False, "skip_saved_state_diff_smoke": False, "skip_saved_state_write_smoke": False, "skip_code_write_saved_state_smoke": False, "skip_saved_state_copy_plan": False, "saved_state_table": None, "require_saved_state_write_smoke": False, "require_code_write_saved_state_smoke": False, "require_selector_chain_write_plan_composition": False, "rest_adapter_url": None, "mcp_url": None, "max_report_age_seconds": None, } defaults.update(overrides) return argparse.Namespace(**defaults) def run_self_test() -> dict[str, Any]: base_id = "self_test" failures: list[dict[str, Any]] = [] duplicate_failure_codes: list[str] = [] if duplicate_values(["self_test", "other", "self_test"]) == ["self_test"]: duplicate_failure_codes.append("duplicate_base_id") else: failures.append({"code": "self_test_duplicate_base_id_detection_failed"}) with tempfile.TemporaryDirectory(prefix="onec_verify_reports_") as tmp: reports_root = Path(tmp) report_dir = reports_root / safe_path_segment(base_id) write_self_test_reports(report_dir, base_id=base_id, composed=False, saved_state_written=False) soft = validate_base(base_id, report_dir, validator_args()) if soft["failures"]: failures.append({"code": "self_test_soft_unexpected_failures", "failures": soft["failures"]}) public_skip_dir = reports_root / "public_skip" write_self_test_reports(public_skip_dir, base_id="public_skip", composed=False, saved_state_written=False) public_form_path = public_skip_dir / "saved-state-write-routes-smoke.json" public_form_report = json.loads(public_form_path.read_text(encoding="utf-8")) public_form_report["saved_state_preflight"]["counts"].pop("scanned", None) write_json(public_form_path, public_form_report) public_module_path = public_skip_dir / "module-stream-write-smoke-script.json" public_module_report = json.loads(public_module_path.read_text(encoding="utf-8")) public_module_report.update({ "status": "skipped_no_saved_state_candidate", "table": "ConfigCASSave", "diagnostics": {"message": "no safe saved-state BSL module stream candidates found"}, }) public_module_report.pop("module_ref", None) public_module_report.pop("saved_state_preflight", None) public_module_report.pop("write_plan", None) public_module_report.pop("metadata_write", None) write_json(public_module_path, public_module_report) public_skip = validate_base("public_skip", public_skip_dir, validator_args()) if public_skip["failures"]: failures.append({"code": "self_test_public_skip_unexpected_failures", "failures": public_skip["failures"]}) public_skip_strict = validate_base( "public_skip", public_skip_dir, validator_args(require_saved_state_write_smoke=True), ) public_skip_strict_codes = {str(failure.get("code")) for failure in public_skip_strict["failures"]} if "saved_state_module_write_required" not in public_skip_strict_codes: failures.append({ "code": "self_test_public_skip_strict_expected_failure_missing", "expected": "saved_state_module_write_required", "actual": sorted(public_skip_strict_codes), }) strict_skip = validate_base( base_id, report_dir, validator_args( require_saved_state_write_smoke=True, require_selector_chain_write_plan_composition=True, ), ) strict_codes = {str(failure.get("code")) for failure in strict_skip["failures"]} for expected in ( "selector_chain_composition_required", "selector_chain_form_composition_required", "saved_state_form_write_required", "saved_state_module_write_required", ): if expected not in strict_codes: failures.append({"code": "self_test_strict_expected_failure_missing", "expected": expected, "actual": sorted(strict_codes)}) broken_dir = reports_root / "broken_coverage" write_self_test_reports(broken_dir, base_id="broken_coverage", composed=False, saved_state_written=False) for name in ("selector-chain-rest-smoke.json", "selector-chain-mcp-smoke.json"): path = broken_dir / name report = json.loads(path.read_text(encoding="utf-8")) coverage = report["coverage"] coverage["resolve_overrides"]["attempted"] = False coverage["resolve_overrides"]["write_plan_evidence"] = False coverage["resolve_overrides"]["next_method"] = "metadata.modules.read" coverage["saved_state_resolution"]["attempted"] = False coverage["saved_state_resolution"]["modules"] = -1 coverage["saved_state_resolution"]["write_plan_target"] = "true" coverage["write_plan_composition"]["attempted"] = False coverage["write_plan_composition"]["composed"] = "false" coverage["skips"] = {} for step in report.get("steps") or []: if isinstance(step, dict) and step.get("name") == "code.search": step["working_state"] = "active" write_json(path, report) coverage_check = validate_base("broken_coverage", broken_dir, validator_args()) coverage_codes = {str(failure.get("code")) for failure in coverage_check["failures"]} for expected in ( "selector_chain_resolve_overrides_not_attempted", "selector_chain_write_plan_evidence_missing", "selector_chain_next_method_unexpected", "selector_chain_saved_state_resolution_not_attempted", "selector_chain_saved_state_modules_invalid", "selector_chain_write_plan_target_not_boolean", "selector_chain_write_plan_composition_not_attempted", "selector_chain_composed_not_boolean", "selector_chain_skips_not_list", "selector_chain_working_state_unexpected", ): if expected not in coverage_codes: failures.append({"code": "self_test_coverage_expected_failure_missing", "expected": expected, "actual": sorted(coverage_codes)}) inconsistent_dir = reports_root / "inconsistent_coverage" write_self_test_reports(inconsistent_dir, base_id="inconsistent_coverage", composed=False, saved_state_written=False) rest_report_path = inconsistent_dir / "selector-chain-rest-smoke.json" rest_report = json.loads(rest_report_path.read_text(encoding="utf-8")) rest_coverage = rest_report["coverage"] rest_coverage["saved_state_resolution"]["modules"] = 1 rest_coverage["saved_state_resolution"]["write_plan_target"] = True rest_coverage["write_plan_composition"]["composed"] = False rest_coverage["write_plan_composition"]["status"] = "skipped_no_saved_state_target" write_json(rest_report_path, rest_report) mcp_report_path = inconsistent_dir / "selector-chain-mcp-smoke.json" mcp_report = json.loads(mcp_report_path.read_text(encoding="utf-8")) mcp_coverage = mcp_report["coverage"] mcp_coverage["saved_state_resolution"]["modules"] = 1 mcp_coverage["saved_state_resolution"]["write_plan_target"] = True mcp_coverage["write_plan_composition"]["composed"] = True mcp_coverage["write_plan_composition"]["status"] = "skipped_no_saved_state_target" write_json(mcp_report_path, mcp_report) consistency_check = validate_base("inconsistent_coverage", inconsistent_dir, validator_args()) consistency_codes = {str(failure.get("code")) for failure in consistency_check["failures"]} for expected in ( "selector_chain_write_plan_target_not_composed", "selector_chain_composed_status_unexpected", ): if expected not in consistency_codes: failures.append({"code": "self_test_consistency_expected_failure_missing", "expected": expected, "actual": sorted(consistency_codes)}) broken_safety_dir = reports_root / "broken_safety" write_self_test_reports(broken_safety_dir, base_id="broken_safety", composed=False, saved_state_written=False) rest_safety_path = broken_safety_dir / "write-plan-safety-smoke.json" rest_safety = json.loads(rest_safety_path.read_text(encoding="utf-8")) rest_safety["checks"]["blocked_effective_form_path"]["routed_method"] = "metadata.module.write" rest_safety["checks"]["replace_with_control_with_control_fragment"]["allowed"] = False rest_safety["checks"]["replace_with_control_drift"]["problem_codes"] = ["other"] write_json(rest_safety_path, rest_safety) mcp_safety_path = broken_safety_dir / "write-plan-safety-mcp-smoke.json" mcp_safety = json.loads(mcp_safety_path.read_text(encoding="utf-8")) mcp_safety["checks"].pop("mcp.blocks_diagnostic_fallback", None) mcp_safety["checks"]["mcp.blocks_missing_base_id"]["reason"] = "other" write_json(mcp_safety_path, mcp_safety) safety_check = validate_base("broken_safety", broken_safety_dir, validator_args()) safety_codes = {str(failure.get("code")) for failure in safety_check["failures"]} for expected in ( "write_plan_safety_check_field_unexpected", "write_plan_safety_check_missing", ): if expected not in safety_codes: failures.append({"code": "self_test_safety_expected_failure_missing", "expected": expected, "actual": sorted(safety_codes)}) broken_preflight_dir = reports_root / "broken_preflight" write_self_test_reports(broken_preflight_dir, base_id="broken_preflight", composed=False, saved_state_written=False) rest_preflight_path = broken_preflight_dir / "write-preflight-smoke.json" rest_preflight = json.loads(rest_preflight_path.read_text(encoding="utf-8")) rest_preflight["checks"]["method_exposed"]["status"] = "missing" rest_preflight["checks"]["effective_path_preflight"]["allowed"] = True write_json(rest_preflight_path, rest_preflight) mcp_preflight_path = broken_preflight_dir / "write-preflight-mcp-smoke.json" mcp_preflight = json.loads(mcp_preflight_path.read_text(encoding="utf-8")) mcp_preflight["checks"].pop("mcp.initialize", None) mcp_preflight["checks"].pop("concrete_saved_state_preflight", None) write_json(mcp_preflight_path, mcp_preflight) preflight_check = validate_base("broken_preflight", broken_preflight_dir, validator_args()) preflight_codes = {str(failure.get("code")) for failure in preflight_check["failures"]} for expected in ( "write_preflight_check_field_unexpected", "write_preflight_check_missing", ): if expected not in preflight_codes: failures.append({"code": "self_test_preflight_expected_failure_missing", "expected": expected, "actual": sorted(preflight_codes)}) broken_rollback_safety_dir = reports_root / "broken_rollback_safety" write_self_test_reports(broken_rollback_safety_dir, base_id="broken_rollback_safety", composed=False, saved_state_written=False) rest_rollback_path = broken_rollback_safety_dir / "write-rollback-safety-smoke.json" rest_rollback = json.loads(rest_rollback_path.read_text(encoding="utf-8")) rest_rollback["checks"]["method_exposed"]["status"] = "missing" rest_rollback["checks"]["rollback_without_gate_blocked"]["status"] = "applied" rest_rollback["checks"]["rollback_without_gate_blocked"]["applied"] = True rest_rollback["checks"]["rollback_without_gate_blocked"]["has_rollback_result"] = True write_json(rest_rollback_path, rest_rollback) mcp_rollback_path = broken_rollback_safety_dir / "write-rollback-safety-mcp-smoke.json" mcp_rollback = json.loads(mcp_rollback_path.read_text(encoding="utf-8")) mcp_rollback["checks"].pop("mcp.initialize", None) mcp_rollback["checks"]["history_available"]["status"] = "error" write_json(mcp_rollback_path, mcp_rollback) rollback_safety_check = validate_base("broken_rollback_safety", broken_rollback_safety_dir, validator_args()) rollback_safety_codes = {str(failure.get("code")) for failure in rollback_safety_check["failures"]} for expected in ( "write_rollback_safety_check_field_unexpected", "write_rollback_safety_check_missing", ): if expected not in rollback_safety_codes: failures.append({"code": "self_test_rollback_safety_expected_failure_missing", "expected": expected, "actual": sorted(rollback_safety_codes)}) broken_saved_state_diff_dir = reports_root / "broken_saved_state_diff" write_self_test_reports(broken_saved_state_diff_dir, base_id="broken_saved_state_diff", composed=False, saved_state_written=False) rest_diff_path = broken_saved_state_diff_dir / "saved-state-diff-smoke.json" rest_diff = json.loads(rest_diff_path.read_text(encoding="utf-8")) rest_diff["checks"]["method_exposed"]["status"] = "missing" rest_diff["checks"]["diff_missing_saved_state"]["needs_prepare"] = False rest_diff["checks"]["diff_missing_saved_state"]["prepare_method"] = "other" write_json(rest_diff_path, rest_diff) mcp_diff_path = broken_saved_state_diff_dir / "saved-state-diff-mcp-smoke.json" mcp_diff = json.loads(mcp_diff_path.read_text(encoding="utf-8")) mcp_diff["checks"].pop("mcp.initialize", None) write_json(mcp_diff_path, mcp_diff) saved_state_diff_check = validate_base("broken_saved_state_diff", broken_saved_state_diff_dir, validator_args()) saved_state_diff_codes = {str(failure.get("code")) for failure in saved_state_diff_check["failures"]} for expected in ( "saved_state_diff_check_field_unexpected", "saved_state_diff_check_missing", ): if expected not in saved_state_diff_codes: failures.append({"code": "self_test_saved_state_diff_expected_failure_missing", "expected": expected, "actual": sorted(saved_state_diff_codes)}) broken_saved_state_changes_dir = reports_root / "broken_saved_state_changes" write_self_test_reports(broken_saved_state_changes_dir, base_id="broken_saved_state_changes", composed=False, saved_state_written=False) rest_changes_path = broken_saved_state_changes_dir / "saved-state-changes-smoke.json" rest_changes = json.loads(rest_changes_path.read_text(encoding="utf-8")) rest_changes["checks"]["method_exposed"]["status"] = "missing" rest_changes["checks"]["changes_list"]["freshness"] = "cache_hit_stale" write_json(rest_changes_path, rest_changes) mcp_changes_path = broken_saved_state_changes_dir / "saved-state-changes-mcp-smoke.json" mcp_changes = json.loads(mcp_changes_path.read_text(encoding="utf-8")) mcp_changes["checks"].pop("mcp.initialize", None) mcp_changes["checks"].pop("changes_list", None) write_json(mcp_changes_path, mcp_changes) saved_state_changes_check = validate_base("broken_saved_state_changes", broken_saved_state_changes_dir, validator_args()) saved_state_changes_codes = {str(failure.get("code")) for failure in saved_state_changes_check["failures"]} for expected in ( "saved_state_changes_check_field_unexpected", "saved_state_changes_check_missing", ): if expected not in saved_state_changes_codes: failures.append({"code": "self_test_saved_state_changes_expected_failure_missing", "expected": expected, "actual": sorted(saved_state_changes_codes)}) broken_schema_dir = reports_root / "broken_schema" write_self_test_reports(broken_schema_dir, base_id="broken_schema", composed=False, saved_state_written=False) for name in ( "selector-chain-rest-smoke.json", "selector-chain-mcp-smoke.json", "write-plan-safety-smoke.json", "write-plan-safety-mcp-smoke.json", "write-preflight-smoke.json", "write-preflight-mcp-smoke.json", "saved-state-changes-smoke.json", "saved-state-changes-mcp-smoke.json", "saved-state-write-routes-smoke.json", "module-stream-write-smoke-script.json", "code-write-saved-state-rest-smoke.json", "code-write-saved-state-mcp-smoke.json", ): path = broken_schema_dir / name report = json.loads(path.read_text(encoding="utf-8")) report["schema"] = "old.schema" write_json(path, report) schema_check = validate_base("broken_schema", broken_schema_dir, validator_args()) schema_codes = {str(failure.get("code")) for failure in schema_check["failures"]} if "report_schema_unexpected" not in schema_codes: failures.append({"code": "self_test_schema_expected_failure_missing", "expected": "report_schema_unexpected", "actual": sorted(schema_codes)}) broken_identity_dir = reports_root / "broken_identity" write_self_test_reports(broken_identity_dir, base_id="broken_identity", composed=False, saved_state_written=False) for name in ( "selector-chain-rest-smoke.json", "write-plan-safety-smoke.json", "saved-state-write-routes-smoke.json", "module-stream-write-smoke-script.json", "code-write-saved-state-rest-smoke.json", "code-write-saved-state-mcp-smoke.json", ): path = broken_identity_dir / name report = json.loads(path.read_text(encoding="utf-8")) report["base_id"] = "other_base" write_json(path, report) for name in ("selector-chain-mcp-smoke.json", "write-plan-safety-mcp-smoke.json"): path = broken_identity_dir / name report = json.loads(path.read_text(encoding="utf-8")) report["transport"] = "rest" write_json(path, report) identity_check = validate_base("broken_identity", broken_identity_dir, validator_args()) identity_codes = {str(failure.get("code")) for failure in identity_check["failures"]} for expected in ("report_base_id_unexpected", "report_transport_unexpected"): if expected not in identity_codes: failures.append({"code": "self_test_identity_expected_failure_missing", "expected": expected, "actual": sorted(identity_codes)}) broken_endpoint_dir = reports_root / "broken_endpoint" write_self_test_reports(broken_endpoint_dir, base_id="broken_endpoint", composed=False, saved_state_written=False) for name in ( "selector-chain-rest-smoke.json", "selector-chain-mcp-smoke.json", "write-plan-safety-smoke.json", "write-plan-safety-mcp-smoke.json", ): path = broken_endpoint_dir / name report = json.loads(path.read_text(encoding="utf-8")) report["endpoint_url"] = "http://other.self-test" write_json(path, report) endpoint_check = validate_base( "broken_endpoint", broken_endpoint_dir, validator_args(rest_adapter_url="http://rest.self-test", mcp_url="http://mcp.self-test"), ) endpoint_codes = {str(failure.get("code")) for failure in endpoint_check["failures"]} if "report_endpoint_url_unexpected" not in endpoint_codes: failures.append({"code": "self_test_endpoint_expected_failure_missing", "expected": "report_endpoint_url_unexpected", "actual": sorted(endpoint_codes)}) stale_dir = reports_root / "stale" write_self_test_reports(stale_dir, base_id="stale", composed=False, saved_state_written=False) stale_mtime = time.time() - 120 for path in stale_dir.glob("*.json"): os.utime(path, (stale_mtime, stale_mtime)) stale_check = validate_base("stale", stale_dir, validator_args(max_report_age_seconds=1)) stale_codes = {str(failure.get("code")) for failure in stale_check["failures"]} if "report_stale" not in stale_codes: failures.append({"code": "self_test_stale_expected_failure_missing", "expected": "report_stale", "actual": sorted(stale_codes)}) broken_saved_state_dir = reports_root / "broken_saved_state" write_self_test_reports(broken_saved_state_dir, base_id="broken_saved_state", composed=True, saved_state_written=True) form_path = broken_saved_state_dir / "saved-state-write-routes-smoke.json" form_report = json.loads(form_path.read_text(encoding="utf-8")) form_report["saved_state_preflight"]["counts"]["forms"] = "1" form_report["routes"][0]["write_plan"]["allowed"] = False form_report["routes"][0]["write_plan"]["apply_method"] = "metadata.write" form_report["routes"][0]["write_plan"]["target_kind"] = "module" form_report["routes"][0]["post_rollback_old"] = "after" form_report["routes"][0].pop("write_path", None) write_json(form_path, form_report) module_path = broken_saved_state_dir / "module-stream-write-smoke-script.json" module_report = json.loads(module_path.read_text(encoding="utf-8")) module_report["write_plan"]["allowed"] = False module_report["write_plan"]["apply_method"] = "metadata.write" module_report["write_plan"]["target_kind"] = "form" module_report["metadata_write"]["status"] = "ok" module_report["metadata_write"]["rolled_back"] = False write_json(module_path, module_report) saved_state_check = validate_base( "broken_saved_state", broken_saved_state_dir, validator_args(require_saved_state_write_smoke=True), ) saved_state_codes = {str(failure.get("code")) for failure in saved_state_check["failures"]} for expected in ( "saved_state_form_preflight_count_invalid", "saved_state_form_route_write_plan_not_allowed", "saved_state_form_route_apply_method_unexpected", "saved_state_form_route_field_missing", "saved_state_form_route_target_kind_unexpected", "saved_state_form_route_rollback_value_unexpected", "saved_state_module_write_plan_not_allowed", "saved_state_module_apply_method_unexpected", "saved_state_module_target_kind_unexpected", "saved_state_module_metadata_write_status_unexpected", "saved_state_module_rollback_missing", ): if expected not in saved_state_codes: failures.append({"code": "self_test_saved_state_expected_failure_missing", "expected": expected, "actual": sorted(saved_state_codes)}) broken_copy_plan_dir = reports_root / "broken_copy_plan" write_self_test_reports(broken_copy_plan_dir, base_id="broken_copy_plan", composed=False, saved_state_written=False) copy_plan_path = broken_copy_plan_dir / "saved-state-copy-plan.json" copy_plan_report = json.loads(copy_plan_path.read_text(encoding="utf-8")) copy_plan_report["status"] = "blocked_target_collision" copy_plan_report["ready_to_copy"] = False copy_plan_report["object"].pop("guid", None) copy_plan_report["source_rows"][0]["table"] = "ConfigSave" copy_plan_report["source_family"]["expected_source_table"] = "Config" copy_plan_report["source_family"]["source_tables"] = ["Config"] copy_plan_report["source_family"]["mismatched_source_tables"] = ["Config"] copy_plan_report["source_family"]["valid"] = False copy_plan_report["source_rows"][1].pop("file_name", None) copy_plan_report["source_row_details"] = {} copy_plan_report["target_collisions"] = { "status": "collision", "table": "ConfigCASSave", "rows": [{"FileName": "self-test-object", "PartNo": 0}], } copy_plan_report["summary"]["planned_source_rows"] = 99 copy_plan_report["summary"]["found_source_storage_rows"] = 0 copy_plan_report["summary"]["target_collision_rows"] = 1 write_json(copy_plan_path, copy_plan_report) copy_plan_check = validate_base("broken_copy_plan", broken_copy_plan_dir, validator_args()) copy_plan_codes = {str(failure.get("code")) for failure in copy_plan_check["failures"]} for expected in ( "saved_state_copy_plan_status_unexpected", "saved_state_copy_plan_not_ready", "saved_state_copy_plan_object_identity_missing", "saved_state_copy_plan_source_family_expected_unexpected", "saved_state_copy_plan_source_family_invalid", "saved_state_copy_plan_source_family_mismatch", "saved_state_copy_plan_source_row_table_unexpected", "saved_state_copy_plan_source_row_file_name_missing", "saved_state_copy_plan_summary_mismatch", "saved_state_copy_plan_found_rows_invalid", "saved_state_copy_plan_source_details_missing", "saved_state_copy_plan_target_collision_status_unexpected", "saved_state_copy_plan_target_collisions_present", "saved_state_copy_plan_target_collision_count_unexpected", ): if expected not in copy_plan_codes: failures.append({"code": "self_test_copy_plan_expected_failure_missing", "expected": expected, "actual": sorted(copy_plan_codes)}) broken_table_dir = reports_root / "broken_saved_state_table" write_self_test_reports(broken_table_dir, base_id="broken_saved_state_table", composed=False, saved_state_written=False) table_form_path = broken_table_dir / "saved-state-write-routes-smoke.json" table_form_report = json.loads(table_form_path.read_text(encoding="utf-8")) table_form_report["table"] = "ConfigSave" write_json(table_form_path, table_form_report) table_module_path = broken_table_dir / "module-stream-write-smoke-script.json" table_module_report = json.loads(table_module_path.read_text(encoding="utf-8")) table_module_report["module_ref"] = "ConfigSave:self-test#stream:0" write_json(table_module_path, table_module_report) table_check = validate_base("broken_saved_state_table", broken_table_dir, validator_args()) table_codes = {str(failure.get("code")) for failure in table_check["failures"]} if "saved_state_table_mismatch" not in table_codes: failures.append({"code": "self_test_saved_state_table_expected_failure_missing", "expected": "saved_state_table_mismatch", "actual": sorted(table_codes)}) broken_readiness_dir = reports_root / "broken_strict_readiness" write_self_test_reports(broken_readiness_dir, base_id="broken_strict_readiness", composed=False, saved_state_written=False) readiness_path = broken_readiness_dir / "saved-state-strict-readiness.json" readiness_report = json.loads(readiness_path.read_text(encoding="utf-8")) readiness_report["status"] = "other" readiness_report["ready"] = "false" readiness_report["saved_state_table"] = "ConfigSave" readiness_report["tables"] = ["ConfigSave"] readiness_report["summary"]["forms"] = "0" write_json(readiness_path, readiness_report) readiness_check = validate_base( "broken_strict_readiness", broken_readiness_dir, validator_args(saved_state_table="ConfigCASSave", require_saved_state_write_smoke=True), ) readiness_codes = {str(failure.get("code")) for failure in readiness_check["failures"]} for expected in ( "saved_state_strict_readiness_status_unexpected", "saved_state_strict_readiness_ready_not_boolean", "saved_state_strict_readiness_table_unexpected", "saved_state_strict_readiness_tables_unexpected", "saved_state_strict_readiness_summary_invalid", "saved_state_strict_readiness_required", ): if expected not in readiness_codes: failures.append({"code": "self_test_strict_readiness_expected_failure_missing", "expected": expected, "actual": sorted(readiness_codes)}) unexpected_table_dir = reports_root / "unexpected_saved_state_table" write_self_test_reports(unexpected_table_dir, base_id="unexpected_saved_state_table", composed=False, saved_state_written=False) unexpected_table_check = validate_base("unexpected_saved_state_table", unexpected_table_dir, validator_args(saved_state_table="ConfigSave")) unexpected_table_codes = {str(failure.get("code")) for failure in unexpected_table_check["failures"]} if "saved_state_table_unexpected" not in unexpected_table_codes: failures.append({ "code": "self_test_saved_state_table_expected_failure_missing", "expected": "saved_state_table_unexpected", "actual": sorted(unexpected_table_codes), }) table_codes = table_codes | unexpected_table_codes write_self_test_reports(report_dir, base_id=base_id, composed=True, saved_state_written=True) strict_pass = validate_base( base_id, report_dir, validator_args( require_saved_state_write_smoke=True, require_selector_chain_write_plan_composition=True, ), ) if strict_pass["failures"]: failures.append({"code": "self_test_strict_pass_unexpected_failures", "failures": strict_pass["failures"]}) return { "schema": "onec_verify_reports_self_test.v1", "passed": not failures, "soft": soft, "public_skip": public_skip, "public_skip_strict_failure_codes": sorted(public_skip_strict_codes), "strict_skip_failure_codes": sorted(strict_codes), "coverage_failure_codes": sorted(coverage_codes), "consistency_failure_codes": sorted(consistency_codes), "safety_failure_codes": sorted(safety_codes), "rollback_safety_failure_codes": sorted(rollback_safety_codes), "saved_state_diff_failure_codes": sorted(saved_state_diff_codes), "saved_state_changes_failure_codes": sorted(saved_state_changes_codes), "schema_failure_codes": sorted(schema_codes), "identity_failure_codes": sorted(identity_codes), "endpoint_failure_codes": sorted(endpoint_codes), "staleness_failure_codes": sorted(stale_codes), "duplicate_failure_codes": duplicate_failure_codes, "saved_state_failure_codes": sorted(saved_state_codes), "saved_state_strict_readiness_failure_codes": sorted(readiness_codes), "saved_state_copy_plan_failure_codes": sorted(copy_plan_codes), "saved_state_table_failure_codes": sorted(table_codes), "strict_pass": strict_pass, "failures": failures, } def main() -> int: parser = argparse.ArgumentParser(description="Offline validation for persisted 1C adapter verify report files.") parser.add_argument("--base-id", nargs="+", help="Base id(s) whose reports should be checked.") parser.add_argument("--reports-root", type=Path, default=DEFAULT_REPORTS_ROOT, help="Root directory containing per-base report folders.") parser.add_argument("--rest-adapter-url", help="Expected REST adapter endpoint_url in persisted REST reports.") parser.add_argument("--mcp-url", help="Expected MCP proxy endpoint_url in persisted MCP reports.") parser.add_argument("--max-report-age-seconds", type=int, help="Fail if any checked report file is older than this many seconds.") parser.add_argument("--skip-rest", action="store_true") parser.add_argument("--skip-mcp", action="store_true") parser.add_argument("--skip-write-plan-safety-smoke", action="store_true") parser.add_argument("--skip-write-rollback-safety-smoke", action="store_true") parser.add_argument("--skip-saved-state-diff-smoke", action="store_true") parser.add_argument("--skip-saved-state-write-smoke", action="store_true") parser.add_argument("--skip-code-write-saved-state-smoke", action="store_true") parser.add_argument("--skip-saved-state-copy-plan", action="store_true") parser.add_argument("--saved-state-table", choices=sorted(SAVED_STATE_TABLES), help="Expected saved-state target table in persisted copy-plan/form/module reports.") parser.add_argument("--require-saved-state-write-smoke", action="store_true") parser.add_argument("--require-code-write-saved-state-smoke", action="store_true") parser.add_argument("--require-selector-chain-write-plan-composition", action="store_true") parser.add_argument("--self-test", action="store_true", help="Run synthetic offline validator self-test.") parser.add_argument("--json", action="store_true", help="Print the full JSON report.") args = parser.parse_args() if args.self_test: report = run_self_test() if args.json or not report["passed"]: print(json.dumps(report, ensure_ascii=False, indent=2), file=sys.stderr if not report["passed"] else sys.stdout) else: print("OK: 1C verify report validator self-test passed.") return 0 if report["passed"] else 1 if not args.base_id: parser.error("--base-id is required unless --self-test is used") duplicates = duplicate_values(args.base_id) if duplicates: report = { "schema": "onec_verify_reports_check.v1", "passed": False, "reports_root": str(args.reports_root if args.reports_root.is_absolute() else ROOT / args.reports_root), "bases": {}, "failures": [{"code": "duplicate_base_id", "base_ids": duplicates}], } print(json.dumps(report, ensure_ascii=False, indent=2), file=sys.stderr) return 1 reports_root = args.reports_root if args.reports_root.is_absolute() else ROOT / args.reports_root bases = { base_id: validate_base(base_id, reports_root / safe_path_segment(base_id), args) for base_id in args.base_id } failures = [failure for result in bases.values() for failure in result["failures"]] report = { "schema": "onec_verify_reports_check.v1", "passed": not failures, "reports_root": str(reports_root), "bases": bases, "failures": failures, } if args.json or failures: print(json.dumps(report, ensure_ascii=False, indent=2), file=sys.stderr if failures else sys.stdout) else: print(f"OK: verified persisted 1C adapter reports for {len(bases)} base(s).") return 0 if report["passed"] else 1 if __name__ == "__main__": raise SystemExit(main())