diff --git a/devtools/command_catalog.py b/devtools/command_catalog.py
index ac37bd9047..45575ac8aa 100644
--- a/devtools/command_catalog.py
+++ b/devtools/command_catalog.py
@@ -30,6 +30,7 @@
     "lab projections",
     "lab smoke",
     "lab schema audit",
+    "lab schema commit",
     "lab schema compare",
     "lab schema explain",
     "lab schema generate",
@@ -1567,6 +1568,20 @@ def to_dict(self) -> dict[str, object]:
         use_when="Refresh provider schema package artifacts from archive observations outside the archive CLI.",
         examples=("devtools lab schema generate --provider chatgpt --cluster",),
     ),
+    CommandSpec(
+        "lab schema commit",
+        "verification lab",
+        "Persist a real full-corpus schema generation into committed provider packages.",
+        "devtools.schema_commit",
+        use_when=(
+            "Actually regenerate and write `polylogue/schemas/providers/<provider>/versions/...` from the live "
+            "archive -- 'lab schema generate' only ever previews and never writes committed package files."
+        ),
+        examples=(
+            "devtools lab schema commit --provider chatgpt --full-corpus --dry-run",
+            "devtools lab schema commit --provider chatgpt --full-corpus",
+        ),
+    ),
     CommandSpec(
         "lab schema promote",
         "verification lab",
diff --git a/devtools/schema_commit.py b/devtools/schema_commit.py
new file mode 100644
index 0000000000..e31f5393a1
--- /dev/null
+++ b/devtools/schema_commit.py
@@ -0,0 +1,132 @@
+"""Commit a real full-corpus schema generation into committed packages.
+
+``devtools lab schema generate`` only ever produces a preview
+``GenerationResult`` -- it never writes to ``polylogue/schemas/providers/``.
+This command is the actual persisting entry point (polylogue-k45pq):
+it calls ``generate_all_schemas`` for real via
+``polylogue.schemas.operator.commit.commit_provider_schema`` and reports
+which package versions changed, plus whether any previously-committed leaf
+type was lost or narrowed.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+from polylogue.cli.shared.schema_command_support import build_schema_privacy_config
+from polylogue.config import get_config
+from polylogue.schemas.operator.commit import commit_provider_schema
+from polylogue.schemas.operator.models import SchemaCommitRequest
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_OUTPUT_DIR = REPO_ROOT / "polylogue" / "schemas" / "providers"
+
+
+def _build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description="Commit a real full-corpus provider schema generation into committed packages."
+    )
+    parser.add_argument("--provider", required=True, help="Provider to generate and commit schema for.")
+    parser.add_argument(
+        "--output-dir",
+        type=Path,
+        default=None,
+        help=f"Committed schema package root to write into (default: {DEFAULT_OUTPUT_DIR}).",
+    )
+    parser.add_argument("--max-samples", type=int, default=None, help="Limit samples for generation.")
+    parser.add_argument(
+        "--full-corpus",
+        action="store_true",
+        default=True,
+        help="Bypass all sample caps for full-corpus schema generation (default: on).",
+    )
+    parser.add_argument(
+        "--no-full-corpus",
+        dest="full_corpus",
+        action="store_false",
+        help="Generate from a capped sample window instead of the full corpus.",
+    )
+    parser.add_argument(
+        "--privacy",
+        choices=("strict", "standard", "permissive"),
+        default=None,
+        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(
+        "--dry-run",
+        "--check",
+        dest="dry_run",
+        action="store_true",
+        help="Preview what a commit would change without writing to --output-dir.",
+    )
+    parser.add_argument("--json", action="store_true", help="Output as JSON.")
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = _build_parser()
+    args = parser.parse_args(argv)
+
+    try:
+        privacy_config = build_schema_privacy_config(
+            privacy=args.privacy,
+            privacy_config_path=args.privacy_config,
+        )
+    except ValueError as exc:
+        print(f"schema-commit: {exc}", file=sys.stderr)
+        return 1
+    output_dir = args.output_dir if args.output_dir is not None else DEFAULT_OUTPUT_DIR
+
+    result = commit_provider_schema(
+        SchemaCommitRequest(
+            provider=str(args.provider),
+            output_dir=output_dir,
+            db_path=get_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 not result.success:
+        error = result.generation.error or "Schema generation failed"
+        if args.json:
+            print(json.dumps({"provider": result.provider, "success": False, "error": error}, sort_keys=True))
+        else:
+            print(f"schema-commit: {error}", file=sys.stderr)
+        return 1
+
+    if args.json:
+        print(json.dumps(result.to_dict(), sort_keys=True, indent=2))
+    else:
+        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}")
+        for version_report in result.versions:
+            flags = []
+            if version_report.narrowed_paths:
+                flags.append(f"NARROWED({len(version_report.narrowed_paths)})")
+            if version_report.added_paths:
+                flags.append(f"added({len(version_report.added_paths)})")
+            suffix = f" [{', '.join(flags)}]" if flags else ""
+            print(
+                f"  {version_report.version}: {version_report.status} "
+                f"sample_count={version_report.sample_count}{suffix}"
+            )
+        if result.narrowed:
+            print(
+                "  WARNING: a previously-committed leaf type was lost or narrowed -- "
+                "review before trusting this commit.",
+                file=sys.stderr,
+            )
+
+    return 1 if result.narrowed else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/devtools/schema_generate.py b/devtools/schema_generate.py
index 4e17e7e313..a64e51b6e9 100644
--- a/devtools/schema_generate.py
+++ b/devtools/schema_generate.py
@@ -1,4 +1,10 @@
-"""Generate provider schema packages from the devtools surface."""
+"""Generate provider schema packages from the devtools surface.
+
+This command previews a schema generation only -- it never writes to
+``polylogue/schemas/providers/``. To actually persist a full-corpus
+generation into the committed package files, use
+``devtools lab schema commit`` (``devtools/schema_commit.py``).
+"""
 
 from __future__ import annotations
 
diff --git a/docs/devtools.md b/docs/devtools.md
index 41dd9d8953..05bff3f96d 100644
--- a/docs/devtools.md
+++ b/docs/devtools.md
@@ -65,6 +65,7 @@ They are not a proof ledger or end-user archive workflow.
 | `devtools lab projections` | Inspect the unified projection inventory that feeds runtime coverage, generated docs, and control-plane maps. |
 | `devtools lab smoke` | Run direct archive and reader smoke sets outside the archive CLI. |
 | `devtools lab schema audit` | Check committed schema package quality gates without presenting them as normal archive usage. |
+| `devtools lab schema commit` | Actually regenerate and write `polylogue/schemas/providers/<provider>/versions/...` from the live archive -- 'lab schema generate' only ever previews and never writes committed package files. |
 | `devtools lab schema compare` | Review schema package drift between committed versions in the lab surface. |
 | `devtools lab schema explain` | Inspect schema package annotations, semantic roles, and review evidence from the lab surface. |
 | `devtools lab schema generate` | Refresh provider schema package artifacts from archive observations outside the archive CLI. |
@@ -157,6 +158,7 @@ These are the commands worth remembering during normal repo work:
 | `devtools lab provider completeness` | Report provider/importer package completeness by origin and capture mode. |
 | `devtools lab pytest-witness-repetitions` | Repeat the exact optimize, WAL, and embedding seed-hang witnesses with durable receipts. |
 | `devtools lab schema audit` | Run committed provider schema package quality checks. |
+| `devtools lab schema commit` | Persist a real full-corpus schema generation into committed provider packages. |
 | `devtools lab schema compare` | Compare two committed schema package versions for a provider. |
 | `devtools lab schema explain` | Explain a committed package element schema with evidence and annotations. |
 | `devtools lab schema generate` | Generate provider schema packages and optional evidence clusters. |
diff --git a/docs/internals.md b/docs/internals.md
index 40f42c990f..d8a340dcde 100644
--- a/docs/internals.md
+++ b/docs/internals.md
@@ -636,8 +636,14 @@ Polylogue has two schema-evolution regimes, keyed by tier durability.
   rows move out of authored-user accounting while provider-role counts remain
   available.
 - Provider schemas (the parsing/validation surface, distinct from the
-  storage schema) are still regenerated fresh via
-  `devtools lab schema generate` and promoted via `devtools lab schema promote`.
+  storage schema) are still regenerated fresh. `devtools lab schema generate`
+  only previews a generation (it never writes committed package files);
+  `devtools lab schema commit --full-corpus` is the entry point that actually
+  persists a full-corpus regeneration into
+  `polylogue/schemas/providers/<provider>/versions/...`, and
+  `devtools lab schema promote` promotes a single reviewed evidence cluster
+  (from `generate --cluster`) into a registered package version -- a
+  narrower, single-version operation `commit` does not replace.
 
 For **derived tiers** (`index.db`, `embeddings.db`) this design intentionally
 rejects in-place upgrade-chain complexity (no Alembic, no forward/reverse
diff --git a/docs/plans/classifier-fingerprints.json b/docs/plans/classifier-fingerprints.json
index 8d077216c6..6c822f63fa 100644
--- a/docs/plans/classifier-fingerprints.json
+++ b/docs/plans/classifier-fingerprints.json
@@ -137,11 +137,12 @@
       }
     },
     "polylogue/sources/parsers/claude/ai_parser.py:looks_like_ai": {
-      "fingerprint": "e5886ec01e23439d5a02ce5515e80de2695c483cd0fe2f0e843c52d504debaa9",
+      "fingerprint": "d30a5251e41ab079c8465c6892bd3124bcdd634a352edfaaf3b4c3faf21d080d",
       "covered_by": {
-        "kind": "acknowledged_safe",
-        "reason": "Baseline snapshot at classifier-fingerprint gate introduction (polylogue-gucv).",
-        "ref": "polylogue-gucv"
+        "kind": "semantic_reparse_version",
+        "reason": "PR #3537 tightened looks_like_ai to require positive chat_messages evidence (role/sender+text/content), same shape as PR #3428's looks_like_code fix",
+        "ref": "polylogue-t0ta",
+        "version": 54
       }
     },
     "polylogue/sources/parsers/claude/ai_parser.py:looks_like_claude_design": {
diff --git a/docs/plans/topology-target.yaml b/docs/plans/topology-target.yaml
index 84d3e87fed..dcaadc673e 100644
--- a/docs/plans/topology-target.yaml
+++ b/docs/plans/topology-target.yaml
@@ -330,7 +330,7 @@ files:
     owner: archive-query
     reason: archive-domain query semantics
   - path: polylogue/archive/query/archive_execution.py
-    loc: 716
+    loc: 718
     target: polylogue/archive/query/archive_execution.py
     owner: archive-query
     reason: archive-domain query semantics
@@ -566,12 +566,12 @@ files:
     owner: archive-semantic
     reason: archive-domain semantics
   - path: polylogue/archive/semantic/facts.py
-    loc: 499
+    loc: 503
     target: polylogue/archive/semantic/facts.py
     owner: archive-semantic
     reason: archive-domain semantics
   - path: polylogue/archive/semantic/models.py
-    loc: 152
+    loc: 159
     target: polylogue/archive/semantic/models.py
     owner: archive-semantic
     reason: archive-domain semantics
@@ -674,7 +674,7 @@ files:
     owner: archive-session
     reason: archive-domain semantics
   - path: polylogue/archive/session/runtime.py
-    loc: 652
+    loc: 679
     target: polylogue/archive/session/runtime.py
     owner: archive-session
     reason: archive-domain semantics
@@ -1552,7 +1552,7 @@ files:
     target: polylogue/daemon/catchup_status.py
     owner: stable
   - path: polylogue/daemon/cli.py
-    loc: 2964
+    loc: 2985
     target: polylogue/daemon/cli.py
     owner: stable
   - path: polylogue/daemon/compare.py
@@ -2499,7 +2499,7 @@ files:
     target: polylogue/pipeline/services/ingest_batch/__init__.py
     owner: stable
   - path: polylogue/pipeline/services/ingest_batch/_core.py
-    loc: 1930
+    loc: 2031
     target: polylogue/pipeline/services/ingest_batch/_core.py
     owner: stable
   - path: polylogue/pipeline/services/ingest_batch/_memory.py
@@ -2507,7 +2507,7 @@ files:
     target: polylogue/pipeline/services/ingest_batch/_memory.py
     owner: stable
   - path: polylogue/pipeline/services/ingest_batch/_models.py
-    loc: 134
+    loc: 142
     target: polylogue/pipeline/services/ingest_batch/_models.py
     owner: stable
     cross_cut: { lifecycle: model }
@@ -2585,7 +2585,7 @@ files:
     target: polylogue/product/continuity_scenarios.py
     owner: stable
   - path: polylogue/product/raw_authority.py
-    loc: 195
+    loc: 205
     target: polylogue/product/raw_authority.py
     owner: stable
   - path: polylogue/product/workflows.py
@@ -2875,7 +2875,7 @@ files:
     target: polylogue/schemas/generation/support.py
     owner: stable
   - path: polylogue/schemas/generation/workflow.py
-    loc: 121
+    loc: 123
     target: polylogue/schemas/generation/workflow.py
     owner: stable
   - path: polylogue/schemas/generation/workload_profiles.py
@@ -2968,12 +2968,16 @@ files:
     loc: 265
     target: polylogue/schemas/operator/annotations.py
     owner: stable
+  - path: polylogue/schemas/operator/commit.py
+    loc: 163
+    target: polylogue/schemas/operator/commit.py
+    owner: stable
   - path: polylogue/schemas/operator/inference.py
-    loc: 361
+    loc: 372
     target: polylogue/schemas/operator/inference.py
     owner: stable
   - path: polylogue/schemas/operator/models.py
-    loc: 347
+    loc: 415
     target: polylogue/schemas/operator/models.py
     owner: stable
   - path: polylogue/schemas/operator/registry.py
@@ -2993,7 +2997,7 @@ files:
     target: polylogue/schemas/operator/verification.py
     owner: stable
   - path: polylogue/schemas/operator/workflow.py
-    loc: 55
+    loc: 59
     target: polylogue/schemas/operator/workflow.py
     owner: stable
   - path: polylogue/schemas/packages.py
@@ -3118,6 +3122,10 @@ files:
     loc: 348
     target: polylogue/schemas/tooling_registry.py
     owner: stable
+  - path: polylogue/schemas/type_narrowing.py
+    loc: 71
+    target: polylogue/schemas/type_narrowing.py
+    owner: stable
   - path: polylogue/schemas/validation/__init__.py
     loc: 1
     target: polylogue/schemas/validation/__init__.py
@@ -3913,7 +3921,7 @@ files:
     owner: storage-root
     reason: storage-root cross-cutting helper
   - path: polylogue/storage/repair.py
-    loc: 7155
+    loc: 7286
     target: polylogue/storage/repair.py
     owner: storage-root
     reason: storage-root cross-cutting helper
@@ -4249,7 +4257,7 @@ files:
     target: polylogue/storage/sqlite/archive_tiers/user_write.py
     owner: stable
   - path: polylogue/storage/sqlite/archive_tiers/write.py
-    loc: 6751
+    loc: 6764
     target: polylogue/storage/sqlite/archive_tiers/write.py
     owner: stable
   - path: polylogue/storage/sqlite/async_sqlite.py
diff --git a/docs/providers/index.md b/docs/providers/index.md
index fbe884036c..b83c0b4e8e 100644
--- a/docs/providers/index.md
+++ b/docs/providers/index.md
@@ -88,6 +88,7 @@ pipeline.
 3. **Register the detector** in `polylogue/sources/dispatch.py` at the
    appropriate priority level
 4. **Add a provider schema bundle** under `polylogue/schemas/providers/`
-   (run `devtools lab schema generate` to bootstrap)
+   (run `devtools lab schema generate` to preview, `devtools lab schema commit`
+   to actually write the committed package files)
 5. **Update schema inference** if the new provider introduces novel content
    block types or message structures
diff --git a/polylogue/schemas/generation/workflow.py b/polylogue/schemas/generation/workflow.py
index 522fcf37a7..db16191fe3 100644
--- a/polylogue/schemas/generation/workflow.py
+++ b/polylogue/schemas/generation/workflow.py
@@ -78,6 +78,7 @@ def generate_all_schemas(
     max_samples: int | None = None,
     privacy_config: SchemaPrivacyConfig | None = None,
     include_archive_workload_profile: bool = False,
+    full_corpus: bool = False,
 ) -> list[GenerationResult]:
     """Generate versioned schemas for all providers."""
     if db_path is None:
@@ -93,6 +94,7 @@ def generate_all_schemas(
             db_path=db_path,
             max_samples=max_samples,
             privacy_config=privacy_config,
+            full_corpus=full_corpus,
         )
         results.append(bundle.result)
         persist_generated_provider_bundle(output_dir, provider, bundle)
diff --git a/polylogue/schemas/operator/commit.py b/polylogue/schemas/operator/commit.py
new file mode 100644
index 0000000000..7570b15744
--- /dev/null
+++ b/polylogue/schemas/operator/commit.py
@@ -0,0 +1,163 @@
+"""Commit a real full-corpus schema generation into registered packages.
+
+``generate_provider_schema``/``infer_schema`` (what ``devtools lab schema
+generate`` calls) only ever return a ``GenerationResult`` -- they never call
+``persist_generated_provider_bundle``, the function that actually writes
+``polylogue/schemas/providers/<provider>/versions/...`` via
+``SchemaRegistry.replace_provider_packages``. That write path exists only in
+``generate_all_schemas``, which had zero CLI wiring before this module
+(polylogue-k45pq): the documented "correct entry point" for regenerating
+committed schema packages silently no-op'd on the committed files.
+
+This module is the real, persisting entry point. It wraps
+``generate_all_schemas`` (so writes go through the same monotonic-merge
+protection ``SchemaRegistry.replace_provider_packages`` already enforces --
+see ``tests/unit/schemas/test_promotion_monotonicity.py``) and adds a
+before/after report: which package versions are new, changed, or unchanged,
+and whether any previously-committed leaf type was lost or narrowed (using
+``polylogue.schemas.type_narrowing``, extracted from the same test file's
+ad hoc ``_types_by_path`` check so the production commit path and its test
+coverage share one implementation).
+
+Deliberately separate from ``promote_schema_cluster``
+(``polylogue.schemas.operator.inference``): that function promotes a single
+evidence *cluster* (from ``generate --cluster`` mode) into one registered
+package version -- a narrow, single-version operation. This module performs
+a full-corpus, potentially multi-version *replace* across every version
+``generate_all_schemas`` produces for a provider. Different shapes; neither
+supersedes the other.
+"""
+
+from __future__ import annotations
+
+import shutil
+import tempfile
+from pathlib import Path
+
+from polylogue.core.json import JSONDocument
+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.registry import SchemaRegistry
+from polylogue.schemas.runtime_registry import canonical_schema_provider
+from polylogue.schemas.type_narrowing import added_paths, narrowed_paths
+
+
+def _element_schemas_by_kind(
+    registry: SchemaRegistry, provider_token: str, version: str, element_kinds: tuple[str, ...]
+) -> dict[str, JSONDocument | None]:
+    return {
+        kind: registry.get_element_schema(provider_token, version=version, element_kind=kind) for kind in element_kinds
+    }
+
+
+def _commit_into(request: SchemaCommitRequest, output_dir: Path) -> SchemaCommitResult:
+    provider_token = str(canonical_schema_provider(request.provider))
+
+    registry_before = SchemaRegistry(storage_root=output_dir)
+    catalog_before = registry_before.load_package_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:
+        for package in catalog_before.packages:
+            element_kinds = tuple(element.element_kind for element in package.elements)
+            before_schemas[package.version] = _element_schemas_by_kind(
+                registry_before, provider_token, package.version, element_kinds
+            )
+
+    generation_results = generate_all_schemas(
+        output_dir,
+        db_path=request.db_path,
+        providers=[request.provider],
+        max_samples=request.max_samples,
+        privacy_config=privacy_config_from_payload(request.privacy_config),
+        full_corpus=request.full_corpus,
+    )
+    generation = (
+        generation_results[0]
+        if generation_results
+        else GenerationResult(
+            provider=request.provider, schema=None, sample_count=0, error="No generation result produced"
+        )
+    )
+
+    version_reports: list[SchemaVersionCommitReport] = []
+    if generation.success:
+        registry_after = SchemaRegistry(storage_root=output_dir)
+        catalog_after = registry_after.load_package_catalog(provider_token)
+        if catalog_after is not None:
+            for package in catalog_after.packages:
+                element_kinds = tuple(element.element_kind for element in package.elements)
+                after_schemas = _element_schemas_by_kind(registry_after, provider_token, package.version, element_kinds)
+                prior_schemas = before_schemas.get(package.version, {})
+
+                version_narrowed: list[str] = []
+                version_added: list[str] = []
+                for element_kind, after_schema in after_schemas.items():
+                    prior_schema = prior_schemas.get(element_kind)
+                    version_narrowed.extend(
+                        f"{element_kind}{path or ':$root'}" for path in narrowed_paths(prior_schema, after_schema)
+                    )
+                    version_added.extend(
+                        f"{element_kind}{path or ':$root'}" for path in added_paths(prior_schema, after_schema)
+                    )
+
+                if package.version not in before_versions:
+                    status = "new"
+                elif not version_narrowed and not version_added:
+                    # Structurally identical to the prior commit -- ignore
+                    # incidental bookkeeping churn (e.g. a fresh
+                    # x-polylogue-registered-at timestamp) that isn't a real
+                    # type-level change.
+                    status = "unchanged"
+                else:
+                    status = "changed"
+
+                version_reports.append(
+                    SchemaVersionCommitReport(
+                        version=package.version,
+                        status=status,
+                        sample_count=package.sample_count,
+                        narrowed_paths=tuple(version_narrowed),
+                        added_paths=tuple(version_added),
+                    )
+                )
+
+    return SchemaCommitResult(
+        provider=request.provider,
+        generation=generation,
+        versions=tuple(version_reports),
+        dry_run=request.dry_run,
+    )
+
+
+def commit_provider_schema(request: SchemaCommitRequest) -> SchemaCommitResult:
+    """Generate a provider's full schema for real and persist it, or preview it.
+
+    With ``request.dry_run`` set, the generation runs against a scratch copy
+    of the provider's current committed directory (so the real registry
+    monotonic-merge/carry-forward behavior in
+    ``SchemaRegistry.replace_provider_packages`` is exercised faithfully) and
+    the scratch copy is discarded -- ``request.output_dir`` is never touched.
+    """
+    if not request.dry_run:
+        return _commit_into(request, request.output_dir)
+
+    provider_token = str(canonical_schema_provider(request.provider))
+    with tempfile.TemporaryDirectory(prefix="polylogue-schema-commit-dry-run-") as tmp_name:
+        staging_root = Path(tmp_name) / "providers"
+        staging_root.mkdir(parents=True, exist_ok=True)
+        committed_provider_dir = request.output_dir / provider_token
+        if committed_provider_dir.exists():
+            shutil.copytree(committed_provider_dir, staging_root / provider_token)
+        result = _commit_into(request, staging_root)
+        return SchemaCommitResult(
+            provider=result.provider,
+            generation=result.generation,
+            versions=result.versions,
+            dry_run=True,
+        )
+
+
+__all__ = ["SchemaCommitRequest", "SchemaCommitResult", "SchemaVersionCommitReport", "commit_provider_schema"]
diff --git a/polylogue/schemas/operator/inference.py b/polylogue/schemas/operator/inference.py
index c38501a188..1d896543e5 100644
--- a/polylogue/schemas/operator/inference.py
+++ b/polylogue/schemas/operator/inference.py
@@ -78,6 +78,16 @@ def _string_list(value: object) -> list[str]:
     return [item for item in value if isinstance(item, str)]
 
 
+def privacy_config_from_payload(payload: Mapping[str, object] | None) -> PrivacyConfig | None:
+    """Build a typed ``PrivacyConfig`` from the CLI/request-facing JSON payload shape.
+
+    Public so other operator workflows (``polylogue.schemas.operator.commit``)
+    that accept the same ``JSONDocument | None`` privacy-config shape as
+    ``SchemaInferRequest`` can share this conversion instead of duplicating it.
+    """
+    return _privacy_config(payload)
+
+
 def _privacy_config(payload: Mapping[str, object] | None) -> PrivacyConfig | None:
     if payload is None:
         return None
@@ -357,5 +367,6 @@ def audit_schemas(request: SchemaAuditRequest) -> AuditReport:
     "list_inferred_corpus_scenarios",
     "list_inferred_corpus_specs",
     "list_schemas",
+    "privacy_config_from_payload",
     "promote_schema_cluster",
 ]
diff --git a/polylogue/schemas/operator/models.py b/polylogue/schemas/operator/models.py
index 67431a3219..36a31c3ea5 100644
--- a/polylogue/schemas/operator/models.py
+++ b/polylogue/schemas/operator/models.py
@@ -345,3 +345,71 @@ class SchemaPayloadResolveResult:
     @property
     def is_resolved(self) -> bool:
         return self.resolution is not None
+
+
+@dataclass(frozen=True)
+class SchemaCommitRequest:
+    """Request to commit (or dry-run) a real full-corpus schema generation.
+
+    ``privacy_config`` takes the same ``JSONDocument | None`` payload shape as
+    ``SchemaInferRequest.privacy_config`` (converted internally via
+    ``privacy_config_from_payload``), so CLI callers can build it the same way
+    ``devtools lab schema generate`` already does.
+    """
+
+    provider: str
+    output_dir: Path
+    db_path: Path | None = None
+    max_samples: int | None = None
+    privacy_config: JSONDocument | None = None
+    full_corpus: bool = True
+    dry_run: bool = False
+
+
+@dataclass(frozen=True)
+class SchemaVersionCommitReport:
+    """What happened to a single package version during a commit."""
+
+    version: str
+    status: str  # "new" | "changed" | "unchanged"
+    sample_count: int
+    narrowed_paths: tuple[str, ...] = field(default_factory=tuple)
+    added_paths: tuple[str, ...] = field(default_factory=tuple)
+
+    def to_dict(self) -> JSONDocument:
+        return {
+            "version": self.version,
+            "status": self.status,
+            "sample_count": self.sample_count,
+            "narrowed_paths": list(self.narrowed_paths),
+            "added_paths": list(self.added_paths),
+        }
+
+
+@dataclass(frozen=True)
+class SchemaCommitResult:
+    """Before/after report for a real (or dry-run) schema commit."""
+
+    provider: str
+    generation: GenerationResult
+    versions: tuple[SchemaVersionCommitReport, ...]
+    dry_run: bool
+
+    @property
+    def success(self) -> bool:
+        return self.generation.success
+
+    @property
+    def narrowed(self) -> bool:
+        """True if any previously-committed leaf type was lost or narrowed."""
+        return any(report.narrowed_paths for report in self.versions)
+
+    def to_dict(self) -> JSONDocument:
+        return {
+            "provider": self.provider,
+            "success": self.success,
+            "narrowed": self.narrowed,
+            "dry_run": self.dry_run,
+            "sample_count": self.generation.sample_count,
+            "versions": [report.to_dict() for report in self.versions],
+        }
diff --git a/polylogue/schemas/operator/workflow.py b/polylogue/schemas/operator/workflow.py
index 76823fee6c..62d44c40e0 100644
--- a/polylogue/schemas/operator/workflow.py
+++ b/polylogue/schemas/operator/workflow.py
@@ -3,6 +3,9 @@
 from __future__ import annotations
 
 from polylogue.schemas.operator.annotations import collect_annotation_summary
+from polylogue.schemas.operator.commit import (
+    commit_provider_schema as _commit_provider_schema,
+)
 from polylogue.schemas.operator.inference import (
     audit_schemas as _audit_schemas,
 )
@@ -42,6 +45,7 @@
 list_schemas = _list_schemas
 compare_schema_versions = _compare_schema_versions
 promote_schema_cluster = _promote_schema_cluster
+commit_provider_schema = _commit_provider_schema
 explain_schema = _explain_schema
 resolve_schema_payload = _resolve_schema_payload
 audit_schemas = _audit_schemas
diff --git a/polylogue/schemas/type_narrowing.py b/polylogue/schemas/type_narrowing.py
new file mode 100644
index 0000000000..6cd0e6b991
--- /dev/null
+++ b/polylogue/schemas/type_narrowing.py
@@ -0,0 +1,84 @@
+"""Structural type-narrowing detection shared by promotion-safety checks.
+
+Extracted from ``tests/unit/schemas/test_promotion_monotonicity.py`` so the
+same "did any leaf type shrink versus what was previously committed?" check
+that guards ``SchemaRegistry.replace_provider_packages`` in tests can also run
+as part of a real commit/promotion report (``polylogue.schemas.operator.commit``),
+instead of only ever existing as test-only logic.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from polylogue.core.json import JSONDocument
+
+_STRUCTURAL_KEYS = ("properties", "items", "additionalProperties", "anyOf", "oneOf", "allOf")
+
+
+def types_by_path(schema: Any, path: str = "") -> dict[str, frozenset[str]]:
+    """Every typed node in a schema, keyed by structural path.
+
+    Two schemas produced for the "same" node are comparable by taking
+    ``types_by_path(before)`` and ``types_by_path(after)`` and checking that
+    every path's type set in ``before`` is a subset of the corresponding set
+    in ``after`` -- see ``narrowed_paths`` for the comparison helper.
+    """
+    found: dict[str, frozenset[str]] = {}
+
+    def merge(other: dict[str, frozenset[str]]) -> None:
+        # A union (anyOf/oneOf) or a list of schemas can contribute distinct
+        # type sets for the SAME path from different branches -- e.g. an
+        # anyOf of {"type": "string"} and {"type": "number"} both project to
+        # this node's own path. A plain dict.update() would let the last
+        # branch silently overwrite an earlier branch's types at that path,
+        # which could hide a real narrowing (the earlier-committed union
+        # member's type would vanish from `found` even though it's still
+        # legitimately part of this schema). Union the sets instead.
+        for other_path, other_types in other.items():
+            found[other_path] = found.get(other_path, frozenset()) | other_types
+
+    if isinstance(schema, dict):
+        declared = schema.get("type")
+        if isinstance(declared, str):
+            found[path] = frozenset({declared})
+        elif isinstance(declared, list):
+            found[path] = frozenset(item for item in declared if isinstance(item, str))
+        for key, value in schema.items():
+            if not isinstance(value, (dict, list)):
+                continue
+            structural = key in _STRUCTURAL_KEYS
+            merge(types_by_path(value, path if structural else f"{path}.{key}"))
+    elif isinstance(schema, list):
+        for entry in schema:
+            merge(types_by_path(entry, path))
+    return found
+
+
+def narrowed_paths(before: JSONDocument | None, after: JSONDocument | None) -> tuple[str, ...]:
+    """Paths whose type set in ``before`` is not a subset of ``after``'s.
+
+    An empty ``before`` (nothing previously committed) never narrows anything.
+    A ``None`` ``after`` narrows every path ``before`` declared a type for.
+    """
+    if before is None:
+        return ()
+    before_types = types_by_path(before)
+    after_types = types_by_path(after) if after is not None else {}
+    return tuple(
+        path
+        for path, before_type_set in before_types.items()
+        if not before_type_set <= after_types.get(path, frozenset())
+    )
+
+
+def added_paths(before: JSONDocument | None, after: JSONDocument | None) -> tuple[str, ...]:
+    """Paths present in ``after`` with no corresponding typed node in ``before``."""
+    if after is None:
+        return ()
+    after_types = types_by_path(after)
+    before_types = types_by_path(before) if before is not None else {}
+    return tuple(path for path in after_types if path not in before_types)
+
+
+__all__ = ["added_paths", "narrowed_paths", "types_by_path"]
diff --git a/polylogue/storage/sqlite/archive_tiers/index.py b/polylogue/storage/sqlite/archive_tiers/index.py
index 0af10d214d..ffdc72ee17 100644
--- a/polylogue/storage/sqlite/archive_tiers/index.py
+++ b/polylogue/storage/sqlite/archive_tiers/index.py
@@ -220,7 +220,20 @@
 # raw evidence recovers the corrected identity split. `polylogue ops reset
 # --index && polylogued run` is required; deliberately NOT executed by this
 # declaration.
