fix(schemas): merge full-corpus regen against committed schema

polylogue/schemas/runtime_registry.py

@@ -540,6 +540,112 @@ def _preflight_package_write(
                     f"Package {package.provider}/{package.version} has a non-JSON workload profile"
                 ) from exc
 
+    def _load_local_element_schema(
+        self,
+        provider_token: str,
+        *,
+        version: str,
+        element_kind: str | None,
+    ) -> PublicSchemaDocument | None:
+        """Load an element schema strictly from this registry's own storage root.
+
+        Mirrors ``_load_local_catalog``'s no-bundled-fallback guarantee: a
+        caller merging against "the committed prior schema"
+        (``replace_provider_packages``) must not silently pull in an
+        unrelated version from the bundled ``SCHEMA_DIR`` tree when
+        ``storage_root`` is isolated (e.g. a test's ``tmp_path``, or a
+        ``devtools schema-generate`` run pointed at a scratch output dir).
+        """
+        catalog = self._load_local_catalog(provider_token)
+        if catalog is None:
+            return None
+        resolved_version = _resolved_package_version(catalog, version)
+        if resolved_version is None:
+            return None
+        package = catalog.package(resolved_version)
+        if package is None:
+            return None
+        element = package.element(element_kind)
+        if element is None or element.schema_file is None:
+            return None
+        path = self._provider_dir(provider_token) / "versions" / package.version / "elements" / element.schema_file
+        if not path.exists():
+            return None
+        return _read_gzip_json_dict(path)
+
+    def _existing_provider_element_schemas(self, provider_token: str) -> dict[str, PublicSchemaDocument]:
+        """Collect every element schema already committed for this provider, by element_kind.
+
+        Collected across *all* existing versions (not just ``default``) so a
+        full-corpus regeneration guarded by ``replace_provider_packages`` can
+        never narrow a leaf path or type union that any previously committed
+        version observed -- the same monotonic-merge guarantee
+        ``promote_cluster``/``_merge_with_promoted_schema`` already provide
+        for the other promotion surface (``tooling_registry.py``).
+        """
+        catalog = self._load_local_catalog(provider_token)
+        if catalog is None:
+            return {}
+        from polylogue.schemas.generation.dynamic_keys import merge_observed_structure_schemas
+
+        merged_by_kind: dict[str, PublicSchemaDocument] = {}
+        for package in catalog.packages:
+            for element in package.elements:
+                if element.schema_file is None:
+                    continue
+                existing = self._load_local_element_schema(
+                    provider_token,
+                    version=package.version,
+                    element_kind=element.element_kind,
+                )
+                if existing is None:
+                    continue
+                prior = merged_by_kind.get(element.element_kind)
+                merged_by_kind[element.element_kind] = (
+                    json_document(merge_observed_structure_schemas([json_document(prior), existing]))
+                    if prior is not None
+                    else existing
+                )
+        return merged_by_kind
+
+    @staticmethod
+    def _merge_element_schema_with_existing(
+        existing: PublicSchemaDocument | None,
+        candidate: PublicSchemaDocument,
+    ) -> PublicSchemaDocument:
+        """Union ``candidate`` with a previously committed element schema.
+
+        Same monotonic-merge contract as
+        ``SchemaRegistryToolingMixin._merge_with_promoted_schema``: a
+        provider package records what the provider has been *observed* to
+        emit across every corpus this registry has ever scanned, so a fresh
+        full-corpus regeneration may only ever broaden it, never replace it
+        outright. Without this, a thinner or differently-shaped sample
+        window (fewer sessions, a narrower clustering pass, a corpus subset)
+        silently drops fields and narrows type unions that an earlier run
+        legitimately observed -- measured on 2026-08-01: claude-code lost
+        722 of 944 typed leaf paths, codex narrowed ``timestamp`` from
+        ``["number", "string"]`` back to ``["string"]``, in a single
+        `devtools schema-generate` run with no merge against history.
+        """
+        if existing is None:
+            return candidate
+        from polylogue.schemas.generation.dynamic_keys import merge_observed_structure_schemas
+
+        merged = json_document(merge_observed_structure_schemas([json_document(existing), candidate]))
+        # Structural merge owns type/properties/items only; provenance and
+        # the x-polylogue-* annotation overlay come from the candidate (this
+        # run's fresh observation), falling back to the existing package so
+        # the merge never drops an annotation the candidate simply didn't
+        # recompute.
+        for key, value in existing.items():
+            if key.startswith("x-polylogue-") and key not in candidate:
+                merged[key] = value
+        for key, value in candidate.items():
+            if key.startswith("x-polylogue-") or key in ("$schema", "title"):
+                merged[key] = value
+        return merged
+
     def replace_provider_packages(
         self,
         provider: str,
@@ -549,20 +655,27 @@ def replace_provider_packages(
         package_workload_profiles: Mapping[str, Mapping[str, object]] | None = None,
     ) -> None:
         provider_token = _provider_token(provider)
+        existing_element_schemas = self._existing_provider_element_schemas(provider_token)
         prepared_packages: list[tuple[SchemaVersionPackage, ElementSchemaMap, Mapping[str, object] | None]] = []
         for package in catalog.packages:
             element_schemas = package_schemas.get(package.version)
             if element_schemas is None:
                 raise ValueError(f"Package {provider_token}/{package.version} has no schema mapping")
+            merged_element_schemas: ElementSchemaMap = {
+                element_kind: self._merge_element_schema_with_existing(
+                    existing_element_schemas.get(element_kind), schema
+                )
+                for element_kind, schema in element_schemas.items()
+            }
             workload_profile = (
                 package_workload_profiles.get(package.version) if package_workload_profiles is not None else None
             )
             self._preflight_package_write(
                 package,
-                element_schemas=element_schemas,
+                element_schemas=merged_element_schemas,
                 workload_profile=workload_profile,
             )
-            prepared_packages.append((package, element_schemas, workload_profile))
+            prepared_packages.append((package, merged_element_schemas, workload_profile))
 
         provider_dir = self._provider_dir(provider_token)
         provider_dir.mkdir(parents=True, exist_ok=True)

tests/unit/schemas/test_promotion_monotonicity.py

@@ -21,6 +21,7 @@
 
 from polylogue.core.json import JSONDocument
 from polylogue.schemas.generation.dynamic_keys import merge_observed_structure_schemas
+from polylogue.schemas.packages import SchemaElementManifest, SchemaPackageCatalog, SchemaVersionPackage
 from polylogue.schemas.registry import SchemaRegistry
 
 
@@ -162,3 +163,139 @@ def test_observed_artifact_count_never_decreases(self, tmp_registry: SchemaRegis
         observed_count = schema["x-polylogue-observed-artifact-count"]
         assert isinstance(observed_count, int)
         assert observed_count >= 10_000
+
+
+def _single_package_catalog(provider: str, version: str, element_kind: str) -> SchemaPackageCatalog:
+    return SchemaPackageCatalog(
+        provider=provider,
+        packages=[
+            SchemaVersionPackage(
+                provider=provider,
+                version=version,
+                anchor_kind=element_kind,
+                default_element_kind=element_kind,
+                first_seen="2026-08-01T00:00:00Z",
+                last_seen="2026-08-01T00:00:00Z",
+                bundle_scope_count=0,
+                sample_count=1,
+                elements=[
+                    SchemaElementManifest(
+                        element_kind=element_kind,
+                        schema_file=f"{element_kind}.schema.json.gz",
+                        sample_count=1,
+                        artifact_count=1,
+                    )
+                ],
+            )
+        ],
+        default_version=version,
+        latest_version=version,
+        recommended_version=version,
+    )
+
+
+class TestReplaceProviderPackagesMonotonicity:
+    """Guards ``SchemaRegistry.replace_provider_packages`` -- the code path every
+    full-corpus ``devtools schema-generate`` run writes through
+    (``persist_generated_provider_bundle`` -> ``replace_provider_packages`` in
+    ``polylogue/schemas/generation/workflow.py``). Reproduces, at unit scale,
+    the 2026-08-01 measured incident (polylogue-ov5r): a real full-corpus
+    ``devtools schema-generate`` run against the live archive lost 722 of 944
+    typed leaf paths for claude-code and narrowed codex's ``timestamp`` union
+    from ``["number", "string"]`` back to ``["string"]`` -- because
+    ``replace_provider_packages`` deleted the provider's entire ``versions/``
+    tree and rewrote it from the fresh generation with no merge against the
+    committed prior schema.
+
+    Anti-vacuity: deleting the
+    ``_merge_element_schema_with_existing``/``_existing_provider_element_schemas``
+    call in ``SchemaRegistry.replace_provider_packages``
+    (``polylogue/schemas/runtime_registry.py``), or reverting
+    ``replace_provider_packages`` to write ``package_schemas`` directly instead
+    of the merged map, makes every test below fail: the second
+    ``replace_provider_packages`` call would then simply overwrite the first
+    package instead of unioning into it.
+    """
+
+    def test_regen_observing_a_subset_of_known_fields_does_not_lose_them(self, tmp_registry: SchemaRegistry) -> None:
+        wide_schema: JSONDocument = {
+            "type": "object",
+            "properties": {
+                "kept": {"type": "string"},
+                "dropped_in_thin_regen": {"type": "integer"},
+            },
+        }
+        tmp_registry.replace_provider_packages(
+            "regen-subset",
+            _single_package_catalog("regen-subset", "v1", "session_record_stream"),
+            {"v1": {"session_record_stream": wide_schema}},
+        )
+
+        # A subsequent full-corpus regeneration observes only a subset of the
+        # previously known fields -- e.g. a differently-shaped sample window,
+        # exactly the scenario that dropped 722/944 claude-code leaf paths.
+        thin_schema: JSONDocument = {"type": "object", "properties": {"kept": {"type": "string"}}}
+        tmp_registry.replace_provider_packages(
+            "regen-subset",
+            _single_package_catalog("regen-subset", "v1", "session_record_stream"),
+            {"v1": {"session_record_stream": thin_schema}},
+        )
+
+        merged = tmp_registry.get_schema("regen-subset", version="v1")
+        assert merged is not None
+        properties = cast("dict[str, Any]", merged["properties"])
+        assert "dropped_in_thin_regen" in properties
+        assert properties["dropped_in_thin_regen"]["type"] == "integer"
+
+    def test_regen_observing_genuinely_new_fields_gains_them(self, tmp_registry: SchemaRegistry) -> None:
+        old_schema: JSONDocument = {"type": "object", "properties": {"kept": {"type": "string"}}}
+        tmp_registry.replace_provider_packages(
+            "regen-new-field",
+            _single_package_catalog("regen-new-field", "v1", "session_record_stream"),
+            {"v1": {"session_record_stream": old_schema}},
+        )
+
+        new_schema: JSONDocument = {
+            "type": "object",
+            "properties": {"kept": {"type": "string"}, "newly_observed": {"type": "boolean"}},
+        }
+        tmp_registry.replace_provider_packages(
+            "regen-new-field",
+            _single_package_catalog("regen-new-field", "v1", "session_record_stream"),
+            {"v1": {"session_record_stream": new_schema}},
+        )
+
+        merged = tmp_registry.get_schema("regen-new-field", version="v1")
+        assert merged is not None
+        properties = cast("dict[str, Any]", merged["properties"])
+        assert "newly_observed" in properties
+        assert properties["newly_observed"]["type"] == "boolean"
+
+    def test_regen_type_unions_only_grow_never_narrow(self, tmp_registry: SchemaRegistry) -> None:
+        """The exact codex incident, reproduced against replace_provider_packages directly."""
+        wide_schema: JSONDocument = {
+            "type": "object",
+            "properties": {"timestamp": {"type": ["string", "number"]}},
+        }
+        tmp_registry.replace_provider_packages(
+            "regen-union",
+            _single_package_catalog("regen-union", "v1", "session_record_stream"),
+            {"v1": {"session_record_stream": wide_schema}},
+        )
+
+        narrow_schema: JSONDocument = {
+            "type": "object",
+            "properties": {"timestamp": {"type": "string"}},
+        }
+        tmp_registry.replace_provider_packages(
+            "regen-union",
+            _single_package_catalog("regen-union", "v1", "session_record_stream"),
+            {"v1": {"session_record_stream": narrow_schema}},
+        )
+
+        merged = tmp_registry.get_schema("regen-union", version="v1")
+        assert merged is not None
+        timestamp = cast("dict[str, Any]", cast("dict[str, Any]", merged["properties"])["timestamp"])
+        declared_type = timestamp["type"]
+        observed = set(declared_type) if isinstance(declared_type, list) else {declared_type}
+        assert observed == {"string", "number"}