diff --git a/devtools/schema_commit.py b/devtools/schema_commit.py index 83f58958d6..45f285491f 100644 --- a/devtools/schema_commit.py +++ b/devtools/schema_commit.py @@ -18,10 +18,6 @@ from polylogue.cli.shared.schema_command_support import build_schema_privacy_config from polylogue.config import get_config -from polylogue.maintenance.schema_inference_gate import ( - authorize_schema_generation, - resolve_schema_inference_archive_root, -) from polylogue.schemas.operator.commit import commit_provider_schema from polylogue.schemas.operator.models import SchemaCommitRequest @@ -60,6 +56,12 @@ def _build_parser() -> argparse.ArgumentParser: help="Privacy preset level. Defaults to standard.", ) parser.add_argument("--privacy-config", type=Path, default=None, help="Path to TOML privacy config overrides.") + parser.add_argument( + "--schema-inference-gate-receipt", + type=Path, + required=True, + help="Accepted PASS receipt from devtools verify schema-inference-gate.", + ) parser.add_argument( "--dry-run", "--check", @@ -68,11 +70,6 @@ def _build_parser() -> argparse.ArgumentParser: help="Preview what a commit would change without writing to --output-dir.", ) parser.add_argument("--json", action="store_true", help="Output as JSON.") - parser.add_argument( - "--schema-inference-receipt", - type=Path, - help="Fresh authoritative PASS receipt from devtools verify schema-inference-gate.", - ) return parser @@ -91,24 +88,26 @@ def main(argv: list[str] | None = None) -> int: output_dir = args.output_dir if args.output_dir is not None else DEFAULT_OUTPUT_DIR config = get_config() - if not args.dry_run and args.schema_inference_receipt is None: - print("schema-commit: --schema-inference-receipt is required when persisting schema packages", file=sys.stderr) + try: + result = commit_provider_schema( + SchemaCommitRequest( + provider=str(args.provider), + output_dir=output_dir, + archive_root=config.archive_root, + db_path=config.db_path, + max_samples=args.max_samples, + privacy_config=privacy_config, + full_corpus=bool(args.full_corpus), + dry_run=bool(args.dry_run), + schema_inference_gate_receipt_path=args.schema_inference_gate_receipt, + ) + ) + except ValueError as exc: + if args.json: + print(json.dumps({"provider": str(args.provider), "success": False, "error": str(exc)}, sort_keys=True)) + else: + print(f"schema-commit: {exc}", file=sys.stderr) return 1 - request = SchemaCommitRequest( - provider=str(args.provider), - output_dir=output_dir, - db_path=config.db_path, - max_samples=args.max_samples, - privacy_config=privacy_config, - full_corpus=bool(args.full_corpus), - dry_run=bool(args.dry_run), - ) - if args.dry_run: - result = commit_provider_schema(request) - else: - archive_root = resolve_schema_inference_archive_root(config, fallback_db_path=config.db_path) - with authorize_schema_generation(archive_root, args.schema_inference_receipt): - result = commit_provider_schema(request) if not result.success: error = result.generation.error or "Schema generation failed" @@ -124,6 +123,10 @@ def main(argv: list[str] | None = None) -> int: mode = "DRY RUN (no files written)" if result.dry_run else f"committed to {output_dir}" print(f"schema-commit: {result.provider} -- {mode}") print(f" sample_count={result.generation.sample_count}") + if result.handoff is not None: + print(f" handoff_digest={result.handoff.receipt_digest}") + if result.handoff_path is not None: + print(f" handoff_path={result.handoff_path}") for version_report in result.versions: flags = [] if version_report.narrowed_paths: diff --git a/polylogue/maintenance/schema_inference_gate.py b/polylogue/maintenance/schema_inference_gate.py index b4cf56748f..668666ea41 100644 --- a/polylogue/maintenance/schema_inference_gate.py +++ b/polylogue/maintenance/schema_inference_gate.py @@ -15,6 +15,7 @@ import platform import sqlite3 import sys +import uuid from collections import Counter from collections.abc import Iterable, Iterator, Mapping, Sequence from contextlib import contextmanager @@ -47,6 +48,7 @@ SCHEMA_INFERENCE_RECEIPT_ENV = "POLYLOGUE_SCHEMA_INFERENCE_RECEIPT" RECEIPT_TTL = timedelta(hours=24) RECEIPT_CLOCK_SKEW = timedelta(minutes=5) +RECEIPT_MAX_AGE_SECONDS = int(RECEIPT_TTL.total_seconds()) _ALLOWED_RESIDUAL_EXPLANATIONS = frozenset( {"materialized", "superseded-duplicate", "legitimately-excluded-non-conversation"} @@ -1365,6 +1367,229 @@ def _as_dict(value: object) -> dict[str, object]: return cast(dict[str, object], value) if isinstance(value, dict) else {} +def _hard_gate_evidence_payload( + query_results: Mapping[str, object], full_blob_hash_verification: Mapping[str, object] +) -> dict[str, object]: + """Return the exact live evidence that a commit is allowed to consume.""" + + return { + "query_results": dict(query_results), + "full_blob_hash_verification": dict(full_blob_hash_verification), + } + + +def schema_inference_hard_gate_evidence_digest( + query_results: Mapping[str, object], full_blob_hash_verification: Mapping[str, object] +) -> str: + """Digest the complete source-query and blob-verifier evidence payload.""" + + evidence = json.dumps( + _hard_gate_evidence_payload(query_results, full_blob_hash_verification), + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + allow_nan=False, + ) + return hashlib.sha256(evidence.encode("utf-8")).hexdigest() + + +def _validate_schema_inference_gate_payload( + payload: Mapping[str, object], + *, + archive_root: Path, + now: datetime | None = None, +) -> str: + """Validate the authoritative PASS receipt consumed by schema commit. + + A verdict alone is not evidence. The handoff contract requires the gate's + complete receipt shape, a fresh nonce, the current archive authority + identity, and an explicit successful full blob verification. + """ + + required_fields = { + "schema", + "gate_version", + "generated_at", + "receipt_nonce", + "verdict", + "archive_root", + "archive_identity_digest", + "schema_identity", + "source_schema_identity", + "query_results", + "source_denominators", + "blob_denominators", + "ground_truth_denominators", + "ground_truth_inputs", + "exemptions", + "corpus_fidelity", + "full_blob_hash_verification", + "input_paths", + "tool_versions", + "pass_fail_reasons", + "sample_limit", + "hard_gate_evidence_digest", + } + missing = sorted(required_fields - set(payload)) + if missing: + raise ValueError(f"schema-inference gate receipt is missing authoritative fields: {missing}") + if payload.get("schema") != RECEIPT_SCHEMA or payload.get("gate_version") != GATE_VERSION: + raise ValueError("schema-inference gate receipt schema or gate version is not authoritative") + if payload.get("verdict") != "PASS": + raise ValueError("schema-inference gate receipt must have verdict PASS") + + nonce = payload.get("receipt_nonce") + try: + valid_nonce = isinstance(nonce, str) and uuid.UUID(nonce).hex == nonce.replace("-", "") + except ValueError: + valid_nonce = False + if not valid_nonce: + raise ValueError("schema-inference gate receipt must contain a valid receipt nonce") + + generated_at = payload.get("generated_at") + if not isinstance(generated_at, str): + raise ValueError("schema-inference gate receipt generated_at is required") + try: + generated = datetime.fromisoformat(generated_at.replace("Z", "+00:00")) + except ValueError as exc: + raise ValueError("schema-inference gate receipt generated_at is invalid") from exc + if generated.tzinfo is None: + raise ValueError("schema-inference gate receipt generated_at must include a timezone") + observed_at = now or datetime.now(UTC) + age_seconds = (observed_at - generated.astimezone(UTC)).total_seconds() + if age_seconds < -300 or age_seconds > RECEIPT_MAX_AGE_SECONDS: + raise ValueError("schema-inference gate receipt is stale or from the future") + + location = ArchiveLocation.resolve(Path(archive_root).absolute()) + expected_root = location.configured_root + recorded_root = payload.get("archive_root") + if not isinstance(recorded_root, str) or Path(recorded_root).absolute() != expected_root: + raise ValueError("schema-inference gate receipt targets a different archive") + try: + expected_identity_digest = ArchiveIdentity.resolve_location(location).authority_identity_digest + except (OSError, ValueError, RuntimeError) as exc: + raise ValueError(f"unable to resolve target archive identity: {exc}") from exc + if payload.get("archive_identity_digest") != expected_identity_digest: + raise ValueError("schema-inference gate receipt archive identity is stale or mismatched") + + sample_limit = payload.get("sample_limit") + if not isinstance(sample_limit, int) or isinstance(sample_limit, bool) or sample_limit <= 0: + raise ValueError("schema-inference gate receipt sample_limit is invalid") + input_paths = payload.get("input_paths") + expected_input_paths = { + "archive_root": str(expected_root), + "source_db": str(expected_root / ARCHIVE_TIER_SPECS[ArchiveTier.SOURCE].filename), + "active_index_db": str(location.active_index_path), + "receipt": str(Path(str(input_paths.get("receipt"))).absolute()) + if isinstance(input_paths, Mapping) and isinstance(input_paths.get("receipt"), str) + else None, + } + if not isinstance(input_paths, Mapping) or any( + input_paths.get(key) != value for key, value in expected_input_paths.items() if key != "receipt" + ): + raise ValueError("schema-inference gate receipt input paths are not bound to the configured archive") + + try: + live_schema_identity = _tier_schema_identity(expected_root, location) + with open_readonly_connection(expected_root / ARCHIVE_TIER_SPECS[ArchiveTier.SOURCE].filename) as source: + referenced_hashes = _referenced_blob_hashes(source) + live_source_gates = _run_source_gates( + expected_root, index_path=location.active_index_path, sample_limit=sample_limit + ) + live_query_results = _as_dict(live_source_gates.get("gates")) + live_duplicate_gate = live_source_gates.get("duplicate_gate") + if isinstance(live_duplicate_gate, Mapping): + live_query_results["zero-unexplained-byte-duplicates"] = dict(live_duplicate_gate) + live_full_blob = _full_blob_hash_evidence(expected_root, referenced_hashes=referenced_hashes) + except (OSError, sqlite3.Error, ValueError) as exc: + raise ValueError(f"unable to recompute schema-inference gate evidence: {exc}") from exc + if payload.get("schema_identity") != live_schema_identity or payload.get("source_schema_identity") != _as_dict( + _as_dict(live_schema_identity.get("tiers")).get("source") + ): + raise ValueError("schema-inference gate receipt schema evidence is stale or mismatched") + if payload.get("query_results") != live_query_results: + raise ValueError("schema-inference gate receipt hard-gate query results changed") + if payload.get("source_denominators") != live_source_gates.get("source_counts", {}): + raise ValueError("schema-inference gate receipt source denominators changed") + if payload.get("blob_denominators") != live_source_gates.get("blob_denominators", {}): + raise ValueError("schema-inference gate receipt blob denominators changed") + recorded_ground_truth = payload.get("ground_truth_inputs") + recorded_origins = recorded_ground_truth.get("origins") if isinstance(recorded_ground_truth, Mapping) else None + live_ground_truth_roots: dict[str, tuple[Path, ...]] = {} + if isinstance(recorded_origins, Mapping): + for origin, raw_evidence in recorded_origins.items(): + if not isinstance(origin, str) or not isinstance(raw_evidence, Mapping): + continue + raw_roots = raw_evidence.get("declared_roots") + if isinstance(raw_roots, list) and all(isinstance(path, str) for path in raw_roots): + live_ground_truth_roots[origin] = tuple(Path(path) for path in raw_roots) + live_ground_truth = _ground_truth_evidence( + expected_root, + index_path=location.active_index_path, + source_counts=cast(Mapping[str, Mapping[str, int]], live_source_gates.get("source_counts", {})), + roots=live_ground_truth_roots, + ) + if payload.get("ground_truth_inputs") != live_ground_truth: + raise ValueError("schema-inference gate receipt ground-truth evidence changed") + try: + live_fidelity = _fidelity_evidence( + verify_archive(expected_root, checks=CORPUS_FIDELITY_CHECKS, sample_limit=sample_limit) + ) + except Exception as exc: + raise ValueError(f"unable to recompute schema-inference corpus fidelity: {exc}") from exc + fidelity_keys = ("passed", "reasons", "typed_residuals", "denominators") + recorded_fidelity = payload.get("corpus_fidelity") + if not isinstance(recorded_fidelity, Mapping) or any( + recorded_fidelity.get(key) != live_fidelity.get(key) for key in fidelity_keys + ): + raise ValueError("schema-inference gate receipt corpus-fidelity evidence changed") + if payload.get("ground_truth_denominators") != live_fidelity.get("denominators"): + raise ValueError("schema-inference gate receipt ground-truth denominators changed") + + full_blob = payload.get("full_blob_hash_verification") + if not isinstance(full_blob, Mapping) or full_blob.get("passed") is not True: + raise ValueError("schema-inference gate receipt lacks an explicit full_blob_hash_verification PASS") + verifier = full_blob.get("verifier") + if ( + not isinstance(verifier, Mapping) + or verifier.get("identity") != "polylogue.storage.blob_store.BlobStore.verify_all" + ): + raise ValueError("schema-inference gate receipt lacks the authoritative full blob verifier") + before = full_blob.get("before_snapshot") + after = full_blob.get("after_snapshot") + if ( + not isinstance(before, Mapping) + or not isinstance(after, Mapping) + or not isinstance(before.get("digest"), str) + or before.get("digest") != after.get("digest") + or full_blob.get("failures") != [] + or full_blob.get("missing_references") != [] + or full_blob.get("errors") != [] + ): + raise ValueError("schema-inference gate receipt full blob verification evidence is incomplete") + if payload.get("hard_gate_evidence_digest") != schema_inference_hard_gate_evidence_digest( + live_query_results, live_full_blob + ): + raise ValueError("schema-inference gate receipt hard-gate evidence digest does not match live evidence") + + query_results = payload.get("query_results") + if ( + not isinstance(query_results, Mapping) + or not query_results + or any(not isinstance(result, Mapping) or result.get("passed") is not True for result in query_results.values()) + ): + raise ValueError("schema-inference gate receipt contains a non-passing pristine query gate") + corpus_fidelity = payload.get("corpus_fidelity") + ground_truth = payload.get("ground_truth_inputs") + if not isinstance(corpus_fidelity, Mapping) or corpus_fidelity.get("passed") is not True: + raise ValueError("schema-inference gate receipt corpus fidelity is not PASS") + if not isinstance(ground_truth, Mapping) or ground_truth.get("passed") is not True: + raise ValueError("schema-inference gate receipt ground truth is not PASS") + if payload.get("pass_fail_reasons") != []: + raise ValueError("schema-inference gate receipt contains pass/fail reasons") + return schema_inference_gate_receipt_digest(payload) + + def _int_or_zero(value: object) -> int: return value if isinstance(value, int) and not isinstance(value, bool) else 0 @@ -1609,7 +1834,7 @@ def _parse_receipt_time(value: object) -> datetime: return parsed.astimezone(UTC) -def validate_schema_inference_gate_receipt( +def _validate_schema_inference_gate_path( receipt_path: Path, *, archive_root: Path, @@ -1665,12 +1890,33 @@ def validate_schema_inference_gate_receipt( return payload +def validate_schema_inference_gate_receipt( + payload_or_path: Mapping[str, object] | Path, + *, + archive_root: Path, + now: datetime | None = None, +) -> str | dict[str, object]: + """Validate either an in-memory gate payload or a receipt path. + + Schema commit and inferred-corpus callers already hold the parsed payload, + while the CLI schema-generation route owns a receipt path. Both routes use + the same authoritative validation contract. + """ + + if isinstance(payload_or_path, Path): + return _validate_schema_inference_gate_path(payload_or_path, archive_root=archive_root, now=now) + return _validate_schema_inference_gate_payload(payload_or_path, archive_root=archive_root, now=now) + + @contextmanager def authorize_schema_generation(archive_root: Path, receipt_path: Path) -> Iterator[dict[str, object]]: """Hold quiescence for one fresh schema operation or compatible short sequence.""" with schema_inference_quiescence(archive_root): - yield validate_schema_inference_gate_receipt(receipt_path, archive_root=archive_root) + yield cast( + dict[str, object], + validate_schema_inference_gate_receipt(receipt_path, archive_root=archive_root), + ) def _run_schema_inference_gate_locked( @@ -1781,6 +2027,7 @@ def _run_schema_inference_gate_locked( "gate_version": GATE_VERSION, "generated_at": datetime.now(UTC).isoformat(), "receipt_nonce": uuid4().hex, + "sample_limit": sample_limit, "verdict": "PASS" if not reasons and passed_hard_gates and schema_identity_ok else "FAIL", "archive_root": str(root), "archive_identity": archive_receipt_identity, @@ -1817,6 +2064,10 @@ def _run_schema_inference_gate_locked( }, "pass_fail_reasons": reasons, } + payload["hard_gate_evidence_digest"] = schema_inference_hard_gate_evidence_digest( + cast(Mapping[str, object], payload["query_results"]), + cast(Mapping[str, object], full_blob_hash_verification), + ) _write_json( safe_receipt_path, payload, @@ -1847,12 +2098,15 @@ def run_schema_inference_gate( "DEFAULT_SAMPLE_LIMIT", "GROUND_TRUTH_INPUTS", "RECEIPT_FILENAME", + "RECEIPT_MAX_AGE_SECONDS", "RECEIPT_SCHEMA", "RECEIPT_CLOCK_SKEW", "RECEIPT_TTL", "SCHEMA_INFERENCE_RECEIPT_ENV", "SchemaInferenceGateError", "SchemaInferenceGateResult", + "schema_inference_gate_receipt_digest", + "schema_inference_hard_gate_evidence_digest", "rebuild_source_revision_snapshot", "resolve_schema_inference_receipt_reference", "authorize_schema_generation", diff --git a/polylogue/schemas/operator/commit.py b/polylogue/schemas/operator/commit.py index 7570b15744..129a17d7d3 100644 --- a/polylogue/schemas/operator/commit.py +++ b/polylogue/schemas/operator/commit.py @@ -30,18 +30,33 @@ from __future__ import annotations +import json import shutil import tempfile +from collections.abc import Mapping from pathlib import Path +from typing import cast from polylogue.core.json import JSONDocument +from polylogue.maintenance.schema_inference_gate import ( + validate_schema_inference_gate_receipt, +) +from polylogue.paths import archive_root as default_archive_root from polylogue.schemas.generation.models import GenerationResult from polylogue.schemas.generation.workflow import generate_all_schemas from polylogue.schemas.operator.inference import privacy_config_from_payload from polylogue.schemas.operator.models import SchemaCommitRequest, SchemaCommitResult, SchemaVersionCommitReport +from polylogue.schemas.operator.receipt import ( + SCHEMA_INFERENCE_HANDOFF_FILENAME, + SchemaInferenceReceipt, + build_schema_inference_receipt, + load_schema_inference_receipt, + write_schema_inference_receipt, +) from polylogue.schemas.registry import SchemaRegistry from polylogue.schemas.runtime_registry import canonical_schema_provider from polylogue.schemas.type_narrowing import added_paths, narrowed_paths +from polylogue.storage.archive_identity import ArchiveLocation def _element_schemas_by_kind( @@ -52,11 +67,56 @@ def _element_schemas_by_kind( } +def _accepted_gate_receipt_digest(path: Path | None, *, archive_root: Path) -> str: + if path is None: + raise ValueError("schema commit requires an accepted schema-inference gate receipt path") + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError, ValueError) as exc: + raise ValueError(f"unable to read schema-inference gate receipt {path}: {exc}") from exc + if not isinstance(payload, Mapping): + raise ValueError("schema-inference gate receipt must be a JSON object") + return cast( + str, + validate_schema_inference_gate_receipt( + cast(Mapping[str, object], payload), + archive_root=archive_root, + ), + ) + + +def _target_archive_location(request: SchemaCommitRequest) -> ArchiveLocation: + configured_root = request.archive_root or default_archive_root() + location = ArchiveLocation.resolve(configured_root) + expected_db_path = location.active_index_path.resolve(strict=False) + if request.db_path is not None and request.db_path.resolve(strict=False) != expected_db_path: + raise ValueError( + "schema commit db_path must identify the active index of the configured archive; " + f"expected={expected_db_path}, actual={request.db_path.resolve(strict=False)}" + ) + return location + + def _commit_into(request: SchemaCommitRequest, output_dir: Path) -> SchemaCommitResult: provider_token = str(canonical_schema_provider(request.provider)) + output_dir = output_dir.absolute() + handoff_path = output_dir / SCHEMA_INFERENCE_HANDOFF_FILENAME + existing_handoff = load_schema_inference_receipt(handoff_path) if handoff_path.exists() else None + archive_location = _target_archive_location(request) + gate_receipt_digest = _accepted_gate_receipt_digest( + request.schema_inference_gate_receipt_path, + archive_root=archive_location.configured_root, + ) + if existing_handoff is not None and existing_handoff.gate_receipt_digest != gate_receipt_digest: + raise ValueError( + "existing schema inference handoff was produced from a different gate receipt; " + "regenerate the handoff from the accepted gate before committing" + ) registry_before = SchemaRegistry(storage_root=output_dir) - catalog_before = registry_before.load_package_catalog(provider_token) + # The bundled registry is a read fallback, not the prior state of this + # commit's output directory. Compare against local persisted packages only. + catalog_before = registry_before._load_local_catalog(provider_token) before_versions = {package.version for package in catalog_before.packages} if catalog_before is not None else set() before_schemas: dict[str, dict[str, JSONDocument | None]] = {} if catalog_before is not None: @@ -68,7 +128,7 @@ def _commit_into(request: SchemaCommitRequest, output_dir: Path) -> SchemaCommit generation_results = generate_all_schemas( output_dir, - db_path=request.db_path, + db_path=archive_location.active_index_path, providers=[request.provider], max_samples=request.max_samples, privacy_config=privacy_config_from_payload(request.privacy_config), @@ -83,6 +143,8 @@ def _commit_into(request: SchemaCommitRequest, output_dir: Path) -> SchemaCommit ) version_reports: list[SchemaVersionCommitReport] = [] + handoff: SchemaInferenceReceipt | None = None + registry_after: SchemaRegistry | None = None if generation.success: registry_after = SchemaRegistry(storage_root=output_dir) catalog_after = registry_after.load_package_catalog(provider_token) @@ -124,11 +186,24 @@ def _commit_into(request: SchemaCommitRequest, output_dir: Path) -> SchemaCommit ) ) + if generation.success: + if registry_after is None: + raise AssertionError("successful schema generation did not produce a persisted registry") + provider_handoff = build_schema_inference_receipt( + registry_after, + provider=provider_token, + gate_receipt_digest=gate_receipt_digest, + ) + handoff = existing_handoff.merged_with(provider_handoff) if existing_handoff is not None else provider_handoff + write_schema_inference_receipt(handoff, handoff_path) + return SchemaCommitResult( provider=request.provider, generation=generation, versions=tuple(version_reports), dry_run=request.dry_run, + handoff=handoff, + handoff_path=handoff_path if generation.success else None, ) @@ -151,12 +226,17 @@ def commit_provider_schema(request: SchemaCommitRequest) -> SchemaCommitResult: committed_provider_dir = request.output_dir / provider_token if committed_provider_dir.exists(): shutil.copytree(committed_provider_dir, staging_root / provider_token) + handoff_path = request.output_dir / SCHEMA_INFERENCE_HANDOFF_FILENAME + if handoff_path.exists(): + shutil.copy2(handoff_path, staging_root / SCHEMA_INFERENCE_HANDOFF_FILENAME) result = _commit_into(request, staging_root) return SchemaCommitResult( provider=result.provider, generation=result.generation, versions=result.versions, dry_run=True, + handoff=result.handoff, + handoff_path=None, ) diff --git a/polylogue/schemas/operator/models.py b/polylogue/schemas/operator/models.py index 36a31c3ea5..81b89d524d 100644 --- a/polylogue/schemas/operator/models.py +++ b/polylogue/schemas/operator/models.py @@ -9,6 +9,7 @@ from polylogue.scenarios import CorpusScenario, CorpusSpec from polylogue.schemas.generation.models import GenerationProgressCallback, GenerationResult +from polylogue.schemas.operator.receipt import SchemaInferenceReceipt from polylogue.schemas.packages import SchemaPackageCatalog, SchemaResolution, SchemaVersionPackage from polylogue.schemas.tooling_registry import ClusterManifest, SchemaDiff from polylogue.schemas.validation.models import ArtifactCoverageReport @@ -364,6 +365,8 @@ class SchemaCommitRequest: privacy_config: JSONDocument | None = None full_corpus: bool = True dry_run: bool = False + schema_inference_gate_receipt_path: Path | None = None + archive_root: Path | None = None @dataclass(frozen=True) @@ -394,6 +397,8 @@ class SchemaCommitResult: generation: GenerationResult versions: tuple[SchemaVersionCommitReport, ...] dry_run: bool + handoff: SchemaInferenceReceipt | None = None + handoff_path: Path | None = None @property def success(self) -> bool: @@ -412,4 +417,6 @@ def to_dict(self) -> JSONDocument: "dry_run": self.dry_run, "sample_count": self.generation.sample_count, "versions": [report.to_dict() for report in self.versions], + "handoff": self.handoff.to_payload() if self.handoff is not None else None, + "handoff_path": str(self.handoff_path) if self.handoff_path is not None else None, } diff --git a/polylogue/schemas/operator/receipt.py b/polylogue/schemas/operator/receipt.py new file mode 100644 index 0000000000..42664e7fd4 --- /dev/null +++ b/polylogue/schemas/operator/receipt.py @@ -0,0 +1,462 @@ +"""Content-addressed handoff receipts for committed inferred schemas.""" + +from __future__ import annotations + +import json +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import Literal, Protocol, TypeAlias, cast + +from polylogue.core.hashing import hash_file, hash_payload +from polylogue.core.json import JSONDocument +from polylogue.core.sources import origin_from_provider +from polylogue.schemas.operator.registry import RuntimeSchemaRegistryLike +from polylogue.schemas.packages import SchemaPackageCatalog, SchemaVersionPackage +from polylogue.schemas.runtime_registry import canonical_schema_provider +from polylogue.schemas.synthetic.classification import classify_schema_constructs +from polylogue.schemas.synthetic.wire_formats import PROVIDER_WIRE_FORMATS + +SCHEMA_INFERENCE_HANDOFF_SCHEMA = "polylogue.schema-inference-handoff.v1" +SCHEMA_INFERENCE_HANDOFF_FILENAME = "schema-inference-handoff.json" + +CoverageDecision: TypeAlias = Literal["committed", "unsupported", "nonrepresentable"] +_COVERAGE_DECISIONS = frozenset({"committed", "unsupported", "nonrepresentable"}) + + +def _require_digest(value: object, *, field: str) -> str: + if not isinstance(value, str) or len(value) != 64 or any(char not in "0123456789abcdef" for char in value): + raise ValueError(f"{field} must be a lowercase SHA-256 digest") + return value + + +@dataclass(frozen=True, order=True, slots=True) +class SchemaInferenceCoverageDecision: + """The explicit decision for one origin/provider pair in a handoff.""" + + origin: str + provider: str + decision: CoverageDecision + reason: str | None = None + + def __post_init__(self) -> None: + if not self.origin or not self.provider or self.decision not in _COVERAGE_DECISIONS: + raise ValueError("schema inference coverage decisions require identity and a valid decision") + if self.decision != "committed" and not self.reason: + raise ValueError("unsupported or nonrepresentable coverage decisions require a reason") + + def to_payload(self) -> JSONDocument: + return { + "origin": self.origin, + "provider": self.provider, + "decision": self.decision, + "reason": self.reason, + } + + +@dataclass(frozen=True, order=True, slots=True) +class SchemaInferenceUnsupportedDecision: + """A typed refusal for an element that cannot enter the inferred corpus.""" + + provider: str + package_version: str + element_kind: str + decision: Literal["unsupported", "nonrepresentable"] + reason: str + details: tuple[str, ...] = () + + def __post_init__(self) -> None: + if not self.provider or not self.package_version or not self.element_kind: + raise ValueError("unsupported schema decisions require package identity") + if self.decision not in {"unsupported", "nonrepresentable"} or not self.reason: + raise ValueError("unsupported schema decisions require a supported decision and reason") + if tuple(sorted(set(self.details))) != self.details: + raise ValueError("unsupported schema decision details must be sorted and unique") + + def to_payload(self) -> JSONDocument: + return { + "provider": self.provider, + "package_version": self.package_version, + "element_kind": self.element_kind, + "decision": self.decision, + "reason": self.reason, + "details": list(self.details), + } + + +@dataclass(frozen=True, order=True, slots=True) +class SchemaElementContentHash: + element_kind: str + content_hash: str + + def __post_init__(self) -> None: + if not self.element_kind: + raise ValueError("schema element hash requires an element kind") + _require_digest(self.content_hash, field="element content_hash") + + def to_payload(self) -> JSONDocument: + return {"element_kind": self.element_kind, "content_hash": self.content_hash} + + +@dataclass(frozen=True, order=True, slots=True) +class SchemaPackageContentHash: + """Exact persisted hashes for one package version and its elements.""" + + provider: str + package_version: str + package_hash: str + version_hash: str + element_hashes: tuple[SchemaElementContentHash, ...] + + def __post_init__(self) -> None: + if not self.provider or not self.package_version: + raise ValueError("schema package hashes require provider and version") + _require_digest(self.package_hash, field="package_hash") + _require_digest(self.version_hash, field="version_hash") + if tuple(sorted(self.element_hashes)) != self.element_hashes: + raise ValueError("schema element hashes must be sorted") + if len({item.element_kind for item in self.element_hashes}) != len(self.element_hashes): + raise ValueError("schema package element hashes must be unique") + + def to_payload(self) -> JSONDocument: + return { + "provider": self.provider, + "package_version": self.package_version, + "package_hash": self.package_hash, + "version_hash": self.version_hash, + "element_hashes": [item.to_payload() for item in self.element_hashes], + } + + +@dataclass(frozen=True, slots=True) +class SchemaInferenceReceipt: + """Immutable aggregate handoff from pristine gate to inferred corpus.""" + + gate_receipt_digest: str + coverage_decisions: tuple[SchemaInferenceCoverageDecision, ...] + packages: tuple[SchemaPackageContentHash, ...] + unsupported_decisions: tuple[SchemaInferenceUnsupportedDecision, ...] = () + + def __post_init__(self) -> None: + _require_digest(self.gate_receipt_digest, field="gate_receipt_digest") + if tuple(sorted(self.coverage_decisions)) != self.coverage_decisions: + raise ValueError("coverage decisions must be sorted") + if tuple(sorted(self.packages)) != self.packages: + raise ValueError("package hashes must be sorted") + if tuple(sorted(self.unsupported_decisions)) != self.unsupported_decisions: + raise ValueError("unsupported decisions must be sorted") + coverage_keys = [(item.origin, item.provider) for item in self.coverage_decisions] + if len(coverage_keys) != len(set(coverage_keys)): + raise ValueError("coverage decisions must have unique origin/provider pairs") + package_keys = [(item.provider, item.package_version) for item in self.packages] + if len(package_keys) != len(set(package_keys)): + raise ValueError("package hashes must have unique provider/version pairs") + unsupported_keys = [ + (item.provider, item.package_version, item.element_kind) for item in self.unsupported_decisions + ] + if len(unsupported_keys) != len(set(unsupported_keys)): + raise ValueError("unsupported decisions must have unique package elements") + + @property + def receipt_digest(self) -> str: + return hash_payload(self._payload_without_digest()) + + def _payload_without_digest(self) -> JSONDocument: + return { + "schema": SCHEMA_INFERENCE_HANDOFF_SCHEMA, + "gate_receipt_digest": self.gate_receipt_digest, + "coverage_decisions": [item.to_payload() for item in self.coverage_decisions], + "packages": [item.to_payload() for item in self.packages], + "unsupported_decisions": [item.to_payload() for item in self.unsupported_decisions], + } + + def to_payload(self) -> JSONDocument: + return {**self._payload_without_digest(), "receipt_digest": self.receipt_digest} + + def merged_with(self, other: SchemaInferenceReceipt) -> SchemaInferenceReceipt: + if self.gate_receipt_digest != other.gate_receipt_digest: + raise ValueError("schema inference handoffs use different gate receipt digests") + providers = {item.provider for item in other.coverage_decisions} + coverage = tuple( + sorted( + [item for item in self.coverage_decisions if item.provider not in providers] + + list(other.coverage_decisions) + ) + ) + package_providers = {item.provider for item in other.packages} + packages = tuple( + sorted([item for item in self.packages if item.provider not in package_providers] + list(other.packages)) + ) + unsupported = tuple( + sorted( + [item for item in self.unsupported_decisions if item.provider not in package_providers] + + list(other.unsupported_decisions) + ) + ) + return SchemaInferenceReceipt(self.gate_receipt_digest, coverage, packages, unsupported) + + @classmethod + def from_payload(cls, payload: Mapping[str, object]) -> SchemaInferenceReceipt: + expected = { + "schema", + "gate_receipt_digest", + "coverage_decisions", + "packages", + "unsupported_decisions", + "receipt_digest", + } + if set(payload) != expected or payload.get("schema") != SCHEMA_INFERENCE_HANDOFF_SCHEMA: + raise ValueError("schema inference handoff fields or schema changed") + receipt = cls( + gate_receipt_digest=_require_digest(payload.get("gate_receipt_digest"), field="gate_receipt_digest"), + coverage_decisions=tuple( + _coverage_from_payload(item) for item in _list_of_mappings(payload, "coverage_decisions") + ), + packages=tuple(_package_from_payload(item) for item in _list_of_mappings(payload, "packages")), + unsupported_decisions=tuple( + _unsupported_from_payload(item) for item in _list_of_mappings(payload, "unsupported_decisions") + ), + ) + if payload.get("receipt_digest") != receipt.receipt_digest: + raise ValueError("schema inference handoff receipt digest mismatch") + return receipt + + +def _list_of_mappings(payload: Mapping[str, object], field: str) -> list[Mapping[str, object]]: + value = payload.get(field) + if not isinstance(value, list) or not all(isinstance(item, Mapping) for item in value): + raise ValueError(f"schema inference handoff {field} must be a list of objects") + return [cast(Mapping[str, object], item) for item in value] + + +def _coverage_from_payload(payload: Mapping[str, object]) -> SchemaInferenceCoverageDecision: + if set(payload) != {"origin", "provider", "decision", "reason"}: + raise ValueError("coverage decision fields changed") + reason = payload.get("reason") + if not isinstance(reason, str): + reason = None + return SchemaInferenceCoverageDecision( + origin=str(payload.get("origin")), + provider=str(payload.get("provider")), + decision=cast(CoverageDecision, payload.get("decision")), + reason=reason, + ) + + +def _unsupported_from_payload(payload: Mapping[str, object]) -> SchemaInferenceUnsupportedDecision: + if set(payload) != {"provider", "package_version", "element_kind", "decision", "reason", "details"}: + raise ValueError("unsupported decision fields changed") + details = payload.get("details") + if not isinstance(details, list) or not all(isinstance(item, str) for item in details): + raise ValueError("unsupported decision details must be a list of strings") + return SchemaInferenceUnsupportedDecision( + provider=str(payload.get("provider")), + package_version=str(payload.get("package_version")), + element_kind=str(payload.get("element_kind")), + decision=cast(Literal["unsupported", "nonrepresentable"], payload.get("decision")), + reason=str(payload.get("reason")), + details=tuple(details), + ) + + +def _element_from_payload(payload: Mapping[str, object]) -> SchemaElementContentHash: + if set(payload) != {"element_kind", "content_hash"}: + raise ValueError("element hash fields changed") + return SchemaElementContentHash( + str(payload.get("element_kind")), _require_digest(payload.get("content_hash"), field="content_hash") + ) + + +def _package_from_payload(payload: Mapping[str, object]) -> SchemaPackageContentHash: + if set(payload) != {"provider", "package_version", "package_hash", "version_hash", "element_hashes"}: + raise ValueError("package hash fields changed") + return SchemaPackageContentHash( + provider=str(payload.get("provider")), + package_version=str(payload.get("package_version")), + package_hash=_require_digest(payload.get("package_hash"), field="package_hash"), + version_hash=_require_digest(payload.get("version_hash"), field="version_hash"), + element_hashes=tuple(_element_from_payload(item) for item in _list_of_mappings(payload, "element_hashes")), + ) + + +class SchemaReceiptRegistry(RuntimeSchemaRegistryLike, Protocol): + """Filesystem-backed registry operations required to hash persisted files.""" + + @property + def storage_root(self) -> Path: ... + + +def _package_hashes_for_package( + registry: SchemaReceiptRegistry, provider: str, package: SchemaVersionPackage +) -> SchemaPackageContentHash: + storage_root = registry.storage_root + provider_token = str(canonical_schema_provider(provider)) + package_dir = storage_root / provider_token / "versions" / package.version + package_path = package_dir / "package.json" + if not package_path.exists(): + raise ValueError(f"persisted schema package is missing: {package_path}") + element_hashes: list[SchemaElementContentHash] = [] + package_hash = hash_file(package_path) + version_files: list[dict[str, str]] = [{"path": "package.json", "hash": package_hash}] + for element in sorted(package.elements, key=lambda item: item.element_kind): + if element.schema_file is None: + continue + path = package_dir / "elements" / element.schema_file + if not path.exists(): + raise ValueError(f"persisted schema element is missing: {path}") + content_hash = hash_file(path) + element_hashes.append(SchemaElementContentHash(element.element_kind, content_hash)) + version_files.append({"path": f"elements/{element.schema_file}", "hash": content_hash}) + if package.workload_profile_file is not None: + path = package_dir / package.workload_profile_file + if not path.exists(): + raise ValueError(f"persisted schema workload profile is missing: {path}") + content_hash = hash_file(path) + version_files.append({"path": package.workload_profile_file, "hash": content_hash}) + return SchemaPackageContentHash( + provider=provider_token, + package_version=package.version, + package_hash=package_hash, + version_hash=hash_payload(version_files), + element_hashes=tuple(element_hashes), + ) + + +def package_hashes_for_registry( + registry: SchemaReceiptRegistry, providers: Sequence[str] | None = None +) -> tuple[SchemaPackageContentHash, ...]: + provider_names = tuple(providers) if providers is not None else tuple(registry.list_providers()) + records: list[SchemaPackageContentHash] = [] + for provider in sorted(set(provider_names)): + catalog = registry.load_package_catalog(provider) + if not isinstance(catalog, SchemaPackageCatalog): + raise ValueError(f"registry provider {provider!r} has no persisted package catalog") + records.extend(_package_hashes_for_package(registry, provider, package) for package in catalog.packages) + return tuple(sorted(records)) + + +def _unsupported_for_package( + registry: SchemaReceiptRegistry, provider: str, package: SchemaVersionPackage +) -> tuple[SchemaInferenceUnsupportedDecision, ...]: + decisions: list[SchemaInferenceUnsupportedDecision] = [] + for element in package.elements: + if not element.supported or element.schema_file is None: + decisions.append( + SchemaInferenceUnsupportedDecision( + provider, + package.version, + element.element_kind, + "unsupported", + "unsupported_element" if not element.supported else "missing_schema", + ) + ) + continue + if provider not in PROVIDER_WIRE_FORMATS: + decisions.append( + SchemaInferenceUnsupportedDecision( + provider, + package.version, + element.element_kind, + "unsupported", + "provider_without_wire_format", + ) + ) + continue + schema = registry.get_element_schema(provider, version=package.version, element_kind=element.element_kind) + if not isinstance(schema, Mapping): + decisions.append( + SchemaInferenceUnsupportedDecision( + provider, package.version, element.element_kind, "unsupported", "missing_schema" + ) + ) + continue + unsupported = tuple( + item.construct for item in classify_schema_constructs(schema) if item.state == "unsupported" + ) + if unsupported: + decisions.append( + SchemaInferenceUnsupportedDecision( + provider, + package.version, + element.element_kind, + "nonrepresentable", + "unsupported_json_schema_construct", + unsupported, + ) + ) + return tuple(sorted(decisions)) + + +def build_schema_inference_receipt( + registry: SchemaReceiptRegistry, *, provider: str, gate_receipt_digest: str +) -> SchemaInferenceReceipt: + provider_token = str(canonical_schema_provider(provider)) + catalog = registry.load_package_catalog(provider_token) + if not isinstance(catalog, SchemaPackageCatalog) or not catalog.packages: + raise ValueError(f"schema commit produced no persisted packages for {provider_token}") + packages = tuple( + sorted(_package_hashes_for_package(registry, provider_token, package) for package in catalog.packages) + ) + unsupported = tuple( + sorted( + item for package in catalog.packages for item in _unsupported_for_package(registry, provider_token, package) + ) + ) + unsupported_keys = {(item.package_version, item.element_kind) for item in unsupported} + representable = { + (package.version, element.element_kind) + for package in catalog.packages + for element in package.elements + if (package.version, element.element_kind) not in unsupported_keys + } + if representable: + coverage_decision: CoverageDecision = "committed" + coverage_reason = "persisted package/version/element hashes recorded" + else: + coverage_decision = ( + "nonrepresentable" + if unsupported and all(item.decision == "nonrepresentable" for item in unsupported) + else "unsupported" + ) + coverage_reason = "provider has no executable persisted schema element" + origin = origin_from_provider(provider_token).value + coverage = SchemaInferenceCoverageDecision( + origin=origin, + provider=provider_token, + decision=coverage_decision, + reason=coverage_reason, + ) + return SchemaInferenceReceipt(gate_receipt_digest, (coverage,), packages, unsupported) + + +def load_schema_inference_receipt(path: Path) -> SchemaInferenceReceipt: + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError, ValueError) as exc: + raise ValueError(f"unable to read schema inference handoff {path}: {exc}") from exc + if not isinstance(payload, Mapping): + raise ValueError("schema inference handoff root must be an object") + return SchemaInferenceReceipt.from_payload(cast(Mapping[str, object], payload)) + + +def write_schema_inference_receipt(receipt: SchemaInferenceReceipt, path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.tmp") + temporary.write_text(json.dumps(receipt.to_payload(), indent=2, sort_keys=True) + "\n", encoding="utf-8") + temporary.replace(path) + + +__all__ = [ + "SCHEMA_INFERENCE_HANDOFF_FILENAME", + "SCHEMA_INFERENCE_HANDOFF_SCHEMA", + "SchemaElementContentHash", + "SchemaInferenceCoverageDecision", + "SchemaInferenceReceipt", + "SchemaInferenceUnsupportedDecision", + "SchemaPackageContentHash", + "SchemaReceiptRegistry", + "build_schema_inference_receipt", + "load_schema_inference_receipt", + "package_hashes_for_registry", + "write_schema_inference_receipt", +] diff --git a/polylogue/schemas/synthetic/classification.py b/polylogue/schemas/synthetic/classification.py new file mode 100644 index 0000000000..e3c1b80ba0 --- /dev/null +++ b/polylogue/schemas/synthetic/classification.py @@ -0,0 +1,519 @@ +"""Canonical executable-support classification for persisted schemas.""" + +from __future__ import annotations + +import math +import re +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from typing import Literal + +from polylogue.schemas.synthetic.models import SchemaRecord + +ConstructSupportState = Literal["supported", "unsupported"] + + +@dataclass(frozen=True, order=True) +class ConstructSupport: + """Support verdict for one schema keyword or runtime annotation.""" + + construct: str + state: ConstructSupportState + + def to_payload(self) -> dict[str, str]: + return {"construct": self.construct, "state": self.state} + + +_SUPPORTED_SCHEMA_CONSTRUCTS = frozenset( + { + "$anchor", + "$comment", + "$id", + "$schema", + "additionalProperties", + "anyOf", + "default", + "deprecated", + "description", + "examples", + "items", + "oneOf", + "properties", + "readOnly", + "title", + "type", + "writeOnly", + } +) +_STANDARD_SCHEMA_KEYWORDS = frozenset( + { + "$anchor", + "$comment", + "$defs", + "$dynamicAnchor", + "$dynamicRef", + "$id", + "$recursiveAnchor", + "$recursiveRef", + "$ref", + "$schema", + "$vocabulary", + "additionalProperties", + "allOf", + "anyOf", + "const", + "contains", + "contentEncoding", + "contentMediaType", + "contentSchema", + "default", + "definitions", + "dependentRequired", + "dependentSchemas", + "dependencies", + "deprecated", + "description", + "else", + "enum", + "examples", + "exclusiveMaximum", + "exclusiveMinimum", + "format", + "formatAssertion", + "if", + "items", + "maxContains", + "maxItems", + "maxLength", + "maxProperties", + "maximum", + "minContains", + "minItems", + "minLength", + "minProperties", + "minimum", + "multipleOf", + "not", + "oneOf", + "pattern", + "patternProperties", + "prefixItems", + "properties", + "propertyNames", + "readOnly", + "required", + "then", + "title", + "unevaluatedItems", + "unevaluatedProperties", + "uniqueItems", + "writeOnly", + } +) +_SCHEMA_MAPPING_KEYWORDS = frozenset( + { + "$defs", + "additionalProperties", + "contentSchema", + "contains", + "dependentSchemas", + "dependencies", + "else", + "if", + "items", + "not", + "patternProperties", + "properties", + "propertyNames", + "then", + "unevaluatedItems", + "unevaluatedProperties", + } +) +_SCHEMA_ARRAY_KEYWORDS = frozenset({"allOf", "anyOf", "oneOf", "prefixItems"}) +_SUPPORTED_SYNTHETIC_ANNOTATIONS = frozenset( + { + "x-polylogue-array-lengths", + "x-polylogue-foreign-keys", + "x-polylogue-format", + "x-polylogue-frequency", + "x-polylogue-multiline", + "x-polylogue-mutually-exclusive", + "x-polylogue-observed-distribution", + "x-polylogue-range", + "x-polylogue-semantic-role", + "x-polylogue-string-lengths", + "x-polylogue-time-deltas", + "x-polylogue-values", + } +) +_SUPPORTED_FORMAT_VALUES = frozenset( + {"uuid4", "uuid", "hex-id", "iso8601", "unix-epoch", "unix-epoch-str", "url", "email", "mime-type", "base64"} +) +_SUPPORTED_SEMANTIC_ROLE_VALUES = frozenset({"message_role", "message_body", "message_timestamp", "session_title"}) +_PERSISTED_SCHEMA_METADATA_ANNOTATIONS = frozenset( + { + "x-polylogue-anchor-profile-family-id", + "x-polylogue-artifact-kind", + "x-polylogue-element-bundle-scope-count", + "x-polylogue-element-first-seen", + "x-polylogue-element-kind", + "x-polylogue-element-last-seen", + "x-polylogue-evidence", + "x-polylogue-evidence-confidence", + "x-polylogue-exact-structure-ids", + "x-polylogue-generated-at", + "x-polylogue-generator", + "x-polylogue-high-cardinality-keys", + "x-polylogue-observed-artifact-count", + "x-polylogue-package-profile-family-ids", + "x-polylogue-package-version", + "x-polylogue-profile-family-ids", + "x-polylogue-profile-tokens", + "x-polylogue-promoted-at", + "x-polylogue-registered-at", + "x-polylogue-sample-count", + "x-polylogue-sample-granularity", + "x-polylogue-score", + "x-polylogue-version", + } +) +_SCHEMA_PATH_SEGMENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_-]*$") + + +def _number(value: object) -> int | float | None: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + try: + return value if math.isfinite(float(value)) else None + except (OverflowError, ValueError): + return None + + +def _valid_pair(value: object, *, integral: bool = False, nonnegative: bool = False) -> bool: + if not isinstance(value, list) or len(value) != 2: + return False + first, second = (_number(item) for item in value) + if first is None or second is None: + return False + return not ( + (integral and (not isinstance(first, int) or not isinstance(second, int))) + or (nonnegative and (first < 0 or second < 0)) + or first > second + ) + + +def _relation_records(value: object, required: tuple[str, ...]) -> list[dict[str, object]] | None: + if not isinstance(value, list) or not value: + return None + records: list[dict[str, object]] = [] + for record in value: + if not isinstance(record, dict) or not all( + isinstance(record.get(field), str) and record[field].strip() for field in required + ): + return None + records.append(record) + return records + + +def _histogram_bucket_is_safe(index: int, log_base: int | float) -> bool: + try: + sampled_value = math.expm1((abs(float(index)) - 0.5) * math.log(float(log_base))) + except (OverflowError, ValueError): + return False + return math.isfinite(sampled_value) + + +def _valid_observed_distribution(value: object) -> bool: + if not isinstance(value, Mapping) or not value: + return False + for distribution in value.values(): + if not isinstance(distribution, Mapping): + return False + histogram = distribution.get("histogram") + log_base = _number(distribution.get("log_base")) + if not isinstance(histogram, list) or not histogram or log_base is None or log_base <= 1: + return False + if not all( + isinstance(bucket, list) + and len(bucket) == 2 + and isinstance(bucket[0], int) + and not isinstance(bucket[0], bool) + and isinstance(bucket[1], int) + and not isinstance(bucket[1], bool) + and bucket[1] > 0 + and _histogram_bucket_is_safe(bucket[0], float(log_base)) + for bucket in histogram + ): + return False + values = tuple(_number(distribution.get(key)) for key in ("p0", "p50", "p90", "p95", "p99", "p100")) + present = tuple(item for item in values if item is not None) + minimum = _number(distribution.get("min")) + maximum = _number(distribution.get("max")) + if minimum is not None and maximum is not None and minimum > maximum: + return False + if any(left > right for left, right in zip(present, present[1:], strict=False)): + return False + if any( + (minimum is not None and item < minimum) or (maximum is not None and item > maximum) for item in present + ): + return False + for key in ("min", "max", "mean", "p0", "p50", "p90", "p95", "p99", "p100", "stddev"): + if key in distribution and _number(distribution[key]) is None: + return False + stddev = _number(distribution.get("stddev")) + if stddev is not None and stddev < 0: + return False + return True + + +def _schema_nodes_at_path(schema: SchemaRecord, path: object) -> tuple[SchemaRecord, ...]: + if not isinstance(path, str) or not path.startswith("$"): + return () + if path == "$": + return (schema,) + if not path.startswith("$."): + return () + nodes: tuple[SchemaRecord, ...] = (schema,) + for raw_segment in path[2:].split("."): + wants_items = raw_segment.endswith("[*]") + segment = raw_segment[:-3] if wants_items else raw_segment + if not _SCHEMA_PATH_SEGMENT.fullmatch(segment): + return () + next_nodes: list[SchemaRecord] = [] + for node in nodes: + variants = [node] + for branch_key in ("anyOf", "oneOf"): + branches = node.get(branch_key) + if isinstance(branches, list): + variants.extend(branch for branch in branches if isinstance(branch, dict)) + for variant in variants: + properties = variant.get("properties") + child = properties.get(segment) if isinstance(properties, dict) else None + if not isinstance(child, dict): + continue + if not wants_items: + next_nodes.append(child) + continue + child_variants = [child] + for branch_key in ("anyOf", "oneOf"): + branches = child.get(branch_key) + if isinstance(branches, list): + child_variants.extend(branch for branch in branches if isinstance(branch, dict)) + next_nodes.extend( + item_schema + for child_variant in child_variants + if isinstance(item_schema := child_variant.get("items"), dict) + ) + nodes = tuple(next_nodes) + if not nodes: + return () + return nodes + + +def _schema_types(node: Mapping[str, object]) -> set[str]: + schema_type = node.get("type") + if isinstance(schema_type, str): + return {schema_type} + if isinstance(schema_type, list): + return {item for item in schema_type if isinstance(item, str)} + types: set[str] = set() + for keyword in ("anyOf", "oneOf"): + variants = node.get(keyword) + if isinstance(variants, list): + for variant in variants: + if isinstance(variant, Mapping): + types.update(_schema_types(variant)) + return types + + +def _schema_type(node: Mapping[str, object]) -> str | None: + schema_type = node.get("type") + if isinstance(schema_type, str): + return schema_type + if isinstance(schema_type, list): + types = [item for item in schema_type if isinstance(item, str) and item != "null"] + return types[0] if len(types) == 1 else None + return None + + +def _schema_nodes_have_type(nodes: tuple[SchemaRecord, ...], allowed: set[str]) -> bool: + return bool(nodes) and all(bool(types := _schema_types(node)) and types <= allowed for node in nodes) + + +def _annotation_supported(key: str, value: object, root_schema: SchemaRecord) -> bool: + if key not in _SUPPORTED_SYNTHETIC_ANNOTATIONS: + return False + if key == "x-polylogue-format": + return isinstance(value, str) and value in _SUPPORTED_FORMAT_VALUES + if key == "x-polylogue-semantic-role": + return isinstance(value, str) and value in _SUPPORTED_SEMANTIC_ROLE_VALUES + if key == "x-polylogue-frequency": + frequency = _number(value) + return frequency is not None and 0 <= frequency <= 1 + if key == "x-polylogue-multiline": + return isinstance(value, bool) + if key == "x-polylogue-values": + return isinstance(value, list) and bool(value) and all(isinstance(item, str) and item for item in value) + if key == "x-polylogue-range": + return _valid_pair(value) + if key == "x-polylogue-array-lengths": + return _valid_pair(value, integral=True, nonnegative=True) + if key == "x-polylogue-observed-distribution": + return _valid_observed_distribution(value) + required = { + "x-polylogue-foreign-keys": ("source", "target"), + "x-polylogue-time-deltas": ("field_a", "field_b"), + "x-polylogue-mutually-exclusive": ("parent",), + "x-polylogue-string-lengths": ("path",), + }.get(key) + if required is None: + return False + records = _relation_records(value, required) + if records is None: + return False + if key == "x-polylogue-foreign-keys": + return all( + _schema_nodes_at_path(root_schema, record["source"]) + and _schema_nodes_at_path(root_schema, record["target"]) + for record in records + ) + if key == "x-polylogue-time-deltas": + return all( + _schema_nodes_have_type( + _schema_nodes_at_path(root_schema, record["field_a"]), {"string", "number", "integer"} + ) + and _schema_nodes_have_type( + _schema_nodes_at_path(root_schema, record["field_b"]), {"string", "number", "integer"} + ) + and _valid_pair([record.get("min_delta"), record.get("max_delta")]) + and _number(record.get("avg_delta")) is not None + for record in records + ) + if key == "x-polylogue-mutually-exclusive": + return all( + isinstance(fields := record.get("fields"), list) + and len(fields) >= 2 + and all( + isinstance(field, str) + and _schema_nodes_have_type( + _schema_nodes_at_path(root_schema, f"{record['parent']}.{field}"), + {"string", "number", "integer", "boolean", "array", "object", "null"}, + ) + for field in fields + ) + for record in records + ) + return all( + _schema_nodes_at_path(root_schema, record["path"]) + and any("string" in _schema_types(node) for node in _schema_nodes_at_path(root_schema, record["path"])) + and _valid_pair([record.get("min"), record.get("max")], integral=True, nonnegative=True) + and _number(record.get("avg")) is not None + and (stddev := _number(record.get("stddev"))) is not None + and stddev >= 0 + for record in records + ) + + +def _annotation_is_supported_at_node( + key: str, value: object, node: Mapping[str, object], path: str, root_schema: SchemaRecord +) -> bool: + schema_type = _schema_type(node) + schema_types = _schema_types(node) + union_path = ".anyOf[" in path or ".oneOf[" in path + if key in { + "x-polylogue-foreign-keys", + "x-polylogue-time-deltas", + "x-polylogue-mutually-exclusive", + "x-polylogue-string-lengths", + }: + if key in {"x-polylogue-foreign-keys", "x-polylogue-time-deltas"}: + return False + return path == "$" and _annotation_supported(key, value, root_schema) + if key == "x-polylogue-frequency": + return path != "$" and _annotation_supported(key, value, root_schema) + if key == "x-polylogue-array-lengths": + return _annotation_supported(key, value, root_schema) and ("array" in schema_types or union_path) + if key == "x-polylogue-observed-distribution": + if schema_type not in {"array", "number", "integer"} or not isinstance(value, Mapping): + return False + expected_distribution = "array_length" if schema_type == "array" else "numeric" + return expected_distribution in value and _annotation_supported(key, value, root_schema) + if key == "x-polylogue-range": + return schema_type in {"number", "integer"} and _annotation_supported(key, value, root_schema) + if key in {"x-polylogue-format", "x-polylogue-values", "x-polylogue-multiline"}: + if key == "x-polylogue-format" and schema_type in {"number", "integer"}: + return value == "unix-epoch" + return _annotation_supported(key, value, root_schema) and ( + schema_type == "string" or "string" in schema_types or union_path + ) + if key == "x-polylogue-semantic-role": + role = value if isinstance(value, str) else None + if role == "message_timestamp": + return bool(schema_types & {"string", "number", "integer"}) and _annotation_supported( + key, value, root_schema + ) + if role == "message_container": + return "object" in schema_types + return schema_type == "string" and _annotation_supported(key, value, root_schema) + return key in _PERSISTED_SCHEMA_METADATA_ANNOTATIONS + + +def classify_schema_constructs(schema: object) -> tuple[ConstructSupport, ...]: + """Classify exactly what the production synthetic runtime can consume.""" + + found: dict[str, ConstructSupportState] = {} + + def record(key: str, state: ConstructSupportState) -> None: + if found.get(key) == "unsupported": + return + found[key] = state + + def visit(node: object, path: str = "$") -> None: + if not isinstance(node, Mapping): + return + for key, value in node.items(): + if not isinstance(key, str): + continue + if key.startswith("x-"): + if key not in _PERSISTED_SCHEMA_METADATA_ANNOTATIONS: + record( + key, + "supported" + if _annotation_is_supported_at_node( + key, value, node, path, schema if isinstance(schema, dict) else {} + ) + else "unsupported", + ) + continue + record(key, "supported" if key in _SUPPORTED_SCHEMA_CONSTRUCTS else "unsupported") + if key in _SCHEMA_MAPPING_KEYWORDS and isinstance(value, Mapping): + if key in {"$defs", "dependentSchemas", "dependencies", "patternProperties", "properties"}: + for child_name, child in value.items(): + child_path = f"{path}.{child_name}" if key == "properties" else f"{path}.{key}.{child_name}" + visit(child, child_path) + else: + visit(value, f"{path}.{key}") + elif key in _SCHEMA_ARRAY_KEYWORDS and isinstance(value, Sequence) and not isinstance(value, str): + for index, child in enumerate(value): + visit(child, f"{path}.{key}[{index}]") + elif key == "items": + if isinstance(value, Sequence) and not isinstance(value, str): + for index, child in enumerate(value): + visit(child, f"{path}.items[{index}]") + else: + visit(value, f"{path}.items") + elif key == "additionalProperties" and isinstance(value, Mapping): + visit(value, f"{path}.additionalProperties") + + visit(schema) + return tuple(ConstructSupport(construct, found[construct]) for construct in sorted(found)) + + +def unsupported_schema_constructs(schema: object) -> tuple[str, ...]: + """Return the canonical unsupported construct details for one schema.""" + + return tuple(item.construct for item in classify_schema_constructs(schema) if item.state == "unsupported") + + +__all__ = ["ConstructSupport", "ConstructSupportState", "classify_schema_constructs", "unsupported_schema_constructs"] diff --git a/tests/infra/inferred_corpus.py b/tests/infra/inferred_corpus.py index a1fdc2236f..dbcf82d30d 100644 --- a/tests/infra/inferred_corpus.py +++ b/tests/infra/inferred_corpus.py @@ -11,22 +11,27 @@ import hashlib import json -import math -import re from collections.abc import Mapping, Sequence from dataclasses import dataclass, replace from pathlib import Path from typing import Literal, TypeAlias, cast from polylogue.core.json import JSONDocument +from polylogue.core.sources import origin_from_provider +from polylogue.maintenance.schema_inference_gate import validate_schema_inference_gate_receipt from polylogue.scenarios import CorpusSpec +from polylogue.schemas.operator.receipt import ( + SchemaInferenceReceipt, + SchemaReceiptRegistry, + package_hashes_for_registry, +) from polylogue.schemas.operator.registry import RuntimeSchemaRegistryLike from polylogue.schemas.packages import SchemaElementManifest, SchemaPackageCatalog, SchemaVersionPackage from polylogue.schemas.synthetic import SyntheticCorpus +from polylogue.schemas.synthetic.classification import ConstructSupport, classify_schema_constructs from polylogue.schemas.synthetic.models import SchemaRecord, SyntheticSchemaSelection from polylogue.schemas.synthetic.wire_formats import PROVIDER_WIRE_FORMATS, WireFormat -ConstructSupportState: TypeAlias = Literal["supported", "unsupported"] INFERRED_CORPUS_MANIFEST_SCHEMA_VERSION = 1 UnsupportedCorpusReason: TypeAlias = Literal[ "provider_without_wire_format", @@ -36,190 +41,6 @@ ] PackageReceipt: TypeAlias = JSONDocument -# These are the schema constructs the current recursive synthetic runtime can -# consume as structure. ``additionalProperties`` is intentionally included: -# the generator emits declared properties and does not need to materialize -# arbitrary unknown keys for the current corpus contract. -_SUPPORTED_SCHEMA_CONSTRUCTS = frozenset( - { - "$anchor", - "$comment", - "$id", - "$schema", - "additionalProperties", - "anyOf", - "default", - "deprecated", - "description", - "examples", - "items", - "oneOf", - "properties", - "readOnly", - "title", - "type", - "writeOnly", - } -) - -# The synthetic runtime can select structural branches and emit declared -# properties, but it does not validate generated values against JSON Schema. -# Every assertion keyword outside the explicit structural subset therefore -# fails closed. This keeps a manifest spec from implying conformance to a -# pattern, format, range, collection bound, or other constraint it cannot -# prove. -_STANDARD_SCHEMA_KEYWORDS = frozenset( - { - "$anchor", - "$comment", - "$defs", - "$dynamicAnchor", - "$dynamicRef", - "$id", - "$recursiveAnchor", - "$recursiveRef", - "$ref", - "$schema", - "$vocabulary", - "additionalProperties", - "allOf", - "anyOf", - "const", - "contains", - "contentEncoding", - "contentMediaType", - "contentSchema", - "default", - "definitions", - "dependentRequired", - "dependentSchemas", - "dependencies", - "deprecated", - "description", - "else", - "enum", - "examples", - "exclusiveMaximum", - "exclusiveMinimum", - "format", - "formatAssertion", - "if", - "items", - "maxContains", - "maxItems", - "maxLength", - "maxProperties", - "maximum", - "minContains", - "minItems", - "minLength", - "minProperties", - "minimum", - "multipleOf", - "not", - "oneOf", - "pattern", - "patternProperties", - "prefixItems", - "properties", - "propertyNames", - "readOnly", - "required", - "then", - "title", - "type", - "unevaluatedItems", - "unevaluatedProperties", - "uniqueItems", - "writeOnly", - } -) -_SCHEMA_MAPPING_KEYWORDS = frozenset( - { - "$defs", - "additionalProperties", - "contentSchema", - "contains", - "dependentSchemas", - "dependencies", - "else", - "if", - "items", - "not", - "patternProperties", - "properties", - "propertyNames", - "then", - "unevaluatedItems", - "unevaluatedProperties", - } -) -_SCHEMA_ARRAY_KEYWORDS = frozenset({"allOf", "anyOf", "oneOf", "prefixItems"}) - -# These annotations are read by the production synthetic runtime, semantic -# value generator, or relation solver. Their support verdict means the -# corresponding generator path consumes the annotation, rather than merely -# tolerating its namespace. -_SUPPORTED_SYNTHETIC_ANNOTATIONS = frozenset( - { - "x-polylogue-array-lengths", - "x-polylogue-foreign-keys", - "x-polylogue-format", - "x-polylogue-frequency", - "x-polylogue-multiline", - "x-polylogue-mutually-exclusive", - "x-polylogue-observed-distribution", - "x-polylogue-range", - "x-polylogue-semantic-role", - "x-polylogue-string-lengths", - "x-polylogue-time-deltas", - "x-polylogue-values", - } -) -_SUPPORTED_FORMAT_VALUES = frozenset( - {"uuid4", "uuid", "hex-id", "iso8601", "unix-epoch", "unix-epoch-str", "url", "email", "mime-type", "base64"} -) -_SUPPORTED_SEMANTIC_ROLE_VALUES = frozenset({"message_role", "message_body", "message_timestamp", "session_title"}) - -_PERSISTED_SCHEMA_METADATA_ANNOTATIONS = frozenset( - { - "x-polylogue-anchor-profile-family-id", - "x-polylogue-artifact-kind", - "x-polylogue-element-bundle-scope-count", - "x-polylogue-element-first-seen", - "x-polylogue-element-kind", - "x-polylogue-element-last-seen", - "x-polylogue-evidence", - "x-polylogue-evidence-confidence", - "x-polylogue-exact-structure-ids", - "x-polylogue-generated-at", - "x-polylogue-generator", - "x-polylogue-high-cardinality-keys", - "x-polylogue-observed-artifact-count", - "x-polylogue-package-profile-family-ids", - "x-polylogue-package-version", - "x-polylogue-profile-family-ids", - "x-polylogue-profile-tokens", - "x-polylogue-promoted-at", - "x-polylogue-registered-at", - "x-polylogue-sample-count", - "x-polylogue-sample-granularity", - "x-polylogue-score", - "x-polylogue-version", - } -) - - -@dataclass(frozen=True, order=True) -class ConstructSupport: - """Support verdict for one JSON Schema construct found in an element.""" - - construct: str - state: ConstructSupportState - - def to_payload(self) -> dict[str, str]: - return {"construct": self.construct, "state": self.state} - @dataclass(frozen=True, order=True) class CorpusManifestKey: @@ -398,6 +219,31 @@ def from_payload(cls, payload: Mapping[str, object]) -> InferredCorpusManifest: return manifest +def _require_inference_handoff(manifest: InferredCorpusManifest) -> SchemaInferenceReceipt: + if manifest.receipt_state == "catalog_only" or manifest.package_receipt is None: + raise ValueError("campaign mode requires a persisted schema-inference handoff, not catalog-only data") + return SchemaInferenceReceipt.from_payload(manifest.package_receipt) + + +def _validate_authoritative_gate_binding( + receipt: SchemaInferenceReceipt, + *, + gate_receipt_path: Path | None, + archive_root: Path | None, +) -> None: + if gate_receipt_path is None or archive_root is None: + raise ValueError("campaign mode requires an authoritative gate receipt path and archive root") + try: + payload = json.loads(gate_receipt_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError, ValueError) as exc: + raise ValueError(f"unable to read authoritative schema-inference gate receipt: {exc}") from exc + if not isinstance(payload, Mapping): + raise ValueError("authoritative schema-inference gate receipt must be a JSON object") + gate_digest = validate_schema_inference_gate_receipt(payload, archive_root=archive_root) + if receipt.gate_receipt_digest != gate_digest: + raise ValueError("schema-inference handoff gate receipt digest does not match the authoritative PASS receipt") + + @dataclass(frozen=True) class InferredCorpusConvergenceHandoff: """Exact executable manifest subset admitted to the convergence loop.""" @@ -552,7 +398,14 @@ def write_inferred_corpus_manifest(manifest: InferredCorpusManifest, path: Path) ) -def read_inferred_corpus_manifest(path: Path) -> InferredCorpusManifest: +def read_inferred_corpus_manifest( + path: Path, + *, + campaign_mode: bool = False, + registry: RuntimeSchemaRegistryLike | None = None, + gate_receipt_path: Path | None = None, + archive_root: Path | None = None, +) -> InferredCorpusManifest: """Read and validate a persisted manifest before exposing executable rows.""" try: @@ -565,437 +418,61 @@ def read_inferred_corpus_manifest(path: Path) -> InferredCorpusManifest: raise ValueError(f"unable to read inferred corpus manifest {path}: {exc}") from exc if not isinstance(payload, dict): raise ValueError("inferred corpus manifest root must be a JSON object") - return InferredCorpusManifest.from_payload(payload) - - -def _is_number(value: object) -> bool: - if not isinstance(value, (int, float)) or isinstance(value, bool): - return False - try: - return math.isfinite(float(value)) - except (OverflowError, ValueError): - return False - - -def _number(value: object) -> int | float | None: - if isinstance(value, int) and not isinstance(value, bool): - return value if _is_number(value) else None - if isinstance(value, float) and not isinstance(value, bool) and math.isfinite(value): - return value - return None - - -def _valid_pair(value: object, *, integral: bool = False, nonnegative: bool = False) -> bool: - if not isinstance(value, list) or len(value) != 2: - return False - first = _number(value[0]) - second = _number(value[1]) - if first is None or second is None: - return False - if integral and (not isinstance(first, int) or not isinstance(second, int)): - return False - if nonnegative and (first < 0 or second < 0): - return False - return first <= second - - -def _histogram_bucket_is_safe(index: int, log_base: int | float) -> bool: - try: - sampled_value = math.expm1((abs(float(index)) - 0.5) * math.log(float(log_base))) - except (OverflowError, ValueError): - return False - return math.isfinite(sampled_value) - - -def _valid_observed_distribution(value: object) -> bool: - if not isinstance(value, Mapping) or not value: - return False - for distribution in value.values(): - if not isinstance(distribution, Mapping): - return False - histogram = distribution.get("histogram") - log_base = _number(distribution.get("log_base")) - if not isinstance(histogram, list) or not histogram or log_base is None or log_base <= 1: - return False - log_base_float = float(log_base) - if not all( - isinstance(bucket, list) - and len(bucket) == 2 - and isinstance(bucket[0], int) - and not isinstance(bucket[0], bool) - and isinstance(bucket[1], int) - and not isinstance(bucket[1], bool) - and _number(bucket[0]) is not None - and _number(bucket[1]) is not None - and bucket[1] > 0 - and _histogram_bucket_is_safe(bucket[0], log_base_float) - for bucket in histogram - ): - return False - for key in ("min", "max", "mean", "p0", "p50", "p90", "p95", "p99", "p100", "stddev"): - if key in distribution and not _is_number(distribution[key]): - return False - minimum = _number(distribution.get("min")) - maximum = _number(distribution.get("max")) - if minimum is not None and maximum is not None and minimum > maximum: - return False - stddev = _number(distribution.get("stddev")) - if stddev is not None and stddev < 0: - return False - ordered_stats = tuple(_number(distribution.get(key)) for key in ("p0", "p50", "p90", "p95", "p99", "p100")) - present_stats = tuple(value for value in ordered_stats if value is not None) - if any(left > right for left, right in zip(present_stats, present_stats[1:], strict=False)): - return False - if any( - (minimum is not None and value < minimum) or (maximum is not None and value > maximum) - for value in present_stats - ): - return False - mean = _number(distribution.get("mean")) - if mean is not None: - if minimum is not None and mean < minimum: - return False - if maximum is not None and mean > maximum: - return False - p0 = _number(distribution.get("p0")) - p100 = _number(distribution.get("p100")) - if p0 is not None and mean < p0: - return False - if p100 is not None and mean > p100: - return False - return True - - -def _relation_records(value: object, required: tuple[str, ...]) -> list[dict[str, object]] | None: - if not isinstance(value, list) or not value: - return None - records: list[dict[str, object]] = [] - for record in value: - if not isinstance(record, dict): - return None - if not all(isinstance(record.get(field), str) and record[field].strip() for field in required): - return None - records.append(record) - return records - - -_SCHEMA_PATH_SEGMENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_-]*$") - - -def _schema_nodes_at_path(schema: SchemaRecord, path: object) -> tuple[SchemaRecord, ...]: - """Resolve the JSONPath subset emitted by the relation solver.""" - - if not isinstance(path, str) or not path.startswith("$"): - return () - if path == "$": - return (schema,) - if not path.startswith("$."): - return () - nodes: tuple[SchemaRecord, ...] = (schema,) - for raw_segment in path[2:].split("."): - if raw_segment.endswith("[*]"): - segment = raw_segment[:-3] - wants_items = True - else: - segment = raw_segment - wants_items = False - if not _SCHEMA_PATH_SEGMENT.fullmatch(segment): - return () - next_nodes: list[SchemaRecord] = [] - for node in nodes: - variants = [node] - for branch_key in ("anyOf", "oneOf"): - branches = node.get(branch_key) - if isinstance(branches, list): - variants.extend(branch for branch in branches if isinstance(branch, dict)) - for variant in variants: - properties = variant.get("properties") - child = properties.get(segment) if isinstance(properties, dict) else None - if not isinstance(child, dict): - continue - if wants_items: - child_variants = [child] - for child_branch_key in ("anyOf", "oneOf"): - child_branches = child.get(child_branch_key) - if isinstance(child_branches, list): - child_variants.extend(branch for branch in child_branches if isinstance(branch, dict)) - for child_variant in child_variants: - items = child_variant.get("items") - if isinstance(items, dict): - next_nodes.append(items) - else: - next_nodes.append(child) - nodes = tuple(next_nodes) - if not nodes: - return () - return nodes - - -def _schema_nodes_have_type(nodes: tuple[SchemaRecord, ...], allowed: set[str]) -> bool: - return bool(nodes) and all(bool(types := _schema_types(node)) and types <= allowed for node in nodes) - - -def _schema_type_family(nodes: tuple[SchemaRecord, ...]) -> str | None: - types = {_schema_type(node) for node in nodes} - if types and types <= {"string"}: - return "string" - if types and types <= {"number", "integer"}: - return "numeric" - return None - - -def _relation_annotation_paths_are_enforced( - key: str, - value: object, - root_schema: SchemaRecord, -) -> bool: - records = _relation_records( - value, - { - "x-polylogue-foreign-keys": ("source", "target"), - "x-polylogue-time-deltas": ("field_a", "field_b"), - "x-polylogue-mutually-exclusive": ("parent",), - "x-polylogue-string-lengths": ("path",), - }[key], - ) - if records is None: - return False - if key == "x-polylogue-foreign-keys": - return False - if key == "x-polylogue-time-deltas": - # The solver parses these records, but no production generation path - # calls get_time_delta yet. Keep them fail-closed until it does. - return False - if key == "x-polylogue-mutually-exclusive": - for record in records: - fields = record.get("fields") - if not isinstance(fields, list) or not all(isinstance(field, str) for field in fields): - return False - if not all( - _schema_nodes_have_type( - _schema_nodes_at_path(root_schema, f"{record['parent']}.{field}"), - {"string", "number", "integer", "boolean", "array", "object", "null"}, - ) - for field in fields - ): - return False - return True - if key == "x-polylogue-string-lengths": - return all( - bool( - _schema_nodes_at_path(root_schema, record["path"]) - and any("string" in _schema_types(node) for node in _schema_nodes_at_path(root_schema, record["path"])) - ) - for record in records + manifest = InferredCorpusManifest.from_payload(payload) + if campaign_mode: + _require_inference_handoff(manifest) + if registry is None: + raise ValueError("campaign mode requires a live schema registry") + providers = tuple(sorted({entry.key.provider for entry in manifest.entries})) + _validate_inference_handoff( + manifest, + registry, + providers=providers, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, ) - return False - - -def _synthetic_annotation_is_enforced(key: str, value: object) -> bool: - """Return whether the production generator or solver enforces this payload.""" - - if key not in _SUPPORTED_SYNTHETIC_ANNOTATIONS: - return False - if key == "x-polylogue-format": - return isinstance(value, str) and value in _SUPPORTED_FORMAT_VALUES - if key == "x-polylogue-semantic-role": - return isinstance(value, str) and value in _SUPPORTED_SEMANTIC_ROLE_VALUES - if key == "x-polylogue-frequency": - frequency = _number(value) - return frequency is not None and 0 <= frequency <= 1 - if key == "x-polylogue-multiline": - return isinstance(value, bool) - if key == "x-polylogue-values": - return isinstance(value, list) and bool(value) and all(isinstance(item, str) and item for item in value) - if key == "x-polylogue-range": - return _valid_pair(value) - if key == "x-polylogue-array-lengths": - return _valid_pair(value, integral=True, nonnegative=True) - if key == "x-polylogue-observed-distribution": - return _valid_observed_distribution(value) - if key == "x-polylogue-foreign-keys": - return _relation_records(value, ("source", "target")) is not None - if key == "x-polylogue-time-deltas": - # The solver parses these records, but no production generation path - # calls get_time_delta yet. Keep them fail-closed until it does. - return False - if key == "x-polylogue-mutually-exclusive": - records = _relation_records(value, ("parent",)) - return records is not None and all( - isinstance(fields := record.get("fields"), list) - and len(fields) >= 2 - and all(isinstance(field, str) and field for field in fields) - for record in records - ) - if key == "x-polylogue-string-lengths": - records = _relation_records(value, ("path",)) - if records is None: - return False - string_records: list[tuple[int | float | None, int | float | None, int | float | None, int | float | None]] = [ - ( - _number(record.get("min")), - _number(record.get("max")), - _number(record.get("avg")), - _number(record.get("stddev")), - ) - for record in records - ] - for minimum, maximum, average, stddev in string_records: - if ( - not _valid_pair([minimum, maximum], integral=True, nonnegative=True) - or average is None - or stddev is None - or minimum is None - or maximum is None - or not minimum <= average <= maximum - or stddev < 0 - ): - return False - return True - raise AssertionError(f"missing annotation validator for {key}") - - -def _schema_type(node: Mapping[str, object]) -> str | None: - schema_type = node.get("type") - if isinstance(schema_type, str): - return schema_type - if isinstance(schema_type, list): - types = [item for item in schema_type if isinstance(item, str) and item != "null"] - return types[0] if len(types) == 1 else None - return None - - -def _schema_types(node: Mapping[str, object]) -> set[str]: - schema_type = node.get("type") - if isinstance(schema_type, str): - return {schema_type} - if isinstance(schema_type, list): - return {item for item in schema_type if isinstance(item, str)} - types: set[str] = set() - for keyword in ("anyOf", "oneOf"): - variants = node.get(keyword) - if isinstance(variants, list): - for variant in variants: - if isinstance(variant, Mapping): - types.update(_schema_types(variant)) - return types - - -def _annotation_is_enforced_at_node( - key: str, - value: object, - node: Mapping[str, object], - path: str, - root_schema: SchemaRecord, -) -> bool: - schema_type = _schema_type(node) - schema_types = _schema_types(node) - union_path = ".anyOf[" in path or ".oneOf[" in path - if key in { - "x-polylogue-foreign-keys", - "x-polylogue-time-deltas", - "x-polylogue-mutually-exclusive", - "x-polylogue-string-lengths", - }: - return ( - path == "$" - and _synthetic_annotation_is_enforced(key, value) - and _relation_annotation_paths_are_enforced(key, value, root_schema) - ) - if key == "x-polylogue-frequency": - return path != "$" and _synthetic_annotation_is_enforced(key, value) - if key in {"x-polylogue-array-lengths", "x-polylogue-observed-distribution"}: - if key == "x-polylogue-array-lengths": - return _synthetic_annotation_is_enforced(key, value) and ( - schema_type == "array" or "array" in schema_types or union_path - ) - if schema_type not in {"array", "number", "integer"} or not isinstance(value, Mapping): - return False - expected_distribution = "array_length" if schema_type == "array" else "numeric" - return expected_distribution in value and _valid_observed_distribution(value) - if key == "x-polylogue-range": - return schema_type in {"number", "integer"} and _synthetic_annotation_is_enforced(key, value) - if key in {"x-polylogue-format", "x-polylogue-values", "x-polylogue-multiline"}: - if key == "x-polylogue-format" and schema_type in {"number", "integer"}: - return value == "unix-epoch" - return _synthetic_annotation_is_enforced(key, value) and ( - schema_type == "string" or "string" in schema_types or union_path - ) - if key == "x-polylogue-semantic-role": - role = value if isinstance(value, str) else None - if role == "message_timestamp": - return bool(schema_types & {"string", "number", "integer"}) and _synthetic_annotation_is_enforced( - key, value - ) - if role == "message_container": - return schema_type == "object" - return schema_type == "string" and _synthetic_annotation_is_enforced(key, value) - return key in _PERSISTED_SCHEMA_METADATA_ANNOTATIONS + return manifest def _schema_constructs(schema: object) -> tuple[ConstructSupport, ...]: - """Census schema keywords and annotations at the paths production consumes.""" - - found: dict[str, ConstructSupportState] = {} - - def record_support(key: str, state: ConstructSupportState) -> None: - if found.get(key) == "unsupported": - return - found[key] = state - - def visit(node: object, path: str = "$") -> None: - if not isinstance(node, Mapping): - return - node_record = node - for key, value in node.items(): - if not isinstance(key, str): - continue - if key.startswith("x-"): - if key in _PERSISTED_SCHEMA_METADATA_ANNOTATIONS: - continue - record_support( - key, - "supported" - if _annotation_is_enforced_at_node(key, value, node_record, path, cast(SchemaRecord, schema)) - else "unsupported", - ) - continue - if key in _SUPPORTED_SCHEMA_CONSTRUCTS: - record_support(key, "supported") - elif key in _STANDARD_SCHEMA_KEYWORDS: - record_support(key, "unsupported") - else: - record_support(key, "unsupported") - - if key in _SCHEMA_MAPPING_KEYWORDS and isinstance(value, Mapping): - if key in {"$defs", "dependentSchemas", "dependencies", "patternProperties", "properties"}: - for child_name, child in value.items(): - child_path = f"{path}.{child_name}" if key == "properties" else f"{path}.{key}.{child_name}" - visit(child, child_path) - else: - visit(value, f"{path}.{key}") - elif key in _SCHEMA_ARRAY_KEYWORDS and isinstance(value, Sequence) and not isinstance(value, str): - for index, child in enumerate(value): - visit(child, f"{path}.{key}[{index}]") - elif key == "items": - if isinstance(value, Sequence) and not isinstance(value, str): - for index, child in enumerate(value): - visit(child, f"{path}.items[{index}]") - else: - visit(value, f"{path}.items") - elif key == "additionalProperties" and isinstance(value, Mapping): - visit(value, f"{path}.additionalProperties") - - visit(schema) - return tuple(ConstructSupport(construct, found[construct]) for construct in sorted(found)) + """Use the production classifier for campaign admission decisions.""" + + return classify_schema_constructs(schema) def build_inferred_corpus_convergence_handoff( manifest: InferredCorpusManifest | Path, + *, + campaign_mode: bool = False, + registry: RuntimeSchemaRegistryLike | None = None, + gate_receipt_path: Path | None = None, + archive_root: Path | None = None, ) -> InferredCorpusConvergenceHandoff: """Bind every supported row from memory or persisted disk to convergence.""" - persisted_manifest = read_inferred_corpus_manifest(manifest) if isinstance(manifest, Path) else manifest + persisted_manifest = ( + read_inferred_corpus_manifest( + manifest, + campaign_mode=campaign_mode, + registry=registry, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + if isinstance(manifest, Path) + else manifest + ) + if campaign_mode: + _require_inference_handoff(persisted_manifest) + if registry is None: + raise ValueError("campaign mode requires a live schema registry") + providers = tuple(sorted({entry.key.provider for entry in persisted_manifest.entries})) + _validate_inference_handoff( + persisted_manifest, + registry, + providers=providers, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) selections = tuple(_selection_for_entry(entry) for entry in persisted_manifest.entries if entry.spec is not None) handoff = InferredCorpusConvergenceHandoff( manifest_id=persisted_manifest.manifest_id, @@ -1053,9 +530,11 @@ def _stable_seed(key: CorpusManifestKey) -> int: def _catalog_entries( registry: RuntimeSchemaRegistryLike, + providers: Sequence[str] | None = None, ) -> tuple[tuple[str, SchemaPackageCatalog, SchemaVersionPackage, SchemaElementManifest], ...]: result: list[tuple[str, SchemaPackageCatalog, SchemaVersionPackage, SchemaElementManifest]] = [] - for provider in sorted(set(registry.list_providers())): + provider_names = providers if providers is not None else registry.list_providers() + for provider in sorted(set(provider_names)): catalog = registry.load_package_catalog(provider) if catalog is None: raise RuntimeError(f"registry provider {provider!r} has no persisted package catalog") @@ -1072,12 +551,12 @@ def _unsupported_reason( wire_format: WireFormat | None, construct_support: tuple[ConstructSupport, ...], ) -> UnsupportedCorpusRecord | None: - if wire_format is None: - return UnsupportedCorpusRecord("provider_without_wire_format") if not element.supported: return UnsupportedCorpusRecord("unsupported_element") if schema is None or element.schema_file is None: return UnsupportedCorpusRecord("missing_schema") + if wire_format is None: + return UnsupportedCorpusRecord("provider_without_wire_format") unsupported_constructs = tuple(item.construct for item in construct_support if item.state == "unsupported") if unsupported_constructs: return UnsupportedCorpusRecord("unsupported_json_schema_construct", unsupported_constructs) @@ -1128,16 +607,23 @@ def _compile_entry( # Construct the production generator against the exact package/version/ # element. No generation is performed here, so compiling the receipt does # not turn an unverified catalog into an inference claim. - SyntheticCorpus.from_selection( - SyntheticSchemaSelection( - provider=provider, - package_version=package.version, - element_kind=element.element_kind, - schema=schema, - wire_format=wire_format, - workload_profile=workload_profile if isinstance(workload_profile, dict) else None, - ) + selection = SyntheticSchemaSelection( + provider=provider, + package_version=package.version, + element_kind=element.element_kind, + schema=schema, + wire_format=wire_format, + workload_profile=workload_profile if isinstance(workload_profile, dict) else None, + ) + witness = SyntheticCorpus.from_selection(selection).generate( + count=1, + messages_per_session=range(spec.messages_min, spec.messages_min + 1), + seed=spec.seed, + style=spec.style, + session_native_ids=spec.session_native_ids[:1], ) + if len(witness) != 1 or not witness[0]: + raise ValueError("persisted schema selection did not produce a real synthetic corpus witness") return InferredCorpusManifestEntry( key=key, spec=spec, @@ -1149,12 +635,14 @@ def _compile_entry( def assert_inferred_corpus_manifest_complete( manifest: InferredCorpusManifest, registry: RuntimeSchemaRegistryLike, + *, + providers: Sequence[str] | None = None, ) -> None: """Fail loudly when a manifest omits any currently persisted catalog entry.""" expected = { CorpusManifestKey(provider, package.version, element.element_kind) - for provider, _catalog, package, element in _catalog_entries(registry) + for provider, _catalog, package, element in _catalog_entries(registry, providers) } actual = { CorpusManifestKey(entry.key.provider, entry.key.package_version, entry.key.element_kind) @@ -1173,10 +661,16 @@ def compile_inferred_corpus_manifest( registry: RuntimeSchemaRegistryLike, package_receipt: PackageReceipt | None = None, wire_formats: Mapping[str, WireFormat] | None = None, + providers: Sequence[str] | None = None, + campaign_mode: bool = False, + gate_receipt_path: Path | None = None, + archive_root: Path | None = None, ) -> InferredCorpusManifest: """Compile every persisted package/version/element into a typed manifest.""" formats = PROVIDER_WIRE_FORMATS if wire_formats is None else wire_formats + if campaign_mode and package_receipt is None: + raise ValueError("campaign mode requires a persisted schema-inference handoff") entries = tuple( _compile_entry( provider=provider, @@ -1185,15 +679,143 @@ def compile_inferred_corpus_manifest( registry=registry, wire_formats=formats, ) - for provider, catalog, package, element in _catalog_entries(registry) + for provider, catalog, package, element in _catalog_entries(registry, providers) ) manifest = InferredCorpusManifest( entries=tuple(sorted(entries, key=lambda entry: entry.key)), package_receipt=package_receipt ) - assert_inferred_corpus_manifest_complete(manifest, registry) + assert_inferred_corpus_manifest_complete(manifest, registry, providers=providers) + if campaign_mode: + _validate_inference_handoff( + manifest, + registry, + providers=providers, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) return manifest +def _validate_inference_handoff( + manifest: InferredCorpusManifest, + registry: RuntimeSchemaRegistryLike, + *, + providers: Sequence[str] | None, + gate_receipt_path: Path | None, + archive_root: Path | None, +) -> None: + receipt = _require_inference_handoff(manifest) + _validate_authoritative_gate_binding( + receipt, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + if not manifest.supported_specs: + raise ValueError("campaign mode has no executable synthetic corpus selection") + expected_packages = package_hashes_for_registry(cast(SchemaReceiptRegistry, registry), providers) + if receipt.packages != expected_packages: + raise ValueError("schema-inference handoff package/version/element hashes do not match the registry") + + catalog_entries = _catalog_entries(registry, providers) + expected_coverage = {(provider, origin_from_provider(provider).value) for provider, *_rest in catalog_entries} + actual_coverage = {(item.provider, item.origin) for item in receipt.coverage_decisions} + if actual_coverage != expected_coverage: + raise ValueError( + "schema-inference handoff does not contain complete origin/provider coverage: " + f"missing={sorted(expected_coverage - actual_coverage)!r}, " + f"unexpected={sorted(actual_coverage - expected_coverage)!r}" + ) + entries_by_provider: dict[str, list[InferredCorpusManifestEntry]] = {} + for entry in manifest.entries: + entries_by_provider.setdefault(entry.key.provider, []).append(entry) + for coverage in receipt.coverage_decisions: + provider_entries = entries_by_provider.get(coverage.provider, []) + if any(entry.spec is not None for entry in provider_entries): + expected_decision = "committed" + elif provider_entries and all( + entry.unsupported is not None and entry.unsupported.reason == "unsupported_json_schema_construct" + for entry in provider_entries + ): + expected_decision = "nonrepresentable" + else: + expected_decision = "unsupported" + if coverage.decision != expected_decision: + raise ValueError("schema-inference handoff coverage decision changed") + + for provider, _catalog, package, element in catalog_entries: + live_entry = next( + ( + candidate + for candidate in manifest.entries + if (candidate.key.provider, candidate.key.package_version, candidate.key.element_kind) + == (provider, package.version, element.element_kind) + ), + None, + ) + if live_entry is None: + raise ValueError("schema-inference manifest is missing a live registry entry") + live_schema = registry.get_element_schema(provider, version=package.version, element_kind=element.element_kind) + live_constructs = _schema_constructs(live_schema) + if live_entry.key.construct_support != live_constructs: + raise ValueError("schema-inference manifest classifier output changed") + live_unsupported = _unsupported_reason( + element=element, + schema=live_schema if isinstance(live_schema, dict) else None, + wire_format=PROVIDER_WIRE_FORMATS.get(provider), + construct_support=live_constructs, + ) + if (live_entry.unsupported is None) != (live_unsupported is None): + raise ValueError("schema-inference manifest executable support changed") + if live_unsupported is not None: + if live_entry.unsupported != live_unsupported: + raise ValueError("schema-inference manifest unsupported decision changed") + continue + if live_entry.generator_schema != live_schema: + raise ValueError("schema-inference manifest generator schema changed") + selection = _selection_for_entry(live_entry) + spec = live_entry.spec + if spec is None: + raise ValueError("schema-inference manifest executable entry has no corpus spec") + witness = SyntheticCorpus.from_selection(selection).generate( + count=1, + messages_per_session=range(spec.messages_min, spec.messages_min + 1), + seed=spec.seed, + style=spec.style, + session_native_ids=spec.session_native_ids[:1], + ) + if len(witness) != 1 or not witness[0]: + raise ValueError("schema-inference manifest selection produced no executable witness") + + expected_unsupported = { + ( + entry.key.provider, + entry.key.package_version, + entry.key.element_kind, + "nonrepresentable" if entry.unsupported.reason == "unsupported_json_schema_construct" else "unsupported", + entry.unsupported.reason, + entry.unsupported.details, + ) + for entry in manifest.entries + if entry.unsupported is not None + } + actual_unsupported = { + ( + item.provider, + item.package_version, + item.element_kind, + item.decision, + item.reason, + item.details, + ) + for item in receipt.unsupported_decisions + } + if actual_unsupported != expected_unsupported: + raise ValueError( + "schema-inference handoff unsupported/nonrepresentable decisions changed: " + f"expected={sorted(expected_unsupported)!r}, actual={sorted(actual_unsupported)!r}" + ) + + __all__ = [ "ConstructSupport", "CorpusManifestKey", diff --git a/tests/unit/devtools/test_schema_commit_command.py b/tests/unit/devtools/test_schema_commit_command.py index f68c657aee..99845ddc51 100644 --- a/tests/unit/devtools/test_schema_commit_command.py +++ b/tests/unit/devtools/test_schema_commit_command.py @@ -18,10 +18,23 @@ from devtools import schema_commit from polylogue.schemas.generation.models import GenerationResult from polylogue.schemas.operator.models import SchemaCommitRequest, SchemaCommitResult, SchemaVersionCommitReport +from polylogue.schemas.operator.receipt import ( + SchemaInferenceCoverageDecision, + SchemaInferenceReceipt, +) + +_HANDOFF = SchemaInferenceReceipt( + gate_receipt_digest="a" * 64, + coverage_decisions=( + SchemaInferenceCoverageDecision(origin="codex-session", provider="codex", decision="committed", reason=None), + ), + packages=(), +) @dataclass(frozen=True) class _ConfigStub: + archive_root: Path db_path: Path @@ -36,7 +49,7 @@ def test_schema_commit_forwards_request_and_defaults_output_dir( captured: list[SchemaCommitRequest] = [] def fake_get_config() -> _ConfigStub: - return _ConfigStub(db_path=tmp_path / "archive.db") + return _ConfigStub(archive_root=tmp_path / "archive", db_path=tmp_path / "archive.db") def fake_commit(request: SchemaCommitRequest) -> SchemaCommitResult: captured.append(request) @@ -49,17 +62,9 @@ def fake_commit(request: SchemaCommitRequest) -> SchemaCommitResult: monkeypatch.setattr(schema_commit, "get_config", fake_get_config) monkeypatch.setattr(schema_commit, "commit_provider_schema", fake_commit) - authorization_calls: list[tuple[object, ...]] = [] - - @contextmanager - def allow_schema_generation(*args: object, **_kwargs: object) -> Iterator[dict[str, object]]: - authorization_calls.append(args) - yield {} - - monkeypatch.setattr(schema_commit, "authorize_schema_generation", allow_schema_generation) - assert ( - schema_commit.main(["--provider", "chatgpt", "--schema-inference-receipt", str(tmp_path / "receipt.json")]) == 0 + schema_commit.main(["--provider", "chatgpt", "--schema-inference-gate-receipt", str(tmp_path / "gate.json")]) + == 0 ) assert len(captured) == 1 @@ -69,23 +74,17 @@ def allow_schema_generation(*args: object, **_kwargs: object) -> Iterator[dict[s assert request.db_path == tmp_path / "archive.db" assert request.full_corpus is True assert request.dry_run is False - assert authorization_calls == [(tmp_path, tmp_path / "receipt.json")] - - -def test_schema_commit_refuses_persistence_without_authoritative_receipt( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str] -) -> None: - monkeypatch.setattr(schema_commit, "get_config", lambda: _ConfigStub(db_path=tmp_path / "archive.db")) - monkeypatch.setattr(schema_commit, "commit_provider_schema", pytest.fail) - - assert schema_commit.main(["--provider", "chatgpt"]) == 1 - assert "schema-inference-receipt is required" in capsys.readouterr().err + assert request.schema_inference_gate_receipt_path == tmp_path / "gate.json" def test_schema_commit_honors_output_dir_and_dry_run_overrides(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: captured: list[SchemaCommitRequest] = [] - monkeypatch.setattr(schema_commit, "get_config", lambda: _ConfigStub(db_path=tmp_path / "archive.db")) + monkeypatch.setattr( + schema_commit, + "get_config", + lambda: _ConfigStub(archive_root=tmp_path / "archive", db_path=tmp_path / "archive.db"), + ) def fake_commit(request: SchemaCommitRequest) -> SchemaCommitResult: captured.append(request) @@ -101,7 +100,16 @@ def fake_commit(request: SchemaCommitRequest) -> SchemaCommitResult: custom_output = tmp_path / "custom-providers" assert ( schema_commit.main( - ["--provider", "chatgpt", "--output-dir", str(custom_output), "--dry-run", "--no-full-corpus"] + [ + "--provider", + "chatgpt", + "--output-dir", + str(custom_output), + "--dry-run", + "--no-full-corpus", + "--schema-inference-gate-receipt", + str(tmp_path / "gate.json"), + ] ) == 0 ) @@ -114,8 +122,11 @@ def fake_commit(request: SchemaCommitRequest) -> SchemaCommitResult: def test_schema_commit_json_output_reports_success( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str] ) -> None: - monkeypatch.setattr(schema_commit, "get_config", lambda: _ConfigStub(db_path=tmp_path / "archive.db")) - monkeypatch.setattr(schema_commit, "authorize_schema_generation", _allow_schema_generation) + monkeypatch.setattr( + schema_commit, + "get_config", + lambda: _ConfigStub(archive_root=tmp_path / "archive", db_path=tmp_path / "archive.db"), + ) monkeypatch.setattr( schema_commit, "commit_provider_schema", @@ -128,12 +139,14 @@ def test_schema_commit_json_output_reports_success( ), ), dry_run=False, + handoff=_HANDOFF, + handoff_path=tmp_path / "handoff.json", ), ) assert ( schema_commit.main( - ["--provider", "chatgpt", "--json", "--schema-inference-receipt", str(tmp_path / "receipt.json")] + ["--provider", "chatgpt", "--json", "--schema-inference-gate-receipt", str(tmp_path / "gate.json")] ) == 0 ) @@ -145,13 +158,18 @@ def test_schema_commit_json_output_reports_success( assert payload["sample_count"] == 42 assert payload["versions"][0]["status"] == "changed" assert payload["versions"][0]["added_paths"] == ["session_document.new"] + assert payload["handoff"]["gate_receipt_digest"] == "a" * 64 + assert payload["handoff_path"] == str(tmp_path / "handoff.json") def test_schema_commit_exits_nonzero_on_generation_failure( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str] ) -> None: - monkeypatch.setattr(schema_commit, "get_config", lambda: _ConfigStub(db_path=tmp_path / "archive.db")) - monkeypatch.setattr(schema_commit, "authorize_schema_generation", _allow_schema_generation) + monkeypatch.setattr( + schema_commit, + "get_config", + lambda: _ConfigStub(archive_root=tmp_path / "archive", db_path=tmp_path / "archive.db"), + ) monkeypatch.setattr( schema_commit, "commit_provider_schema", @@ -165,7 +183,13 @@ def test_schema_commit_exits_nonzero_on_generation_failure( assert ( schema_commit.main( - ["--provider", "broken-provider", "--json", "--schema-inference-receipt", str(tmp_path / "receipt.json")] + [ + "--provider", + "broken-provider", + "--json", + "--schema-inference-gate-receipt", + str(tmp_path / "gate.json"), + ] ) == 1 ) @@ -178,8 +202,11 @@ def test_schema_commit_exits_nonzero_when_narrowed(monkeypatch: pytest.MonkeyPat """A commit that succeeds but narrows a previously-committed type must not report a clean exit code -- the whole point of the report is that a bad promotion can't land unnoticed.""" - monkeypatch.setattr(schema_commit, "get_config", lambda: _ConfigStub(db_path=tmp_path / "archive.db")) - monkeypatch.setattr(schema_commit, "authorize_schema_generation", _allow_schema_generation) + monkeypatch.setattr( + schema_commit, + "get_config", + lambda: _ConfigStub(archive_root=tmp_path / "archive", db_path=tmp_path / "archive.db"), + ) monkeypatch.setattr( schema_commit, "commit_provider_schema", @@ -196,5 +223,6 @@ def test_schema_commit_exits_nonzero_when_narrowed(monkeypatch: pytest.MonkeyPat ) assert ( - schema_commit.main(["--provider", "chatgpt", "--schema-inference-receipt", str(tmp_path / "receipt.json")]) == 1 + schema_commit.main(["--provider", "chatgpt", "--schema-inference-gate-receipt", str(tmp_path / "gate.json")]) + == 1 ) diff --git a/tests/unit/maintenance/test_schema_inference_gate.py b/tests/unit/maintenance/test_schema_inference_gate.py index cb81bb8567..d18edda3d0 100644 --- a/tests/unit/maintenance/test_schema_inference_gate.py +++ b/tests/unit/maintenance/test_schema_inference_gate.py @@ -15,6 +15,7 @@ RECEIPT_FILENAME, SchemaInferenceGateError, run_schema_inference_gate, + schema_inference_gate_receipt_digest, validate_schema_inference_gate_receipt, ) from polylogue.storage.blob_store import BlobStore @@ -489,3 +490,13 @@ def to_json(self) -> dict[str, object]: payload = _run(root, tmp_path, ground_truth=ground_truth) assert payload["verdict"] == "FAIL" assert any("untyped corpus residual" in reason for reason in payload["pass_fail_reasons"]) + + +def test_gate_receipt_digest_is_canonical_and_content_bound() -> None: + first = {"verdict": "PASS", "schema": "polylogue.schema-inference-gate.v1", "nested": {"b": 2, "a": 1}} + reordered = {"nested": {"a": 1, "b": 2}, "schema": first["schema"], "verdict": first["verdict"]} + + assert schema_inference_gate_receipt_digest(first) == schema_inference_gate_receipt_digest(reordered) + altered = dict(first) + altered["verdict"] = "FAIL" + assert schema_inference_gate_receipt_digest(first) != schema_inference_gate_receipt_digest(altered) diff --git a/tests/unit/schemas/test_inferred_corpus_manifest.py b/tests/unit/schemas/test_inferred_corpus_manifest.py index af59b08e6c..94086a4488 100644 --- a/tests/unit/schemas/test_inferred_corpus_manifest.py +++ b/tests/unit/schemas/test_inferred_corpus_manifest.py @@ -8,6 +8,14 @@ import pytest from polylogue.core.json import JSONValue +from polylogue.maintenance.schema_inference_gate import ( + run_schema_inference_gate, + schema_inference_gate_receipt_digest, +) +from polylogue.schemas.operator.receipt import ( + SchemaInferenceUnsupportedDecision, + build_schema_inference_receipt, +) from polylogue.schemas.registry import SCHEMA_DIR, SchemaRegistry from polylogue.schemas.synthetic.models import SchemaRecord from polylogue.schemas.synthetic.wire_formats import PROVIDER_WIRE_FORMATS @@ -21,12 +29,26 @@ read_inferred_corpus_manifest, write_inferred_corpus_manifest, ) +from tests.unit.maintenance.test_schema_inference_gate import _seed_archive def _registry() -> SchemaRegistry: return SchemaRegistry(storage_root=SCHEMA_DIR) +def _authoritative_gate(tmp_path: Path) -> tuple[Path, Path, str]: + archive_root = tmp_path / "archive" + receipt_path = tmp_path / "schema-inference-gate-receipt.json" + _seed_archive(archive_root) + result = run_schema_inference_gate( + archive_root, + receipt_path=receipt_path, + ground_truth_roots={"codex-session": (tmp_path / "archive-codex-ground-truth",)}, + ) + assert result.passed + return archive_root, receipt_path, schema_inference_gate_receipt_digest(result.payload) + + def _catalog_keys(registry: SchemaRegistry) -> set[CorpusManifestKey]: keys: set[CorpusManifestKey] = set() for provider in registry.list_providers(): @@ -88,6 +110,245 @@ def test_persisted_manifest_round_trip_validates_identity_and_integrity(tmp_path assert read_inferred_corpus_manifest(path) == manifest +def test_campaign_read_revalidates_live_schema_and_classifier(tmp_path: Path) -> None: + registry = _registry() + provider = "codex" + archive_root, gate_receipt_path, gate_digest = _authoritative_gate(tmp_path) + receipt = build_schema_inference_receipt(registry, provider=provider, gate_receipt_digest=gate_digest) + manifest = compile_inferred_corpus_manifest( + registry=registry, + providers=(provider,), + package_receipt=receipt.to_payload(), + campaign_mode=True, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + path = tmp_path / "campaign.json" + + supported = next(entry for entry in manifest.entries if entry.spec is not None) + tampered_schema = replace(supported, generator_schema={"type": "string"}) + tampered = replace( + manifest, + entries=tuple( + sorted( + (tampered_schema if entry is supported else entry for entry in manifest.entries), + key=lambda entry: entry.key, + ) + ), + ) + write_inferred_corpus_manifest(tampered, path) + with pytest.raises(ValueError, match="package/version/element hashes|generator schema changed"): + read_inferred_corpus_manifest( + path, + campaign_mode=True, + registry=registry, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + + tampered_key = replace(supported.key, construct_support=()) + tampered_entry = replace(supported, key=tampered_key) + tampered = replace( + manifest, + entries=tuple( + sorted( + (tampered_entry if entry is supported else entry for entry in manifest.entries), + key=lambda entry: entry.key, + ) + ), + ) + with pytest.raises(ValueError, match="classifier output changed"): + build_inferred_corpus_convergence_handoff( + tampered, + campaign_mode=True, + registry=registry, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + + +def test_campaign_mode_rejects_catalog_only_manifest(tmp_path: Path) -> None: + manifest = compile_inferred_corpus_manifest(registry=_registry()) + path = tmp_path / "catalog-only.json" + write_inferred_corpus_manifest(manifest, path) + + with pytest.raises(ValueError, match="catalog-only"): + read_inferred_corpus_manifest(path, campaign_mode=True) + with pytest.raises(ValueError, match="handoff"): + compile_inferred_corpus_manifest(registry=_registry(), campaign_mode=True) + + +def test_campaign_receipt_rejects_tampered_gate_package_and_unsupported_decisions(tmp_path: Path) -> None: + registry = _registry() + provider = "codex" + archive_root, gate_receipt_path, gate_digest = _authoritative_gate(tmp_path) + receipt = build_schema_inference_receipt( + registry, + provider=provider, + gate_receipt_digest=gate_digest, + ) + compile_inferred_corpus_manifest( + registry=registry, + providers=(provider,), + package_receipt=receipt.to_payload(), + campaign_mode=True, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + + tampered_gate = replace(receipt, gate_receipt_digest="b" * 64) + with pytest.raises(ValueError, match="different gate receipt digests"): + tampered_gate.merged_with(receipt) + + tampered_package = replace( + receipt, + packages=(replace(receipt.packages[0], package_hash="b" * 64), *receipt.packages[1:]), + ) + with pytest.raises(ValueError, match="package/version/element hashes"): + compile_inferred_corpus_manifest( + registry=registry, + providers=(provider,), + package_receipt=tampered_package.to_payload(), + campaign_mode=True, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + + +def test_campaign_rejects_fabricated_gate_digest_even_when_shape_is_valid(tmp_path: Path) -> None: + registry = _registry() + archive_root, gate_receipt_path, _gate_digest = _authoritative_gate(tmp_path) + receipt = build_schema_inference_receipt(registry, provider="codex", gate_receipt_digest="a" * 64) + + with pytest.raises(ValueError, match="does not match the authoritative PASS receipt"): + compile_inferred_corpus_manifest( + registry=registry, + providers=("codex",), + package_receipt=receipt.to_payload(), + campaign_mode=True, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + + +def test_campaign_rejects_tampered_ground_truth_denominators_after_digest_recompute(tmp_path: Path) -> None: + registry = _registry() + archive_root, gate_receipt_path, gate_digest = _authoritative_gate(tmp_path) + receipt = build_schema_inference_receipt(registry, provider="codex", gate_receipt_digest=gate_digest) + + valid_manifest = compile_inferred_corpus_manifest( + registry=registry, + providers=("codex",), + package_receipt=receipt.to_payload(), + campaign_mode=True, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + assert valid_manifest.receipt_state == "package_receipt_attached" + + tampered_gate = json.loads(gate_receipt_path.read_text(encoding="utf-8")) + denominators = dict(tampered_gate["ground_truth_denominators"]) + denominators["documents_known"] += 1 + tampered_gate["ground_truth_denominators"] = denominators + gate_receipt_path.write_text(json.dumps(tampered_gate, sort_keys=True) + "\n", encoding="utf-8") + tampered_digest = schema_inference_gate_receipt_digest(tampered_gate) + tampered_receipt = build_schema_inference_receipt( + registry, + provider="codex", + gate_receipt_digest=tampered_digest, + ) + + with pytest.raises(ValueError, match="ground-truth denominators changed"): + compile_inferred_corpus_manifest( + registry=registry, + providers=("codex",), + package_receipt=tampered_receipt.to_payload(), + campaign_mode=True, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + + +def test_bundled_registry_relation_annotations_share_one_receipt_classification(tmp_path: Path) -> None: + registry = _registry() + provider = "chatgpt" + archive_root, gate_receipt_path, gate_digest = _authoritative_gate(tmp_path) + receipt = build_schema_inference_receipt(registry, provider=provider, gate_receipt_digest=gate_digest) + manifest = compile_inferred_corpus_manifest( + registry=registry, + providers=(provider,), + package_receipt=receipt.to_payload(), + campaign_mode=False, + ) + + expected_annotations = { + "x-polylogue-foreign-keys", + "x-polylogue-mutually-exclusive", + "x-polylogue-string-lengths", + "x-polylogue-time-deltas", + } + observed = { + item.construct + for entry in manifest.entries + for item in entry.key.construct_support + if item.construct in expected_annotations + } + assert observed == expected_annotations + receipt_decisions = { + (item.provider, item.package_version, item.element_kind, item.decision, item.reason, item.details) + for item in receipt.unsupported_decisions + if item.provider == provider + } + manifest_decisions: set[tuple[str, str, str, str, str, tuple[str, ...]]] = set() + for entry in manifest.entries: + if entry.unsupported is None: + continue + unsupported = entry.unsupported + manifest_decisions.add( + ( + entry.key.provider, + entry.key.package_version, + entry.key.element_kind, + "nonrepresentable" if unsupported.reason == "unsupported_json_schema_construct" else "unsupported", + unsupported.reason, + unsupported.details, + ) + ) + assert receipt_decisions == manifest_decisions + assert all( + annotation in details + for *_identity, details in receipt_decisions + for annotation in expected_annotations + if annotation in observed + ) + + package = receipt.packages[0] + if receipt.unsupported_decisions: + first = receipt.unsupported_decisions[0] + changed = replace(first, decision="unsupported" if first.decision == "nonrepresentable" else "nonrepresentable") + tampered_decisions = (changed, *receipt.unsupported_decisions[1:]) + else: + changed = SchemaInferenceUnsupportedDecision( + provider=provider, + package_version=package.package_version, + element_kind="tampered-element", + decision="nonrepresentable", + reason="tampered decision", + details=("tampered_construct",), + ) + tampered_decisions = (changed,) + tampered_unsupported = replace(receipt, unsupported_decisions=tuple(sorted(tampered_decisions))) + with pytest.raises(ValueError, match="no executable synthetic corpus selection"): + compile_inferred_corpus_manifest( + registry=registry, + providers=(provider,), + package_receipt=tampered_unsupported.to_payload(), + campaign_mode=True, + gate_receipt_path=gate_receipt_path, + archive_root=archive_root, + ) + + @pytest.mark.parametrize("field", ["manifest_id", "payload_sha256"]) def test_persisted_manifest_rejects_tampered_hash_fields(tmp_path: Path, field: str) -> None: manifest = compile_inferred_corpus_manifest(registry=_registry()) diff --git a/tests/unit/schemas/test_operator_commit.py b/tests/unit/schemas/test_operator_commit.py index cff34a4a80..0c4e0c0510 100644 --- a/tests/unit/schemas/test_operator_commit.py +++ b/tests/unit/schemas/test_operator_commit.py @@ -8,7 +8,8 @@ called: every assertion below reads back real gzip/JSON files written by ``SchemaRegistry.replace_provider_packages`` under a real ``tmp_path``, using a fictional provider token so nothing here can read or write the repo's real -committed ``polylogue/schemas/providers/`` tree. +committed ``polylogue/schemas/providers/`` tree. The real bundled ``chatgpt`` +wire format is used for campaign execution. Only ``_build_provider_bundle`` (the sample-observation step) is mocked, the same seam ``tests/unit/core/test_schema_generation.py`` uses for @@ -21,18 +22,64 @@ import gzip import json +from dataclasses import replace from pathlib import Path from types import SimpleNamespace from typing import Any, cast from unittest.mock import patch +import pytest + +from polylogue.maintenance.schema_inference_gate import ( + run_schema_inference_gate, + schema_inference_gate_receipt_digest, + schema_inference_hard_gate_evidence_digest, +) from polylogue.schemas.generation.models import GenerationResult from polylogue.schemas.operator.commit import commit_provider_schema from polylogue.schemas.operator.models import SchemaCommitRequest +from polylogue.schemas.operator.receipt import SCHEMA_INFERENCE_HANDOFF_FILENAME, load_schema_inference_receipt from polylogue.schemas.packages import SchemaElementManifest, SchemaPackageCatalog, SchemaVersionPackage +from polylogue.schemas.registry import SchemaRegistry from polylogue.schemas.tooling_models import ClusterManifest +from tests.infra.frozen_clock import FrozenClock +from tests.infra.inferred_corpus import compile_inferred_corpus_manifest +from tests.unit.maintenance.test_schema_inference_gate import _seed_archive + +_PROVIDER = "chatgpt" -_PROVIDER = "commit-fixture-k45pq" + +def _gate_receipt(output_dir: Path) -> Path: + path = output_dir.parent / "schema-inference-gate-receipt.json" + archive_root = output_dir.parent / "archive" + if path.exists(): + return path + if not (archive_root / "source.db").exists(): + _seed_archive(archive_root) + ground_truth = archive_root.parent / f"{archive_root.name}-codex-ground-truth" + run_schema_inference_gate( + archive_root, + receipt_path=path, + ground_truth_roots={"codex-session": (ground_truth,)}, + ) + return path + + +def _request( + output_dir: Path, + *, + dry_run: bool = False, + gate_path: Path | None = None, +) -> SchemaCommitRequest: + return SchemaCommitRequest( + provider=_PROVIDER, + output_dir=output_dir, + archive_root=output_dir.parent / "archive", + db_path=output_dir.parent / "archive" / "index.db", + full_corpus=True, + schema_inference_gate_receipt_path=gate_path or _gate_receipt(output_dir), + dry_run=dry_run, + ) def _bundle( @@ -90,16 +137,19 @@ def _read_element_schema(output_dir: Path, version: str, element_kind: str = "se return cast("dict[str, Any]", json.load(handle)) +@pytest.mark.frozen_clock_modules("polylogue.maintenance.schema_inference_gate") class TestCommitProviderSchemaWritesRealFiles: + @pytest.fixture(autouse=True) + def _freeze_gate_clock(self, frozen_clock: FrozenClock) -> None: + pass + def test_new_provider_writes_catalog_and_element_files(self, tmp_path: Path) -> None: output_dir = tmp_path / "providers" schema = {"type": "object", "properties": {"id": {"type": "string"}}} bundle = _bundle(version="v1", schema=schema, sample_count=5) with patch("polylogue.schemas.generation.workflow._build_provider_bundle", return_value=bundle): - commit_result = commit_provider_schema( - SchemaCommitRequest(provider=_PROVIDER, output_dir=output_dir, full_corpus=True) - ) + commit_result = commit_provider_schema(_request(output_dir)) assert commit_result.success assert not commit_result.dry_run @@ -115,6 +165,167 @@ def test_new_provider_writes_catalog_and_element_files(self, tmp_path: Path) -> assert version_report.sample_count == 5 assert not version_report.narrowed_paths assert "session_document.id" in version_report.added_paths + assert commit_result.handoff is not None + gate_payload = json.loads(_gate_receipt(output_dir).read_text(encoding="utf-8")) + assert commit_result.handoff.gate_receipt_digest == schema_inference_gate_receipt_digest(gate_payload) + assert commit_result.handoff_path == output_dir / SCHEMA_INFERENCE_HANDOFF_FILENAME + assert load_schema_inference_receipt(commit_result.handoff_path) == commit_result.handoff + assert commit_result.handoff.packages[0].element_hashes[0].element_kind == "session_document" + + def test_commit_to_registry_to_campaign_manifest_is_a_real_route(self, tmp_path: Path) -> None: + output_dir = tmp_path / "providers" + bundle = _bundle( + version="v1", + schema={"type": "object", "properties": {"id": {"type": "string"}}}, + sample_count=5, + ) + with patch("polylogue.schemas.generation.workflow._build_provider_bundle", return_value=bundle): + result = commit_provider_schema(_request(output_dir)) + + assert result.handoff is not None + registry = SchemaRegistry(storage_root=output_dir) + manifest = compile_inferred_corpus_manifest( + registry=registry, + providers=(_PROVIDER,), + package_receipt=result.handoff.to_payload(), + campaign_mode=True, + gate_receipt_path=_gate_receipt(output_dir), + archive_root=output_dir.parent / "archive", + ) + assert manifest.receipt_state == "package_receipt_attached" + assert len(manifest.entries) == 1 + entry = manifest.entries[0] + assert entry.spec is not None + assert entry.generator_schema is not None + assert entry.key.provider == "chatgpt" + + element_path = output_dir / _PROVIDER / "versions" / "v1" / "elements" / "session_document.schema.json.gz" + mutated_schema = _read_element_schema(output_dir, "v1") + mutated_schema["title"] = "mutation" + with gzip.open(element_path, "wt", encoding="utf-8") as handle: + json.dump(mutated_schema, handle) + with pytest.raises(ValueError, match="package/version/element hashes"): + compile_inferred_corpus_manifest( + registry=SchemaRegistry(storage_root=output_dir), + providers=(_PROVIDER,), + package_receipt=result.handoff.to_payload(), + campaign_mode=True, + gate_receipt_path=_gate_receipt(output_dir), + archive_root=output_dir.parent / "archive", + ) + + def test_commit_rejects_receipt_for_archive_a_when_generation_targets_archive_b(self, tmp_path: Path) -> None: + output_dir = tmp_path / "providers" + request = replace( + _request(output_dir), + db_path=tmp_path / "archive-b" / "index.db", + ) + bundle = _bundle( + version="v1", + schema={"type": "object", "properties": {"id": {"type": "string"}}}, + sample_count=5, + ) + + with patch("polylogue.schemas.generation.workflow._build_provider_bundle", return_value=bundle): + with pytest.raises(ValueError, match="db_path must identify the active index"): + commit_provider_schema(request) + + def test_commit_requires_an_accepted_gate_receipt(self, tmp_path: Path) -> None: + output_dir = tmp_path / "providers" + with pytest.raises(ValueError, match="accepted schema-inference gate receipt"): + commit_provider_schema(SchemaCommitRequest(provider=_PROVIDER, output_dir=output_dir, full_corpus=True)) + + def test_commit_rejects_a_minimal_or_mutated_pass_payload(self, tmp_path: Path) -> None: + output_dir = tmp_path / "providers" + receipt_path = _gate_receipt(output_dir) + payload = json.loads(receipt_path.read_text(encoding="utf-8")) + payload.pop("full_blob_hash_verification") + receipt_path.write_text(json.dumps(payload), encoding="utf-8") + with pytest.raises(ValueError, match="authoritative fields"): + commit_provider_schema(_request(output_dir, gate_path=receipt_path)) + + receipt_path.unlink() + payload = json.loads(_gate_receipt(output_dir).read_text(encoding="utf-8")) + payload["full_blob_hash_verification"]["passed"] = False + receipt_path.write_text(json.dumps(payload), encoding="utf-8") + with pytest.raises(ValueError, match="full_blob_hash_verification PASS"): + commit_provider_schema(_request(output_dir, gate_path=receipt_path)) + + receipt_path.unlink() + payload = json.loads(_gate_receipt(output_dir).read_text(encoding="utf-8")) + payload["archive_identity_digest"] = "0" * 64 + receipt_path.write_text(json.dumps(payload), encoding="utf-8") + with pytest.raises(ValueError, match="archive identity"): + commit_provider_schema(_request(output_dir, gate_path=receipt_path)) + + receipt_path.unlink() + payload = json.loads(_gate_receipt(output_dir).read_text(encoding="utf-8")) + payload["receipt_nonce"] = "forged" + receipt_path.write_text(json.dumps(payload), encoding="utf-8") + with pytest.raises(ValueError, match="receipt nonce"): + commit_provider_schema(_request(output_dir, gate_path=receipt_path)) + + receipt_path.unlink() + payload = json.loads(_gate_receipt(output_dir).read_text(encoding="utf-8")) + payload["generated_at"] = "2020-01-01T00:00:00+00:00" + receipt_path.write_text(json.dumps(payload), encoding="utf-8") + with pytest.raises(ValueError, match="stale or from the future"): + commit_provider_schema(_request(output_dir, gate_path=receipt_path)) + + def test_commit_rejects_recomputed_forgery_of_live_gate_evidence(self, tmp_path: Path) -> None: + output_dir = tmp_path / "providers" + receipt_path = _gate_receipt(output_dir) + payload = json.loads(receipt_path.read_text(encoding="utf-8")) + payload["query_results"]["zero-surviving-quarantine"]["count"] = 99 + payload["hard_gate_evidence_digest"] = schema_inference_hard_gate_evidence_digest( + payload["query_results"], payload["full_blob_hash_verification"] + ) + receipt_path.write_text(json.dumps(payload), encoding="utf-8") + with pytest.raises(ValueError, match="hard-gate query results changed"): + commit_provider_schema(_request(output_dir, gate_path=receipt_path)) + + receipt_path.unlink() + payload = json.loads(_gate_receipt(output_dir).read_text(encoding="utf-8")) + payload["full_blob_hash_verification"]["before_snapshot"]["digest"] = "0" * 64 + payload["full_blob_hash_verification"]["after_snapshot"]["digest"] = "0" * 64 + payload["hard_gate_evidence_digest"] = schema_inference_hard_gate_evidence_digest( + payload["query_results"], payload["full_blob_hash_verification"] + ) + receipt_path.write_text(json.dumps(payload), encoding="utf-8") + with pytest.raises(ValueError, match="hard-gate evidence digest"): + commit_provider_schema(_request(output_dir, gate_path=receipt_path)) + + def test_registry_construct_rejection_remains_an_explicit_unsupported_entry(self, tmp_path: Path) -> None: + output_dir = tmp_path / "providers" + bundle = _bundle( + version="v1", + schema={"type": "object", "properties": {"id": {"type": "string", "enum": ["x"]}}}, + sample_count=5, + ) + with patch("polylogue.schemas.generation.workflow._build_provider_bundle", return_value=bundle): + result = commit_provider_schema(_request(output_dir)) + + assert result.handoff is not None + with pytest.raises(ValueError, match="no executable synthetic corpus selection"): + compile_inferred_corpus_manifest( + registry=SchemaRegistry(storage_root=output_dir), + providers=(_PROVIDER,), + package_receipt=result.handoff.to_payload(), + campaign_mode=True, + gate_receipt_path=_gate_receipt(output_dir), + archive_root=output_dir.parent / "archive", + ) + manifest = compile_inferred_corpus_manifest( + registry=SchemaRegistry(storage_root=output_dir), + providers=(_PROVIDER,), + package_receipt=result.handoff.to_payload(), + campaign_mode=False, + ) + entry = manifest.entries[0] + assert entry.spec is None + assert entry.unsupported is not None + assert entry.unsupported.reason == "unsupported_json_schema_construct" + assert "enum" in entry.unsupported.details def test_regeneration_with_new_field_reports_changed_and_added(self, tmp_path: Path) -> None: output_dir = tmp_path / "providers" @@ -123,7 +334,7 @@ def test_regeneration_with_new_field_reports_changed_and_added(self, tmp_path: P "polylogue.schemas.generation.workflow._build_provider_bundle", return_value=_bundle(version="v1", schema=first_schema, sample_count=5), ): - commit_provider_schema(SchemaCommitRequest(provider=_PROVIDER, output_dir=output_dir, full_corpus=True)) + commit_provider_schema(_request(output_dir)) second_schema = { "type": "object", @@ -133,9 +344,7 @@ def test_regeneration_with_new_field_reports_changed_and_added(self, tmp_path: P "polylogue.schemas.generation.workflow._build_provider_bundle", return_value=_bundle(version="v1", schema=second_schema, sample_count=9), ): - commit_result = commit_provider_schema( - SchemaCommitRequest(provider=_PROVIDER, output_dir=output_dir, full_corpus=True) - ) + commit_result = commit_provider_schema(_request(output_dir)) assert commit_result.success version_report = commit_result.versions[0] @@ -156,10 +365,8 @@ def test_identical_regeneration_reports_unchanged(self, tmp_path: Path) -> None: "polylogue.schemas.generation.workflow._build_provider_bundle", return_value=_bundle(version="v1", schema=schema, sample_count=5), ): - commit_provider_schema(SchemaCommitRequest(provider=_PROVIDER, output_dir=output_dir, full_corpus=True)) - commit_result = commit_provider_schema( - SchemaCommitRequest(provider=_PROVIDER, output_dir=output_dir, full_corpus=True) - ) + commit_provider_schema(_request(output_dir)) + commit_result = commit_provider_schema(_request(output_dir)) assert commit_result.versions[0].status == "unchanged" @@ -177,16 +384,14 @@ def test_thin_regeneration_window_cannot_narrow_committed_union(self, tmp_path: "polylogue.schemas.generation.workflow._build_provider_bundle", return_value=_bundle(version="v1", schema=wide_schema, sample_count=100), ): - commit_provider_schema(SchemaCommitRequest(provider=_PROVIDER, output_dir=output_dir, full_corpus=True)) + commit_provider_schema(_request(output_dir)) thin_schema = {"type": "object", "properties": {"timestamp": {"type": "string"}}} with patch( "polylogue.schemas.generation.workflow._build_provider_bundle", return_value=_bundle(version="v1", schema=thin_schema, sample_count=3), ): - commit_result = commit_provider_schema( - SchemaCommitRequest(provider=_PROVIDER, output_dir=output_dir, full_corpus=True) - ) + commit_result = commit_provider_schema(_request(output_dir)) assert not commit_result.narrowed assert not commit_result.versions[0].narrowed_paths @@ -200,9 +405,10 @@ def test_dry_run_does_not_touch_output_dir(self, tmp_path: Path) -> None: "polylogue.schemas.generation.workflow._build_provider_bundle", return_value=_bundle(version="v1", schema=schema, sample_count=5), ): - commit_provider_schema(SchemaCommitRequest(provider=_PROVIDER, output_dir=output_dir, full_corpus=True)) + commit_provider_schema(_request(output_dir)) catalog_before_bytes = (output_dir / _PROVIDER / "catalog.json").read_bytes() + handoff_before_bytes = (output_dir / SCHEMA_INFERENCE_HANDOFF_FILENAME).read_bytes() second_schema = { "type": "object", @@ -212,15 +418,14 @@ def test_dry_run_does_not_touch_output_dir(self, tmp_path: Path) -> None: "polylogue.schemas.generation.workflow._build_provider_bundle", return_value=_bundle(version="v1", schema=second_schema, sample_count=9), ): - commit_result = commit_provider_schema( - SchemaCommitRequest(provider=_PROVIDER, output_dir=output_dir, full_corpus=True, dry_run=True) - ) + commit_result = commit_provider_schema(_request(output_dir, dry_run=True)) assert commit_result.dry_run assert commit_result.versions[0].status == "changed" assert "session_document.would_be_added" in commit_result.versions[0].added_paths # The real committed directory was never touched. assert (output_dir / _PROVIDER / "catalog.json").read_bytes() == catalog_before_bytes + assert (output_dir / SCHEMA_INFERENCE_HANDOFF_FILENAME).read_bytes() == handoff_before_bytes on_disk = _read_element_schema(output_dir, "v1") assert "would_be_added" not in on_disk["properties"] @@ -230,7 +435,14 @@ def test_failed_generation_reports_no_success_and_no_versions(self, tmp_path: Pa # mocking needed to exercise the failure path for real. output_dir = tmp_path / "providers" commit_result = commit_provider_schema( - SchemaCommitRequest(provider="not-a-real-provider-k45pq", output_dir=output_dir, full_corpus=True) + SchemaCommitRequest( + provider="not-a-real-provider-k45pq", + output_dir=output_dir, + archive_root=output_dir.parent / "archive", + db_path=output_dir.parent / "archive" / "index.db", + full_corpus=True, + schema_inference_gate_receipt_path=_gate_receipt(output_dir), + ) ) assert not commit_result.success