-INDEX_SCHEMA_VERSION = 53
+INDEX_SCHEMA_VERSION = 54
+
+# PR #3537: tightened `claude.looks_like_ai` (polylogue/sources/parsers/
+# claude/ai_parser.py) to require at least one `chat_messages` entry
+# carrying a role/sender field plus a text/content field, instead of
+# accepting a bare (even empty) `chat_messages` key -- the same
+# positive-evidence-required shape as PR #3428's `looks_like_code` fix.
+# This moves the classifier's decision boundary for identical input
+# bytes: some payloads previously admitted as claude-ai-export sessions
+# are now refused. SEMANTIC_REPARSE, not a free fast-forward: only
+# re-parsing already-acquired raw evidence applies the corrected
+# admission decision to existing rows. `polylogue ops reset --index &&
+# polylogued run` is required; deliberately NOT executed by this
+# declaration.
 
 # polylogue-v6i3: shared WHEN-clause fragment gating the blocks_command_trigram
 # trigger BODIES on the same dedicated bulk-build guard row messages_fts's
diff --git a/polylogue/storage/sqlite/lifecycle.py b/polylogue/storage/sqlite/lifecycle.py
index 94331348a6..725b85d995 100644
--- a/polylogue/storage/sqlite/lifecycle.py
+++ b/polylogue/storage/sqlite/lifecycle.py
@@ -660,6 +660,19 @@ class IndexDeltaDeclarationReport(TypedDict):
         # this declaration.
         classes=(DerivedDeltaClass.SEMANTIC_REPARSE,),
     ),
