feat: gate infrastructure changes on provider inventory

docs/operations/INSTALLATION_AND_DEPLOYMENT_ARCHITECTURE.md

@@ -100,6 +100,7 @@ The private installation directory contains:
 | `plan.json` | Latest desired-state diff and readiness findings |
 | `receipt.json` | Last successfully applied immutable identities |
 | `infrastructure-capabilities.json` | Deterministic non-secret capability states, endpoint metadata, secret references, consumers, and resumable post-install tasks |
+| `infrastructure-dependency-inventory.json` | Owner-only, short-lived Ops evidence of actual module-owned configuration and data that depend on infrastructure capabilities |
 | `distribution-manifest.json` | Canonical signed runtime/image selection adopted by the installer |
 | `distribution-keyring.json` | Explicitly installed public trust anchor for runtime releases |
 | `backup-evidence.json` | Signed provider-neutral coordinated backup and isolated-restore receipt |
@@ -123,6 +124,16 @@ environment or initiating an implicit object migration. Invalid receipts fail
 closed, while a deployment without a mounted receipt continues to run but
 cannot apply receipt-bound configuration fragments.
 
+Enabled modules may also register a Core infrastructure-dependency provider.
+The authorized Ops endpoint aggregates those providers without importing their
+tables. Mail reports persisted SMTP endpoints, credential-binding counts and
+legacy profiles; Files reports its runtime storage binding plus persisted blob
+counts and byte totals grouped by backend. Ops reports the active PostgreSQL,
+Redis coordination, ingress, and load-balancing runtime bindings. Provider output contains stable
+references, bounded numeric metrics and required migration actions, never
+credentials, endpoint secrets, tenant identifiers or file keys. A provider
+failure makes the entire inventory incomplete.
+
 Build the same dependency-free tool as one downloadable artifact:
 
 ```sh
@@ -424,10 +435,16 @@ infrastructure capability projections.
 
 - Adding a managed component creates its service and persistent volume.
 - Removing a component removes its service container on apply.
-- Replacing or removing a capability adds a review action that names the prior
+- Reconfiguring, replacing or removing a capability adds a review action that
-  and desired state/source plus declared module consumers. This does not claim
+  names the prior and desired state/source, declared consumers, actual
-  that the deployer can inspect module-owned database configuration; the
+  provider-reported dependency records and each required migration action.
-  operator must review that inventory before apply.
+- The deployer blocks that change when provider inventory is missing,
+  incomplete, more than five minutes old, from another installation, timestamped
+  in the future, or does not cover every impacted capability. It never treats
+  installer-declared consumers as proof that persisted module state is absent.
+- The inventory reports impact; it does not migrate or delete module-owned
+  configuration or data. Complete the reported preparation and collect again
+  immediately before apply.
 - Volumes are retained by default; deleting data requires a separate,
   deliberately destructive workflow.
 - Existing generated credentials are retained unless an explicit future rotate
@@ -455,6 +472,29 @@ dedicated ConfigMap and read-only file mount. Ops validates the bounded schema
 before displaying configured, externally supplied, available-unconfigured, or
 unavailable states and any pending post-install tasks.
 
+Collect current dependency evidence with an API key whose principal has one of
+the Ops read scopes:
+
+```sh
+export GOVOPLAN_OPS_API_KEY='<short-lived operator API key>'
+python3 govoplan-deploy.pyz collect-infrastructure-inventory \
+  --directory /srv/govoplan/example
+python3 govoplan-deploy.pyz doctor \
+  --directory /srv/govoplan/example
+python3 govoplan-deploy.pyz apply \
+  --directory /srv/govoplan/example
+unset GOVOPLAN_OPS_API_KEY
+```
+
+The command defaults to
+`<public-url>/api/v1/ops/infrastructure/dependencies`; `--ops-url` may select an
+explicit HTTPS endpoint (plain HTTP is accepted only on loopback). `apply`
+refreshes the inventory automatically when `GOVOPLAN_OPS_API_KEY` is present.
+Otherwise an already collected, current inventory may be used. The API key is
+sent only as `X-API-Key`, is never written to the bundle, and the inventory file
+is owner-readable only. Because it contains operational references and counts,
+handle it as private evidence even though it contains no secret material.
+
 Every apply operation is journalled before image pulls or runtime mutation. A
 failure before migration may restore a verified previous bundle. Once migration
 starts, recovery is forward-only unless an independently verified database

packages/govoplan-meta/pyproject.toml

@@ -4,13 +4,13 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "govoplan"
-version = "0.1.41"
+version = "0.1.42"
 description = "Developer convenience package for a versioned GovOPlaN composition"
 readme = "README.md"
 requires-python = ">=3.12"
 license = { text = "AGPL-3.0-or-later" }
 dependencies = [
-  "govoplan-core[server]==0.1.41",
+  "govoplan-core[server]==0.1.42",
   "govoplan-tenancy==0.1.20",
   "govoplan-organizations==0.1.20",
   "govoplan-identity==0.1.20",
@@ -20,12 +20,12 @@ dependencies = [
   "govoplan-policy==0.1.22",
   "govoplan-audit==0.1.20",
   "govoplan-dashboard==0.1.20",
-  "govoplan-files==0.1.23",
+  "govoplan-files==0.1.24",
-  "govoplan-mail==0.1.25",
+  "govoplan-mail==0.1.26",
   "govoplan-campaign==0.1.27",
   "govoplan-calendar==0.1.22",
   "govoplan-docs==0.1.22",
-  "govoplan-ops==0.1.20",
+  "govoplan-ops==0.1.21",
 ]
 
 [project.optional-dependencies]

requirements-release.txt

@@ -10,9 +10,9 @@ govoplan-admin @ git+ssh://git@git.add-ideas.de/GovOPlaN/govoplan-admin.git@v0.1
 govoplan-policy @ git+ssh://git@git.add-ideas.de/GovOPlaN/govoplan-policy.git@v0.1.22
 govoplan-audit @ git+ssh://git@git.add-ideas.de/GovOPlaN/govoplan-audit.git@v0.1.20
 govoplan-dashboard @ git+ssh://git@git.add-ideas.de/GovOPlaN/govoplan-dashboard.git@v0.1.20
-govoplan-files @ git+ssh://git@git.add-ideas.de/GovOPlaN/govoplan-files.git@v0.1.23
+govoplan-files @ git+ssh://git@git.add-ideas.de/GovOPlaN/govoplan-files.git@v0.1.24
-govoplan-mail @ git+ssh://git@git.add-ideas.de/GovOPlaN/govoplan-mail.git@v0.1.25
+govoplan-mail @ git+ssh://git@git.add-ideas.de/GovOPlaN/govoplan-mail.git@v0.1.26
 govoplan-campaign @ git+ssh://git@git.add-ideas.de/GovOPlaN/govoplan-campaign.git@v0.1.27
 govoplan-calendar @ git+ssh://git@git.add-ideas.de/GovOPlaN/govoplan-calendar.git@v0.1.22
 govoplan-docs @ git+ssh://git@git.add-ideas.de/GovOPlaN/govoplan-docs.git@v0.1.22
-govoplan-ops @ git+ssh://git@git.add-ideas.de/GovOPlaN/govoplan-ops.git@v0.1.20
+govoplan-ops @ git+ssh://git@git.add-ideas.de/GovOPlaN/govoplan-ops.git@v0.1.21

tests/test_deployment_installer.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 from contextlib import redirect_stderr, redirect_stdout
+from datetime import UTC, datetime, timedelta
 import io
 import json
+import os
 from pathlib import Path
 import stat
 import subprocess
@@ -36,6 +38,7 @@ import govoplan_deploy.cli as deployment_cli  # noqa: E402
 from govoplan_deploy.capabilities import (  # noqa: E402
     capability_change_impacts,
     infrastructure_capability_document,
+    infrastructure_dependency_inventory_from_mapping,
 )
 from govoplan_deploy.cluster_evidence import (  # noqa: E402
     collect_kubernetes_evidence,
@@ -87,6 +90,41 @@ def _kubernetes_test_deployment(component: str, replicas: int) -> dict:
     }
 
 
+def _dependency_inventory(
+    installation_id: str,
+    *,
+    generated_at: datetime | None = None,
+) -> dict:
+    return {
+        "schema_version": 1,
+        "installation_id": installation_id,
+        "generated_at": (generated_at or datetime.now(UTC)).isoformat(),
+        "complete": True,
+        "inspected_capability_ids": ["coordination.redis", "mail.smtp"],
+        "providers": [
+            {
+                "module_id": "mail",
+                "state": "complete",
+                "capability_ids": ["mail.smtp"],
+                "dependency_count": 1,
+            }
+        ],
+        "dependencies": [
+            {
+                "capability_id": "mail.smtp",
+                "module_id": "mail",
+                "dependency_type": "smtp_endpoint",
+                "dependency_ref": "endpoint:17",
+                "state": "active",
+                "scope": "system",
+                "summary": "Persisted SMTP endpoint has one credential binding.",
+                "metrics": {"credential_binding_count": 1},
+                "required_action": "Rebind or migrate this SMTP endpoint.",
+            }
+        ],
+    }
+
+
 class DeploymentInstallerTests(unittest.TestCase):
     def test_kubernetes_evidence_requires_two_node_spread_and_safe_runtime(
         self,
@@ -1034,6 +1072,28 @@ class DeploymentInstallerTests(unittest.TestCase):
         self.assertNotIn("old-secret", impacts["database.postgresql"].detail)
         self.assertNotIn("new-secret", impacts["database.postgresql"].detail)
 
+    def test_capability_impact_includes_provider_dependency_evidence(self) -> None:
+        previous_spec = default_spec(mail_mode="test-mail", module_set="full")
+        desired_spec = default_spec(mail_mode="disabled", module_set="full")
+        inventory = infrastructure_dependency_inventory_from_mapping(
+            _dependency_inventory(previous_spec.installation_id)
+        )
+
+        impacts = {
+            item.capability_id: item
+            for item in capability_change_impacts(
+                infrastructure_capability_document(previous_spec, {}),
+                infrastructure_capability_document(desired_spec, {}),
+                dependency_inventory=inventory,
+            )
+        }
+
+        mail = impacts["mail.smtp"]
+        self.assertTrue(mail.inventory_inspected)
+        self.assertEqual("endpoint:17", mail.actual_dependencies[0].dependency_ref)
+        self.assertIn("mail:endpoint:17", mail.detail)
+        self.assertIn("Rebind or migrate", mail.required_action)
+
     def test_replica_counts_drive_compose_and_load_balancer_discovery(self) -> None:
         spec = default_spec(
             storage_mode="garage",
@@ -1221,6 +1281,65 @@ class DeploymentInstallerTests(unittest.TestCase):
                     for check in second_plan.checks
                 )
             )
+            self.assertTrue(second_plan.blocked)
+            self.assertTrue(
+                any(
+                    check.id == "capability.dependency_inventory.missing"
+                    and check.level == "error"
+                    for check in second_plan.checks
+                )
+            )
+
+            atomic_write(
+                paths.dependency_inventory,
+                canonical_json(_dependency_inventory(second_spec.installation_id)),
+                mode=0o600,
+            )
+            evidenced_plan = build_plan(
+                second_spec,
+                paths,
+                include_host_checks=False,
+            )
+
+            self.assertFalse(
+                any(
+                    check.level == "error"
+                    and check.id.startswith("capability.dependency_inventory.")
+                    for check in evidenced_plan.checks
+                )
+            )
+            self.assertEqual(
+                "endpoint:17",
+                {
+                    item.capability_id: item
+                    for item in evidenced_plan.capability_impacts
+                }["mail.smtp"].actual_dependencies[0].dependency_ref,
+            )
+            self.assertTrue(
+                any(
+                    check.id == "capability.dependency_inventory.current"
+                    and check.level == "ok"
+                    for check in evidenced_plan.checks
+                )
+            )
+
+            stale = _dependency_inventory(
+                second_spec.installation_id,
+                generated_at=datetime.now(UTC) - timedelta(minutes=6),
+            )
+            atomic_write(
+                paths.dependency_inventory,
+                canonical_json(stale),
+                mode=0o600,
+            )
+            stale_plan = build_plan(second_spec, paths, include_host_checks=False)
+            self.assertTrue(stale_plan.blocked)
+            self.assertTrue(
+                any(
+                    check.id == "capability.dependency_inventory.stale"
+                    for check in stale_plan.checks
+                )
+            )
 
     def test_secret_change_is_planned_without_exposing_secret_values(self) -> None:
         with tempfile.TemporaryDirectory(prefix="govoplan-deploy-test-") as directory:
@@ -1330,6 +1449,54 @@ class DeploymentInstallerTests(unittest.TestCase):
                 )[0],
             )
 
+    def test_cli_collects_bounded_private_dependency_inventory(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="govoplan-deploy-test-") as directory:
+            root = Path(directory) / "installation"
+            self.assertEqual(
+                0,
+                run_cli(
+                    [
+                        "init",
+                        "--non-interactive",
+                        "--directory",
+                        str(root),
+                    ]
+                )[0],
+            )
+            payload = _dependency_inventory("govoplan-local")
+            response = MagicMock()
+            response.__enter__.return_value = response
+            response.geturl.return_value = "https://ops.example.test/inventory"
+            response.read.return_value = json.dumps(payload).encode("utf-8")
+            fetch = MagicMock(return_value=response)
+
+            with (
+                patch.dict(os.environ, {"TEST_OPS_KEY": "secret-api-key"}),
+                patch.object(deployment_cli, "urlopen", fetch),
+            ):
+                result, stdout, stderr = run_cli(
+                    [
+                        "collect-infrastructure-inventory",
+                        "--directory",
+                        str(root),
+                        "--ops-url",
+                        "https://ops.example.test/inventory",
+                        "--api-key-env",
+                        "TEST_OPS_KEY",
+                    ]
+                )
+
+            self.assertEqual(0, result, stderr)
+            self.assertIn("1 record(s)", stdout)
+            evidence_path = root / "infrastructure-dependency-inventory.json"
+            self.assertEqual(0o600, stat.S_IMODE(evidence_path.stat().st_mode))
+            self.assertNotIn(
+                "secret-api-key",
+                evidence_path.read_text(encoding="utf-8"),
+            )
+            request = fetch.call_args.args[0]
+            self.assertEqual("secret-api-key", request.get_header("X-api-key"))
+
     def test_cli_requires_external_url_when_switching_from_managed(self) -> None:
         with tempfile.TemporaryDirectory(prefix="govoplan-deploy-test-") as directory:
             root = Path(directory) / "installation"

tools/deployment/govoplan_deploy/bundle.py

@@ -27,6 +27,7 @@ EXISTING_PROXY_FILENAME = "existing-proxy.json"
 PLAN_FILENAME = "plan.json"
 RECEIPT_FILENAME = "receipt.json"
 CAPABILITIES_FILENAME = "infrastructure-capabilities.json"
+DEPENDENCY_INVENTORY_FILENAME = "infrastructure-dependency-inventory.json"
 MANIFEST_FILENAME = "distribution-manifest.json"
 KEYRING_FILENAME = "distribution-keyring.json"
 BACKUP_EVIDENCE_FILENAME = "backup-evidence.json"
@@ -116,6 +117,7 @@ class BundlePaths:
     plan: Path
     receipt: Path
     capabilities: Path
+    dependency_inventory: Path
     manifest: Path
     keyring: Path
     backup_evidence: Path
@@ -141,6 +143,7 @@ def bundle_paths(root: Path) -> BundlePaths:
         plan=resolved / PLAN_FILENAME,
         receipt=resolved / RECEIPT_FILENAME,
         capabilities=resolved / CAPABILITIES_FILENAME,
+        dependency_inventory=resolved / DEPENDENCY_INVENTORY_FILENAME,
         manifest=resolved / MANIFEST_FILENAME,
         keyring=resolved / KEYRING_FILENAME,
         backup_evidence=resolved / BACKUP_EVIDENCE_FILENAME,

tools/deployment/govoplan_deploy/capabilities.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from dataclasses import asdict, dataclass
+from datetime import UTC, datetime
 from typing import Mapping
 from urllib.parse import urlsplit
 
@@ -18,6 +19,10 @@ CAPABILITY_STATES = frozenset(
         "unavailable",
     }
 )
+DEPENDENCY_INVENTORY_SCHEMA_VERSION = 1
+DEPENDENCY_STATES = frozenset(
+    {"active", "inactive", "data_present", "pending_work", "runtime_binding"}
+)
 
 
 @dataclass(frozen=True, slots=True)
@@ -50,13 +55,161 @@ class CapabilityChangeImpact:
     dependent_modules: tuple[str, ...]
     detail: str
     required_action: str
+    actual_dependencies: tuple["CapabilityDependency", ...] = ()
+    inventory_inspected: bool = False
 
     def to_dict(self) -> dict[str, object]:
         value = asdict(self)
         value["dependent_modules"] = list(self.dependent_modules)
+        value["actual_dependencies"] = [
+            item.to_dict() for item in self.actual_dependencies
+        ]
         return value
 
 
+@dataclass(frozen=True, slots=True)
+class CapabilityDependency:
+    capability_id: str
+    module_id: str
+    dependency_type: str
+    dependency_ref: str
+    state: str
+    scope: str
+    summary: str
+    metrics: Mapping[str, int]
+    required_action: str
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "capability_id": self.capability_id,
+            "module_id": self.module_id,
+            "dependency_type": self.dependency_type,
+            "dependency_ref": self.dependency_ref,
+            "state": self.state,
+            "scope": self.scope,
+            "summary": self.summary,
+            "metrics": dict(sorted(self.metrics.items())),
+            "required_action": self.required_action,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class InfrastructureDependencyInventory:
+    installation_id: str
+    generated_at: datetime
+    complete: bool
+    inspected_capability_ids: tuple[str, ...]
+    provider_count: int
+    dependencies: tuple[CapabilityDependency, ...]
+
+    def dependencies_for(
+        self,
+        capability_id: str,
+    ) -> tuple[CapabilityDependency, ...]:
+        return tuple(
+            item for item in self.dependencies if item.capability_id == capability_id
+        )
+
+
+def infrastructure_dependency_inventory_from_mapping(
+    value: object,
+) -> InfrastructureDependencyInventory:
+    if (
+        not isinstance(value, Mapping)
+        or value.get("schema_version") != DEPENDENCY_INVENTORY_SCHEMA_VERSION
+    ):
+        raise ValueError("Infrastructure dependency inventory schema is unsupported.")
+    installation_id = _inventory_text(value, "installation_id", maximum=100)
+    generated_at_text = _inventory_text(value, "generated_at", maximum=100)
+    try:
+        generated_at = datetime.fromisoformat(generated_at_text.replace("Z", "+00:00"))
+    except ValueError as exc:
+        raise ValueError(
+            "Infrastructure dependency inventory timestamp is invalid."
+        ) from exc
+    if generated_at.tzinfo is None:
+        raise ValueError("Infrastructure dependency inventory timestamp needs a timezone.")
+    generated_at = generated_at.astimezone(UTC)
+    complete = value.get("complete")
+    if type(complete) is not bool:
+        raise ValueError("Infrastructure dependency inventory completion state is invalid.")
+    inspected = _inventory_string_list(
+        value.get("inspected_capability_ids"),
+        maximum_items=100,
+        maximum_length=120,
+    )
+    if len(inspected) != len(set(inspected)):
+        raise ValueError("Infrastructure dependency inventory repeats a capability id.")
+    providers = value.get("providers")
+    if not isinstance(providers, list) or len(providers) > 100:
+        raise ValueError("Infrastructure dependency provider reports are invalid.")
+    provider_states: list[str] = []
+    provider_declarations: dict[str, tuple[str, ...]] = {}
+    provider_counts: dict[str, int] = {}
+    for provider in providers:
+        if not isinstance(provider, Mapping):
+            raise ValueError("Infrastructure dependency provider report is invalid.")
+        module_id = _inventory_text(provider, "module_id", maximum=120)
+        if module_id in provider_declarations:
+            raise ValueError("Infrastructure dependency provider is repeated.")
+        state = _inventory_text(provider, "state", maximum=40)
+        if state not in {"complete", "error"}:
+            raise ValueError("Infrastructure dependency provider state is invalid.")
+        provider_states.append(state)
+        count = provider.get("dependency_count")
+        if type(count) is not int or count < 0:
+            raise ValueError("Infrastructure dependency provider count is invalid.")
+        capability_ids = _inventory_string_list(
+            provider.get("capability_ids"),
+            maximum_items=30,
+            maximum_length=120,
+        )
+        if len(capability_ids) != len(set(capability_ids)):
+            raise ValueError("Infrastructure dependency provider capability is repeated.")
+        provider_declarations[module_id] = capability_ids
+        provider_counts[module_id] = count
+    if complete and any(state != "complete" for state in provider_states):
+        raise ValueError("Complete dependency inventory contains a failed provider.")
+    raw_dependencies = value.get("dependencies")
+    if not isinstance(raw_dependencies, list) or len(raw_dependencies) > 10_000:
+        raise ValueError("Infrastructure dependency records are invalid.")
+    dependencies = tuple(_inventory_dependency(item) for item in raw_dependencies)
+    if any(
+        capability_id not in inspected
+        for capability_ids in provider_declarations.values()
+        for capability_id in capability_ids
+    ):
+        raise ValueError(
+            "Infrastructure dependency provider was not covered by the inspection."
+        )
+    if any(item.capability_id not in inspected for item in dependencies):
+        raise ValueError("Dependency record was not covered by the inventory inspection.")
+    identities = {
+        (item.capability_id, item.module_id, item.dependency_type, item.dependency_ref)
+        for item in dependencies
+    }
+    if len(identities) != len(dependencies):
+        raise ValueError("Infrastructure dependency inventory repeats a record.")
+    observed_counts = {module_id: 0 for module_id in provider_counts}
+    for dependency in dependencies:
+        declarations = provider_declarations.get(dependency.module_id)
+        if declarations is None or dependency.capability_id not in declarations:
+            raise ValueError(
+                "Infrastructure dependency is outside its provider declaration."
+            )
+        observed_counts[dependency.module_id] += 1
+    if observed_counts != provider_counts:
+        raise ValueError("Infrastructure dependency provider count does not match records.")
+    return InfrastructureDependencyInventory(
+        installation_id=installation_id,
+        generated_at=generated_at,
+        complete=complete,
+        inspected_capability_ids=inspected,
+        provider_count=len(providers),
+        dependencies=dependencies,
+    )
+
+
 def infrastructure_capability_document(
     spec: InstallationSpec,
     environment: Mapping[str, str],
@@ -89,6 +242,8 @@ def infrastructure_capability_document(
 def capability_change_impacts(
     previous_document: object,
     desired_document: Mapping[str, object],
+    *,
+    dependency_inventory: InfrastructureDependencyInventory | None = None,
 ) -> tuple[CapabilityChangeImpact, ...]:
     previous = _capability_map(previous_document)
     desired = _capability_map(desired_document)
@@ -141,6 +296,43 @@ def capability_change_impacts(
             previous_secret_refs,
             desired_secret_refs,
         )
+        actual_dependencies = (
+            dependency_inventory.dependencies_for(capability_id)
+            if dependency_inventory is not None
+            else ()
+        )
+        inventory_inspected = bool(
+            dependency_inventory is not None
+            and capability_id in dependency_inventory.inspected_capability_ids
+        )
+        if inventory_inspected and actual_dependencies:
+            references = ", ".join(
+                f"{item.module_id}:{item.dependency_ref}"
+                for item in actual_dependencies
+            )
+            inventory_detail = (
+                f" Provider inventory reports {len(actual_dependencies)} persisted "
+                f"dependency record(s): {references}."
+            )
+        elif inventory_inspected:
+            inventory_detail = (
+                " Provider inventory reports no persisted module-owned dependencies."
+            )
+        else:
+            inventory_detail = " Provider inventory did not inspect this capability."
+        dependency_actions = tuple(
+            dict.fromkeys(
+                item.required_action
+                for item in actual_dependencies
+                if item.required_action.strip()
+            )
+        )
+        required_action = (
+            "Review module-owned configuration and data migration or recovery "
+            "evidence before apply."
+        )
+        if dependency_actions:
+            required_action = f"{required_action} {' '.join(dependency_actions)}"
         impacts.append(
             CapabilityChangeImpact(
                 capability_id=capability_id,
@@ -153,11 +345,11 @@ def capability_change_impacts(
                 detail=(
                     f"{capability_id} changes from {previous_state}/{previous_source} "
                     f"to {desired_state}/{desired_source}{binding_change}; "
-                    f"declared consumers: {dependent_label}."
+                    f"declared consumers: {dependent_label}.{inventory_detail}"
-                ),
-                required_action=(
-                    "Review module-owned configuration and data migration or recovery evidence before apply."
                 ),
+                required_action=required_action,
+                actual_dependencies=actual_dependencies,
+                inventory_inspected=inventory_inspected,
             )
         )
     return tuple(impacts)
@@ -487,3 +679,73 @@ def _binding_change_label(
     if previous_secret_refs != desired_secret_refs:
         changes.append("secret-reference binding")
     return f" with changed {' and '.join(changes)}" if changes else ""
+
+
+def _inventory_text(
+    value: Mapping[str, object],
+    key: str,
+    *,
+    maximum: int,
+) -> str:
+    raw = value.get(key)
+    if not isinstance(raw, str):
+        raise ValueError(f"Infrastructure dependency inventory {key} is invalid.")
+    result = raw.strip()
+    if not result or len(result) > maximum or any(ord(char) < 32 for char in result):
+        raise ValueError(f"Infrastructure dependency inventory {key} is invalid.")
+    return result
+
+
+def _inventory_string_list(
+    value: object,
+    *,
+    maximum_items: int,
+    maximum_length: int,
+) -> tuple[str, ...]:
+    if not isinstance(value, list) or len(value) > maximum_items:
+        raise ValueError("Infrastructure dependency inventory list is invalid.")
+    items: list[str] = []
+    for raw in value:
+        if not isinstance(raw, str):
+            raise ValueError("Infrastructure dependency inventory list is invalid.")
+        item = raw.strip()
+        if (
+            not item
+            or len(item) > maximum_length
+            or any(ord(char) < 32 for char in item)
+        ):
+            raise ValueError("Infrastructure dependency inventory list is invalid.")
+        items.append(item)
+    return tuple(items)
+
+
+def _inventory_dependency(value: object) -> CapabilityDependency:
+    if not isinstance(value, Mapping):
+        raise ValueError("Infrastructure dependency record is invalid.")
+    state = _inventory_text(value, "state", maximum=40)
+    if state not in DEPENDENCY_STATES:
+        raise ValueError("Infrastructure dependency state is invalid.")
+    raw_metrics = value.get("metrics")
+    if not isinstance(raw_metrics, Mapping) or len(raw_metrics) > 20:
+        raise ValueError("Infrastructure dependency metrics are invalid.")
+    metrics: dict[str, int] = {}
+    for raw_key, raw_count in raw_metrics.items():
+        if not isinstance(raw_key, str):
+            raise ValueError("Infrastructure dependency metric name is invalid.")
+        key = raw_key.strip()
+        if not key or len(key) > 80 or any(ord(char) < 32 for char in key):
+            raise ValueError("Infrastructure dependency metric name is invalid.")
+        if type(raw_count) is not int or raw_count < 0:
+            raise ValueError("Infrastructure dependency metric value is invalid.")
+        metrics[key] = raw_count
+    return CapabilityDependency(
+        capability_id=_inventory_text(value, "capability_id", maximum=120),
+        module_id=_inventory_text(value, "module_id", maximum=120),
+        dependency_type=_inventory_text(value, "dependency_type", maximum=120),
+        dependency_ref=_inventory_text(value, "dependency_ref", maximum=240),
+        state=state,
+        scope=_inventory_text(value, "scope", maximum=120),
+        summary=_inventory_text(value, "summary", maximum=1000),
+        metrics=metrics,
+        required_action=_inventory_text(value, "required_action", maximum=1000),
+    )

tools/deployment/govoplan_deploy/cli.py

@@ -18,7 +18,8 @@ import sys
 import time
 from typing import Iterator, Mapping, Sequence
 from urllib.error import URLError
-from urllib.request import urlopen
+from urllib.parse import urlsplit
+from urllib.request import Request, urlopen
 
 from .backup_evidence import (
     DEFAULT_MAX_BACKUP_AGE_SECONDS,
@@ -28,6 +29,7 @@ from .backup_evidence import (
 )
 from .bundle import (
     BACKUP_RUNTIME_ENV_KEYS,
+    BundlePaths,
     atomic_write,
     bundle_paths,
     canonical_json,
@@ -45,7 +47,11 @@ from .bundle import (
     service_names,
     write_env,
 )
-from .capabilities import infrastructure_capability_document
+from .capabilities import (
+    InfrastructureDependencyInventory,
+    infrastructure_capability_document,
+    infrastructure_dependency_inventory_from_mapping,
+)
 from .cluster_evidence import collect_kubernetes_evidence
 from .distribution import (
     MAX_KEYRING_BYTES,
@@ -81,6 +87,7 @@ from .kubernetes import (
     write_secret_creation_hint,
 )
 from .planning import (
+    MAX_DEPENDENCY_INVENTORY_BYTES,
     DeploymentPlan,
     build_plan,
     release_change_requires_backup,
@@ -152,6 +159,39 @@ def build_parser() -> argparse.ArgumentParser:
         default=120.0,
         help="Maximum time to wait for the public health endpoint.",
     )
+    apply_parser.add_argument(
+        "--ops-url",
+        help=(
+            "Dependency inventory URL; defaults to "
+            "<public-url>/api/v1/ops/infrastructure/dependencies."
+        ),
+    )
+    apply_parser.add_argument(
+        "--api-key-env",
+        default="GOVOPLAN_OPS_API_KEY",
+        help=(
+            "Environment variable containing an API key authorized to read "
+            "Ops dependency inventory."
+        ),
+    )
+
+    collect_inventory = subparsers.add_parser(
+        "collect-infrastructure-inventory",
+        help="Collect current module-owned capability dependencies from Ops.",
+    )
+    _directory_argument(collect_inventory)
+    collect_inventory.add_argument(
+        "--ops-url",
+        help=(
+            "Dependency inventory URL; defaults to "
+            "<public-url>/api/v1/ops/infrastructure/dependencies."
+        ),
+    )
+    collect_inventory.add_argument(
+        "--api-key-env",
+        default="GOVOPLAN_OPS_API_KEY",
+        help="Environment variable containing an authorized Ops API key.",
+    )
 
     status = subparsers.add_parser(
         "status", help="Show desired state and current Compose process state."
@@ -425,6 +465,8 @@ def main(argv: Sequence[str] | None = None) -> int:
             return _render_or_doctor(args)
         if args.command == "apply":
             return _apply(args)
+        if args.command == "collect-infrastructure-inventory":
+            return _collect_infrastructure_inventory(args)
         if args.command == "status":
             return _status(args)
         if args.command == "verify-release":
@@ -586,6 +628,25 @@ def _apply(args: argparse.Namespace) -> int:
             )
         secrets = reconcile_runtime_environment(spec, read_env(paths.env))
         secrets = _write_bundle(spec, paths, secrets)
+        preliminary_plan = build_plan(spec, paths, include_host_checks=False)
+        api_key_env = str(
+            getattr(args, "api_key_env", "GOVOPLAN_OPS_API_KEY")
+        ).strip()
+        api_key = os.environ.get(api_key_env, "").strip()
+        if preliminary_plan.capability_impacts and api_key:
+            try:
+                _collect_dependency_inventory(
+                    spec,
+                    paths,
+                    ops_url=getattr(args, "ops_url", None),
+                    api_key=api_key,
+                )
+                print("Refreshed infrastructure dependency inventory from Ops.")
+            except (OSError, ValueError, json.JSONDecodeError) as exc:
+                print(
+                    f"warning: could not refresh dependency inventory: {exc}",
+                    file=sys.stderr,
+                )
         plan = build_plan(spec, paths, include_host_checks=True)
         _write_plan(paths.plan, plan)
         effective_errors = [
@@ -613,6 +674,8 @@ def _apply(args: argparse.Namespace) -> int:
         if effective_errors:
             _print_plan(plan)
             raise ValueError("deployment plan is blocked; resolve doctor errors first")
+        if plan.capability_impacts:
+            _print_plan(plan)
         docker = shutil.which("docker")
         if docker is None:
             raise ValueError("Docker CLI is required for apply")
@@ -761,6 +824,70 @@ def _apply(args: argparse.Namespace) -> int:
         return 0
 
 
+def _collect_infrastructure_inventory(args: argparse.Namespace) -> int:
+    paths = bundle_paths(args.directory)
+    spec = load_spec(paths.spec)
+    api_key_env = str(args.api_key_env).strip()
+    api_key = os.environ.get(api_key_env, "").strip()
+    if not api_key:
+        raise ValueError(f"{api_key_env} must contain an authorized Ops API key")
+    inventory = _collect_dependency_inventory(
+        spec,
+        paths,
+        ops_url=args.ops_url,
+        api_key=api_key,
+    )
+    state = "complete" if inventory.complete else "incomplete"
+    print(
+        f"Collected {state} provider dependency inventory with "
+        f"{len(inventory.dependencies)} record(s) at {paths.dependency_inventory}."
+    )
+    return 0 if inventory.complete else 1
+
+
+def _collect_dependency_inventory(
+    spec: InstallationSpec,
+    paths: BundlePaths,
+    *,
+    ops_url: str | None,
+    api_key: str,
+) -> InfrastructureDependencyInventory:
+    url = str(ops_url or "").strip() or (
+        spec.public_url.rstrip("/")
+        + "/api/v1/ops/infrastructure/dependencies"
+    )
+    _validate_ops_inventory_url(url)
+    request = Request(
+        url,
+        headers={"Accept": "application/json", "X-API-Key": api_key},
+    )
+    with urlopen(request, timeout=15) as response:  # noqa: S310
+        _validate_ops_inventory_url(response.geturl())
+        encoded = response.read(MAX_DEPENDENCY_INVENTORY_BYTES + 1)
+    if len(encoded) > MAX_DEPENDENCY_INVENTORY_BYTES:
+        raise ValueError("Ops dependency inventory exceeds its size limit")
+    value = json.loads(encoded)
+    inventory = infrastructure_dependency_inventory_from_mapping(value)
+    if inventory.installation_id != spec.installation_id:
+        raise ValueError(
+            "Ops dependency inventory belongs to a different installation"
+        )
+    ensure_private_directory(paths.root)
+    atomic_write(paths.dependency_inventory, canonical_json(value), mode=0o600)
+    return inventory
+
+
+def _validate_ops_inventory_url(url: str) -> None:
+    parsed = urlsplit(url)
+    if not parsed.hostname or parsed.username or parsed.password or parsed.fragment:
+        raise ValueError("Ops dependency inventory URL is invalid")
+    loopback = parsed.hostname in {"localhost", "127.0.0.1", "::1"}
+    if parsed.scheme != "https" and not (parsed.scheme == "http" and loopback):
+        raise ValueError(
+            "Ops dependency inventory URL requires HTTPS except on loopback"
+        )
+
+
 def _status(args: argparse.Namespace) -> int:
     paths = bundle_paths(args.directory)
     spec = load_spec(paths.spec)

tools/deployment/govoplan_deploy/planning.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from dataclasses import asdict, dataclass
+from datetime import UTC, datetime
 import hashlib
 import json
 import os
@@ -37,8 +38,10 @@ from .bundle import (
 )
 from .capabilities import (
     CapabilityChangeImpact,
+    InfrastructureDependencyInventory,
     capability_change_impacts,
     infrastructure_capability_document,
+    infrastructure_dependency_inventory_from_mapping,
 )
 from .distribution import (
     MAX_KEYRING_BYTES,
@@ -110,6 +113,9 @@ class DeploymentPlan:
 
 
 CommandRunner = Callable[[Sequence[str], Path], subprocess.CompletedProcess[str]]
+MAX_DEPENDENCY_INVENTORY_BYTES = 2 * 1024 * 1024
+DEPENDENCY_INVENTORY_MAX_AGE_SECONDS = 300
+DEPENDENCY_INVENTORY_MAX_FUTURE_SECONDS = 60
 
 
 def build_plan(
@@ -135,9 +141,13 @@ def build_plan(
         spec,
         read_env(paths.env),
     )
+    dependency_inventory, dependency_inventory_error = (
+        _read_dependency_inventory(paths.dependency_inventory)
+    )
     capability_impacts = capability_change_impacts(
         previous.get("infrastructure_capabilities"),
         infrastructure_capabilities,
+        dependency_inventory=dependency_inventory,
     )
 
     actions: list[PlanAction] = []
@@ -215,6 +225,14 @@ def build_plan(
         )
         for impact in capability_impacts
     )
+    checks.extend(
+        _dependency_inventory_checks(
+            spec,
+            capability_impacts,
+            dependency_inventory,
+            dependency_inventory_error,
+        )
+    )
     if include_host_checks:
         checks.extend(host_checks(spec, paths, command_runner=command_runner))
     return DeploymentPlan(
@@ -1187,6 +1205,106 @@ def _read_receipt(path: Path) -> Mapping[str, object]:
     return value if isinstance(value, dict) else {}
 
 
+def _read_dependency_inventory(
+    path: Path,
+) -> tuple[InfrastructureDependencyInventory | None, str]:
+    if not path.exists():
+        return None, "missing"
+    try:
+        value = load_bounded_json(
+            path,
+            maximum_bytes=MAX_DEPENDENCY_INVENTORY_BYTES,
+        )
+        return infrastructure_dependency_inventory_from_mapping(value), ""
+    except (DistributionError, ValueError) as exc:
+        return None, str(exc)
+
+
+def _dependency_inventory_checks(
+    spec: InstallationSpec,
+    impacts: tuple[CapabilityChangeImpact, ...],
+    inventory: InfrastructureDependencyInventory | None,
+    inventory_error: str,
+) -> tuple[Check, ...]:
+    if not impacts:
+        return ()
+    collect_action = (
+        "Run govoplan-deploy collect-infrastructure-inventory with an Ops API "
+        "key, then review the capability impacts before apply."
+    )
+    if inventory is None:
+        if inventory_error == "missing":
+            message = "Current provider dependency inventory is missing."
+            check_id = "capability.dependency_inventory.missing"
+        else:
+            message = f"Provider dependency inventory is invalid: {inventory_error}"
+            check_id = "capability.dependency_inventory.invalid"
+        return (Check(check_id, "error", message, collect_action),)
+    if inventory.installation_id != spec.installation_id:
+        return (
+            Check(
+                "capability.dependency_inventory.installation",
+                "error",
+                "Provider dependency inventory belongs to a different installation.",
+                collect_action,
+            ),
+        )
+    if not inventory.complete:
+        return (
+            Check(
+                "capability.dependency_inventory.incomplete",
+                "error",
+                "Provider dependency inventory is incomplete because at least one provider failed.",
+                "Resolve the provider failure and collect the inventory again.",
+            ),
+        )
+    age_seconds = (datetime.now(UTC) - inventory.generated_at).total_seconds()
+    if age_seconds < -DEPENDENCY_INVENTORY_MAX_FUTURE_SECONDS:
+        return (
+            Check(
+                "capability.dependency_inventory.future",
+                "error",
+                "Provider dependency inventory timestamp is in the future.",
+                "Correct host clock skew and collect the inventory again.",
+            ),
+        )
+    if age_seconds > DEPENDENCY_INVENTORY_MAX_AGE_SECONDS:
+        return (
+            Check(
+                "capability.dependency_inventory.stale",
+                "error",
+                "Provider dependency inventory is older than five minutes.",
+                collect_action,
+            ),
+        )
+    impacted_ids = {item.capability_id for item in impacts}
+    missing_ids = sorted(impacted_ids - set(inventory.inspected_capability_ids))
+    if missing_ids:
+        return (
+            Check(
+                "capability.dependency_inventory.coverage",
+                "error",
+                "Provider dependency inventory did not inspect impacted capabilities: "
+                + ", ".join(missing_ids)
+                + ".",
+                collect_action,
+            ),
+        )
+    matching_dependencies = sum(
+        len(inventory.dependencies_for(capability_id))
+        for capability_id in impacted_ids
+    )
+    return (
+        Check(
+            "capability.dependency_inventory.current",
+            "ok",
+            "Current provider inventory inspected every impacted capability and "
+            f"reported {matching_dependencies} persisted dependency record(s) from "
+            f"{inventory.provider_count} provider(s).",
+        ),
+    )
+
+
 def _memory_bytes() -> int | None:
     try:
         for line in Path("/proc/meminfo").read_text(encoding="utf-8").splitlines():

tools/deployment/govoplan_deploy/recovery.py

@@ -34,6 +34,7 @@ _BUNDLE_FILES = (
     "backup-verification.json",
     "receipt.json",
     "infrastructure-capabilities.json",
+    "infrastructure-dependency-inventory.json",
 )
 
 

tools/inventory/endpoint-surface-declarations.json

@@ -1379,6 +1379,13 @@
       "rationale": "Operational worker, scheduler, reconciliation, or health endpoint; it is not a direct user surface.",
       "repository": "govoplan-notifications"
     },
+    {
+      "category": "intentionally_headless",
+      "method": "GET",
+      "path": "/ops/infrastructure/dependencies",
+      "rationale": "Authorized host-deployer preflight consumes this provider inventory directly; it is private operational evidence rather than a product page.",
+      "repository": "govoplan-ops"
+    },
     {
       "category": "worker_internal",
       "method": "GET",