+    IndexDeltaDeclaration(
+        version=54,
+        # PR #3537: tightened claude.looks_like_ai (sources/parsers/claude/
+        # ai_parser.py) to require a genuine chat_messages entry (role/sender
+        # + text/content) rather than a bare chat_messages key -- see
+        # INDEX_SCHEMA_VERSION's v54 comment (archive_tiers/index.py) for the
+        # full writeup. Moves the classifier's admission decision for
+        # already-acquired raw evidence, so it requires re-parsing, not a
+        # clone-safe fast-forward. SEMANTIC_REPARSE routes through `polylogue
+        # ops reset --index && polylogued run`, deliberately NOT executed by
+        # this declaration.
+        classes=(DerivedDeltaClass.SEMANTIC_REPARSE,),
+    ),
 )
 
 
diff --git a/tests/unit/devtools/test_schema_commit_command.py b/tests/unit/devtools/test_schema_commit_command.py
new file mode 100644
index 0000000000..892fa3473e
--- /dev/null
+++ b/tests/unit/devtools/test_schema_commit_command.py
@@ -0,0 +1,157 @@
+"""``devtools lab schema commit`` -- the real, persisting schema-commit CLI.
+
+Unit-level plumbing tests: the request built from CLI args, and rendering of
+the ``SchemaCommitResult``. The commit path's actual file-writing behavior is
+covered end-to-end in ``tests/unit/schemas/test_operator_commit.py``.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+
+import pytest
+
+from devtools import schema_commit
+from polylogue.schemas.generation.models import GenerationResult
+from polylogue.schemas.operator.models import SchemaCommitRequest, SchemaCommitResult, SchemaVersionCommitReport
+
+
+@dataclass(frozen=True)
+class _ConfigStub:
+    db_path: Path
+
+
+def test_schema_commit_forwards_request_and_defaults_output_dir(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    captured: list[SchemaCommitRequest] = []
+
+    def fake_get_config() -> _ConfigStub:
+        return _ConfigStub(db_path=tmp_path / "archive.db")
+
+    def fake_commit(request: SchemaCommitRequest) -> SchemaCommitResult:
+        captured.append(request)
+        return SchemaCommitResult(
+            provider=request.provider,
+            generation=GenerationResult(provider=request.provider, schema={"type": "object"}, sample_count=3),
+            versions=(SchemaVersionCommitReport(version="v1", status="new", sample_count=3),),
+            dry_run=request.dry_run,
+        )
+
+    monkeypatch.setattr(schema_commit, "get_config", fake_get_config)
+    monkeypatch.setattr(schema_commit, "commit_provider_schema", fake_commit)
+
+    assert schema_commit.main(["--provider", "chatgpt"]) == 0
+
+    assert len(captured) == 1
+    request = captured[0]
+    assert request.provider == "chatgpt"
+    assert request.output_dir == schema_commit.DEFAULT_OUTPUT_DIR
+    assert request.db_path == tmp_path / "archive.db"
+    assert request.full_corpus is True
+    assert request.dry_run is False
+
+
+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"))
+
+    def fake_commit(request: SchemaCommitRequest) -> SchemaCommitResult:
+        captured.append(request)
+        return SchemaCommitResult(
+            provider=request.provider,
+            generation=GenerationResult(provider=request.provider, schema={"type": "object"}, sample_count=1),
+            versions=(SchemaVersionCommitReport(version="v1", status="unchanged", sample_count=1),),
+            dry_run=request.dry_run,
+        )
+
+    monkeypatch.setattr(schema_commit, "commit_provider_schema", fake_commit)
+
+    custom_output = tmp_path / "custom-providers"
+    assert (
+        schema_commit.main(
+            ["--provider", "chatgpt", "--output-dir", str(custom_output), "--dry-run", "--no-full-corpus"]
+        )
+        == 0
+    )
+
+    assert captured[0].output_dir == custom_output
+    assert captured[0].dry_run is True
+    assert captured[0].full_corpus is False
+
+
+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,
+        "commit_provider_schema",
+        lambda request: SchemaCommitResult(
+            provider=request.provider,
+            generation=GenerationResult(provider=request.provider, schema={"type": "object"}, sample_count=42),
+            versions=(
+                SchemaVersionCommitReport(
+                    version="v2", status="changed", sample_count=42, added_paths=("session_document.new",)
+                ),
+            ),
+            dry_run=False,
+        ),
+    )
+
+    assert schema_commit.main(["--provider", "chatgpt", "--json"]) == 0
+
+    payload = json.loads(capsys.readouterr().out)
+    assert payload["provider"] == "chatgpt"
+    assert payload["success"] is True
+    assert payload["narrowed"] is False
+    assert payload["sample_count"] == 42
+    assert payload["versions"][0]["status"] == "changed"
+    assert payload["versions"][0]["added_paths"] == ["session_document.new"]
+
+
+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,
+        "commit_provider_schema",
+        lambda request: SchemaCommitResult(
+            provider=request.provider,
+            generation=GenerationResult(provider=request.provider, schema=None, sample_count=0, error="No samples"),
+            versions=(),
+            dry_run=False,
+        ),
+    )
+
+    assert schema_commit.main(["--provider", "broken-provider", "--json"]) == 1
+    payload = json.loads(capsys.readouterr().out)
+    assert payload["success"] is False
+    assert payload["error"] == "No samples"
+
+
+def test_schema_commit_exits_nonzero_when_narrowed(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    """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,
+        "commit_provider_schema",
+        lambda request: SchemaCommitResult(
+            provider=request.provider,
+            generation=GenerationResult(provider=request.provider, schema={"type": "object"}, sample_count=3),
+            versions=(
+                SchemaVersionCommitReport(
+                    version="v1", status="changed", sample_count=3, narrowed_paths=("session_document.timestamp",)
+                ),
+            ),
+            dry_run=False,
+        ),
+    )
+
+    assert schema_commit.main(["--provider", "chatgpt"]) == 1
diff --git a/tests/unit/schemas/test_operator_commit.py b/tests/unit/schemas/test_operator_commit.py
new file mode 100644
index 0000000000..cff34a4a80
--- /dev/null
+++ b/tests/unit/schemas/test_operator_commit.py
@@ -0,0 +1,238 @@
+"""``commit_provider_schema`` -- the real, persisting full-corpus commit path.
+
+``devtools lab schema generate`` (``generate_provider_schema``/``infer_schema``)
+never writes to ``polylogue/schemas/providers/`` -- only
+``generate_all_schemas`` does, and it had zero CLI wiring before
+``polylogue.schemas.operator.commit`` (polylogue-k45pq). These tests prove the
+new command actually changes files on disk, not merely that a function was
+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.
+
+Only ``_build_provider_bundle`` (the sample-observation step) is mocked, the
+same seam ``tests/unit/core/test_schema_generation.py`` uses for
+``generate_all_schemas`` -- the persistence path under test
+(``generate_all_schemas`` -> ``persist_generated_provider_bundle`` ->
+``SchemaRegistry.replace_provider_packages``) runs for real.
+"""
+
+from __future__ import annotations
+
+import gzip
+import json
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any, cast
+from unittest.mock import patch
+
+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.packages import SchemaElementManifest, SchemaPackageCatalog, SchemaVersionPackage
+from polylogue.schemas.tooling_models import ClusterManifest
+
+_PROVIDER = "commit-fixture-k45pq"
+
+
+def _bundle(
+    *,
+    version: str,
+    schema: dict[str, Any],
+    sample_count: int,
+    element_kind: str = "session_document",
+) -> SimpleNamespace:
+    package = SchemaVersionPackage(
+        provider=_PROVIDER,
+        version=version,
+        anchor_kind=element_kind,
+        default_element_kind=element_kind,
+        first_seen="2026-08-01T00:00:00+00:00",
+        last_seen="2026-08-01T00:00:00+00:00",
+        bundle_scope_count=1,
+        sample_count=sample_count,
+        elements=[
+            SchemaElementManifest(
+                element_kind=element_kind,
+                schema_file=f"{element_kind}.schema.json.gz",
+                sample_count=sample_count,
+                artifact_count=sample_count,
+            )
+        ],
+    )
+    result = GenerationResult(
+        provider=_PROVIDER,
+        sample_count=sample_count,
+        schema=schema,
+        error=None,
+        versions=[version],
+        default_version=version,
+        package_count=1,
+        cluster_count=1,
+    )
+    return SimpleNamespace(
+        result=result,
+        catalog=SchemaPackageCatalog(
+            provider=_PROVIDER,
+            packages=[package],
+            latest_version=version,
+            default_version=version,
+            recommended_version=version,
+        ),
+        package_schemas={version: {element_kind: schema}},
+        manifest=ClusterManifest(provider=_PROVIDER, clusters=[], artifact_counts={}),
+    )
+
+
+def _read_element_schema(output_dir: Path, version: str, element_kind: str = "session_document") -> dict[str, Any]:
+    path = output_dir / _PROVIDER / "versions" / version / "elements" / f"{element_kind}.schema.json.gz"
+    with gzip.open(path, "rt", encoding="utf-8") as handle:
+        return cast("dict[str, Any]", json.load(handle))
+
+
+class TestCommitProviderSchemaWritesRealFiles:
+    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)
+            )
+
+        assert commit_result.success
+        assert not commit_result.dry_run
+        # Real files on disk, not merely "a function returned a result".
+        assert (output_dir / _PROVIDER / "catalog.json").exists()
+        on_disk = _read_element_schema(output_dir, "v1")
+        assert on_disk["properties"]["id"]["type"] == "string"
+
+        assert len(commit_result.versions) == 1
+        version_report = commit_result.versions[0]
+        assert version_report.version == "v1"
+        assert version_report.status == "new"
+        assert version_report.sample_count == 5
+        assert not version_report.narrowed_paths
+        assert "session_document.id" in version_report.added_paths
+
+    def test_regeneration_with_new_field_reports_changed_and_added(self, tmp_path: Path) -> None:
+        output_dir = tmp_path / "providers"
+        first_schema = {"type": "object", "properties": {"id": {"type": "string"}}}
+        with patch(
+            "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))
+
+        second_schema = {
+            "type": "object",
+            "properties": {"id": {"type": "string"}, "newly_observed": {"type": "boolean"}},
+        }
+        with patch(
+            "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)
+            )
+
+        assert commit_result.success
+        version_report = commit_result.versions[0]
+        assert version_report.status == "changed"
+        assert version_report.sample_count == 9
+        assert not version_report.narrowed_paths
+        assert "session_document.newly_observed" in version_report.added_paths
+
+        # The field really is on disk, not just in the in-memory report.
+        on_disk = _read_element_schema(output_dir, "v1")
+        assert on_disk["properties"]["newly_observed"]["type"] == "boolean"
+        assert on_disk["properties"]["id"]["type"] == "string"
+
+    def test_identical_regeneration_reports_unchanged(self, tmp_path: Path) -> None:
+        output_dir = tmp_path / "providers"
+        schema = {"type": "object", "properties": {"id": {"type": "string"}}}
+        with patch(
+            "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)
+            )
+
+        assert commit_result.versions[0].status == "unchanged"
+
+    def test_thin_regeneration_window_cannot_narrow_committed_union(self, tmp_path: Path) -> None:
+        """End-to-end proof that the real commit path inherits
+        ``SchemaRegistry.replace_provider_packages``'s monotonic-merge safety
+        net (the ov5r/polylogue-46kg incident class): a second, thinner
+        generation window that would -- if written directly -- narrow a
+        previously-observed type union instead leaves the union intact on
+        disk, and the commit report correctly finds zero narrowed paths.
+        """
+        output_dir = tmp_path / "providers"
+        wide_schema = {"type": "object", "properties": {"timestamp": {"type": ["string", "number"]}}}
+        with patch(
+            "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))
+
+        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)
+            )
+
+        assert not commit_result.narrowed
+        assert not commit_result.versions[0].narrowed_paths
+        on_disk = _read_element_schema(output_dir, "v1")
+        assert set(on_disk["properties"]["timestamp"]["type"]) == {"string", "number"}
+
+    def test_dry_run_does_not_touch_output_dir(self, tmp_path: Path) -> None:
+        output_dir = tmp_path / "providers"
+        schema = {"type": "object", "properties": {"id": {"type": "string"}}}
+        with patch(
+            "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))
+
+        catalog_before_bytes = (output_dir / _PROVIDER / "catalog.json").read_bytes()
+
+        second_schema = {
+            "type": "object",
+            "properties": {"id": {"type": "string"}, "would_be_added": {"type": "boolean"}},
+        }
+        with patch(
+            "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)
+            )
+
+        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
+        on_disk = _read_element_schema(output_dir, "v1")
+        assert "would_be_added" not in on_disk["properties"]
+
+    def test_failed_generation_reports_no_success_and_no_versions(self, tmp_path: Path) -> None:
+        # An unrecognized provider token fails inside `_build_provider_bundle`
+        # itself (unknown-provider guard) with no DB access required -- no
+        # 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)
+        )
+
+        assert not commit_result.success
+        assert not commit_result.versions
+        assert not (output_dir / "not-a-real-provider-k45pq").exists()
diff --git a/tests/unit/schemas/test_promotion_monotonicity.py b/tests/unit/schemas/test_promotion_monotonicity.py
index caac007f83..7aa86a5a88 100644
--- a/tests/unit/schemas/test_promotion_monotonicity.py
+++ b/tests/unit/schemas/test_promotion_monotonicity.py
@@ -23,26 +23,7 @@
 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
-
-
-def _types_by_path(schema: Any, path: str = "") -> dict[str, frozenset[str]]:
-    """Every typed node in a schema, keyed by structural path."""
-    found: dict[str, frozenset[str]] = {}
-    if isinstance(schema, dict):
-        declared = schema.get("type")
-        if isinstance(declared, str):
-            found[path] = frozenset({declared})
-        elif isinstance(declared, list):
-            found[path] = frozenset(item for item in declared if isinstance(item, str))
-        for key, value in schema.items():
-            if not isinstance(value, (dict, list)):
-                continue
-            structural = key in ("properties", "items", "additionalProperties", "anyOf", "oneOf", "allOf")
-            found.update(_types_by_path(value, path if structural else f"{path}.{key}"))
-    elif isinstance(schema, list):
-        for entry in schema:
-            found.update(_types_by_path(entry, path))
-    return found
+from polylogue.schemas.type_narrowing import types_by_path as _types_by_path
 
 
 def test_merging_a_thin_window_cannot_narrow_a_union() -> None:
@@ -62,6 +43,21 @@ def test_merging_a_thin_window_cannot_narrow_a_union() -> None:
     assert set(timestamp["type"]) == {"string", "number"}
 
 
+def test_types_by_path_unions_anyof_branches_at_the_same_path() -> None:
+    """CodeRabbit finding on PR #3538: found.update() overwrote an earlier
+    anyOf branch's types at a shared path instead of merging them, so a
+    later schema keeping only the last branch's type wouldn't register as
+    narrowed even though the earlier branch's type was really lost."""
+    schema: JSONDocument = {
+        "type": "object",
+        "properties": {
+            "value": {"anyOf": [{"type": "string"}, {"type": "number"}]},
+        },
+    }
+    types = _types_by_path(schema)
+    assert types[".value"] == frozenset({"string", "number"})
+
+
 def test_merging_cannot_drop_a_field_the_package_already_carried() -> None:
     """173 fields vanished in the real incident; absence from a window is not evidence of removal."""
     promoted: JSONDocument = {