diff --git a/openstack_hypervisor/hooks.py b/openstack_hypervisor/hooks.py
index 7de5c79..78b2489 100644
--- a/openstack_hypervisor/hooks.py
+++ b/openstack_hypervisor/hooks.py
@@ -3096,6 +3096,7 @@ def configure(snap: Snap) -> None:
             )
         else:
             _ensure_internal_ovs_services(snap, exclude_services)
+            _ensure_internal_ovs_dependent_services(snap, exclude_services)
 
 
 def _get_configure_context(snap: Snap) -> dict:
@@ -3184,6 +3185,10 @@ def _get_configure_context(snap: Snap) -> dict:
     "ovs-exporter",
 ]
 
+INTERNAL_OVS_DEPENDENT_SERVICES = [
+    "neutron-ovn-metadata-agent",
+]
+
 
 def _ensure_internal_ovs_services(snap: Snap, exclude_services: list[str]) -> None:
     """Ensure internal OVS services are enabled when not excluded.
@@ -3205,6 +3210,17 @@ def _ensure_internal_ovs_services(snap: Snap, exclude_services: list[str]) -> No
         services[service].start(enable=True)
 
 
+def _ensure_internal_ovs_dependent_services(snap: Snap, exclude_services: list[str]) -> None:
+    """Ensure services that need internal OVS are enabled when not excluded."""
+    services = snap.services.list()
+
+    for service in INTERNAL_OVS_DEPENDENT_SERVICES:
+        if service in exclude_services:
+            continue
+        logging.info("Ensuring internal OVS-dependent service is enabled: %s", service)
+        services[service].start(enable=True)
+
+
 def _get_exclude_services(context: dict) -> list[str]:
     """Get list of services to exclude based on configuration and external OVS state.
 
diff --git a/openstack_hypervisor/services.py b/openstack_hypervisor/services.py
index 4e471d6..978f696 100644
--- a/openstack_hypervisor/services.py
+++ b/openstack_hypervisor/services.py
@@ -3,10 +3,12 @@
 
 import base64
 import binascii
+import configparser
 import logging
 import os
 import subprocess
 import sys
+import time
 from functools import partial
 from pathlib import Path
 
@@ -15,6 +17,9 @@
 from openstack_hypervisor.log import setup_logging
 from openstack_hypervisor.mount_validation import validate_instances_mount
 
+OVSDB_SCHEMA_TIMEOUT = 60
+OVSDB_SCHEMA_CHECK_INTERVAL = 2
+
 
 def entry_point(service_class):
     """Entry point wrapper for services."""
@@ -131,6 +136,79 @@ class NeutronOVNMetadataAgentService(OpenStackService):
 
     executable = Path("usr/bin/neutron-ovn-metadata-agent")
 
+    def run(self, snap: Snap) -> int:
+        """Run neutron-ovn-metadata-agent once local OVSDB is ready."""
+        setup_logging(snap.paths.common / f"{self.executable.name}-{snap.name}.log")
+        ovsdb_connection = self._ovsdb_connection(snap)
+        if not ovsdb_connection:
+            return 1
+
+        if not self._wait_for_ovsdb_schema(snap, ovsdb_connection):
+            return 1
+
+        return super().run(snap)
+
+    def _ovsdb_connection(self, snap: Snap) -> str | None:
+        """Read the configured local OVSDB connection string."""
+        config_path = snap.paths.common / "etc/neutron/neutron_ovn_metadata_agent.ini"
+        parser = configparser.ConfigParser()
+        try:
+            if not parser.read(config_path):
+                logging.error("Unable to read OVN metadata agent config: %s", config_path)
+                return None
+            ovsdb_connection = parser.get("ovs", "ovsdb_connection", fallback="").strip()
+        except configparser.Error as exc:
+            logging.error("Unable to parse OVN metadata agent config %s: %s", config_path, exc)
+            return None
+
+        if not ovsdb_connection:
+            logging.error("ovsdb_connection is not configured in %s", config_path)
+            return None
+        return ovsdb_connection
+
+    def _wait_for_ovsdb_schema(self, snap: Snap, ovsdb_connection: str) -> bool:
+        """Wait until ovsdb-client can retrieve the Open_vSwitch schema."""
+        deadline = time.monotonic() + OVSDB_SCHEMA_TIMEOUT
+        socket_path = self._unix_socket_path(ovsdb_connection)
+        command = [
+            str(snap.paths.snap / "usr" / "bin" / "ovsdb-client"),
+            "get-schema",
+            ovsdb_connection,
+            "Open_vSwitch",
+        ]
+
+        while True:
+            if socket_path and not socket_path.exists():
+                logging.info("Waiting for local OVSDB socket: %s", socket_path)
+            elif self._ovsdb_schema_available(command):
+                return True
+
+            if time.monotonic() >= deadline:
+                logging.error(
+                    "Timed out waiting for Open_vSwitch schema from %s", ovsdb_connection
+                )
+                return False
+            time.sleep(OVSDB_SCHEMA_CHECK_INTERVAL)
+
+    def _ovsdb_schema_available(self, command: list[str]) -> bool:
+        """Return whether the OVSDB Open_vSwitch schema is reachable."""
+        try:
+            completed_process = subprocess.run(
+                command,
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+            )
+        except OSError as exc:
+            logging.info("Unable to query local OVSDB schema yet: %s", exc)
+            return False
+        return completed_process.returncode == 0
+
+    def _unix_socket_path(self, ovsdb_connection: str) -> Path | None:
+        """Return the socket path for unix OVSDB connections."""
+        if not ovsdb_connection.startswith("unix:"):
+            return None
+        return Path(ovsdb_connection.removeprefix("unix:"))
+
 
 neutron_ovn_metadata_agent = partial(entry_point, NeutronOVNMetadataAgentService)
 
diff --git a/snap/snapcraft.yaml b/snap/snapcraft.yaml
index 3d4d415..457d2a8 100644
--- a/snap/snapcraft.yaml
+++ b/snap/snapcraft.yaml
@@ -263,6 +263,8 @@ apps:
   neutron-ovn-metadata-agent:
     command: 'bin/neutron-ovn-metadata-agent-service'
     daemon: simple
+    restart-condition: on-failure
+    restart-delay: 7s
     plugs:
       - network
       - network-bind
diff --git a/tests/unit/test_hooks.py b/tests/unit/test_hooks.py
index 07846c2..2e2d6e1 100644
--- a/tests/unit/test_hooks.py
+++ b/tests/unit/test_hooks.py
@@ -1431,6 +1431,30 @@ def test_does_not_enable_ovs_exporter_when_monitoring_disabled(self, snap):
         services["ovs-exporter"].start.assert_not_called()
 
 
+class TestEnsureInternalOVSDependentServices:
+    """Tests for _ensure_internal_ovs_dependent_services function."""
+
+    def test_starts_non_excluded_services(self, snap):
+        services = {"neutron-ovn-metadata-agent": mock.Mock()}
+        snap.services.list.return_value = services
+        helper = getattr(hooks, "_ensure_internal_ovs_dependent_services", None)
+
+        assert helper is not None
+        helper(snap, exclude_services=[])
+
+        services["neutron-ovn-metadata-agent"].start.assert_called_once_with(enable=True)
+
+    def test_skips_excluded_services(self, snap):
+        services = {"neutron-ovn-metadata-agent": mock.Mock()}
+        snap.services.list.return_value = services
+        helper = getattr(hooks, "_ensure_internal_ovs_dependent_services", None)
+
+        assert helper is not None
+        helper(snap, exclude_services=["neutron-ovn-metadata-agent"])
+
+        services["neutron-ovn-metadata-agent"].start.assert_not_called()
+
+
 class TestInternalOVSReady:
     """Tests for internal OVS readiness detection."""
 
@@ -1601,6 +1625,12 @@ def test_internal_ovs_not_ready_defers_ovs_configuration(self, mocker, snap):
             "_ensure_internal_ovs_services",
             side_effect=lambda *_: order.append("ensure"),
         )
+        mocker.patch.object(
+            hooks,
+            "_ensure_internal_ovs_dependent_services",
+            side_effect=lambda *_: order.append("metadata"),
+            create=True,
+        )
         # Simulate charm already configured (real identity URL) so the OVS
         # startup guard does not interfere with what this test is checking.
         snap.config.get_options.return_value.get.return_value = "http://10.0.0.1:5000/v3"
@@ -1608,7 +1638,7 @@ def test_internal_ovs_not_ready_defers_ovs_configuration(self, mocker, snap):
         hooks.configure(snap)
 
         services["svc1"].stop.assert_called_once_with(disable=True)
-        assert order == ["tls", "ensure"]
+        assert order == ["tls", "ensure", "metadata"]
 
     def test_internal_ovs_ready_runs_configuration(self, mocker, snap):
         """Internal OVS configuration runs when the services are already ready."""
@@ -1642,13 +1672,19 @@ def test_internal_ovs_ready_runs_configuration(self, mocker, snap):
             "_ensure_internal_ovs_services",
             side_effect=lambda *_: order.append("ensure"),
         )
+        mocker.patch.object(
+            hooks,
+            "_ensure_internal_ovs_dependent_services",
+            side_effect=lambda *_: order.append("metadata"),
+            create=True,
+        )
         # Simulate charm already configured (real identity URL) so the OVS
         # startup guard does not interfere with what this test is checking.
         snap.config.get_options.return_value.get.return_value = "http://10.0.0.1:5000/v3"
 
         hooks.configure(snap)
 
-        assert order == ["tls", "network", "ensure"]
+        assert order == ["tls", "network", "ensure", "metadata"]
 
     def test_external_ovs_skips_internal_deferral_and_enable(self, mocker, snap):
         """External OVS never triggers internal OVS bootstrap or enablement."""
@@ -1674,11 +1710,15 @@ def test_external_ovs_skips_internal_deferral_and_enable(self, mocker, snap):
         mocker.patch.object(hooks, "_configure_sriov_agent_service")
         mock_ready = mocker.patch.object(hooks, "_internal_ovs_ready")
         mock_ensure = mocker.patch.object(hooks, "_ensure_internal_ovs_services")
+        mock_metadata = mocker.patch.object(
+            hooks, "_ensure_internal_ovs_dependent_services", create=True
+        )
 
         hooks.configure(snap)
 
         mock_ready.assert_not_called()
         mock_ensure.assert_not_called()
+        mock_metadata.assert_not_called()
 
     def test_external_ovs_deferred_when_microovn_not_installed(self, mocker, snap):
         """When microovn is not yet installed, OVS/OVN configuration is deferred."""
@@ -1756,6 +1796,9 @@ def test_internal_ovs_not_started_on_unconfigured_first_run(self, mocker, snap):
         mocker.patch.object(hooks, "_configure_masakari_services")
         mocker.patch.object(hooks, "_configure_sriov_agent_service")
         mock_ensure = mocker.patch.object(hooks, "_ensure_internal_ovs_services")
+        mock_metadata = mocker.patch.object(
+            hooks, "_ensure_internal_ovs_dependent_services", create=True
+        )
 
         # Simulate snap not yet configured by the charm: identity URL is the placeholder
         # AND username has not been set yet (None).  Both conditions must hold for the
@@ -1774,6 +1817,7 @@ def unconfigured_identity_get(key, default=None):
         # _ensure_internal_ovs_services must NOT have been called — starting
         # ovs-vswitchd here would create system@ovs-system and block microovn.
         mock_ensure.assert_not_called()
+        mock_metadata.assert_not_called()
 
     def test_internal_ovs_started_when_managed_by_hypervisor_with_default_identity(
         self, mocker, snap
@@ -1814,6 +1858,9 @@ def test_internal_ovs_started_when_managed_by_hypervisor_with_default_identity(
         mocker.patch.object(hooks, "_configure_masakari_services")
         mocker.patch.object(hooks, "_configure_sriov_agent_service")
         mock_ensure = mocker.patch.object(hooks, "_ensure_internal_ovs_services")
+        mock_metadata = mocker.patch.object(
+            hooks, "_ensure_internal_ovs_dependent_services", create=True
+        )
         # Identity is still unconfigured (placeholder URL, no username) — the guard
         # must be bypassed because mode is explicitly 'hypervisor'.
         snap.config.get_options.return_value.get.return_value = hooks.DEFAULT_CONFIG[
@@ -1823,6 +1870,7 @@ def test_internal_ovs_started_when_managed_by_hypervisor_with_default_identity(
         hooks.configure(snap)
 
         mock_ensure.assert_called_once()
+        mock_metadata.assert_called_once()
 
     def test_internal_ovs_started_when_identity_configured(self, mocker, snap):
         """Internal OVS must start once the charm has provided a real identity URL.
@@ -1852,12 +1900,16 @@ def test_internal_ovs_started_when_identity_configured(self, mocker, snap):
         mocker.patch.object(hooks, "_configure_masakari_services")
         mocker.patch.object(hooks, "_configure_sriov_agent_service")
         mock_ensure = mocker.patch.object(hooks, "_ensure_internal_ovs_services")
+        mock_metadata = mocker.patch.object(
+            hooks, "_ensure_internal_ovs_dependent_services", create=True
+        )
         # Charm has provided the real Keystone URL.
         snap.config.get_options.return_value.get.return_value = "http://10.0.0.1:5000/v3"
 
         hooks.configure(snap)
 
         mock_ensure.assert_called_once()
+        mock_metadata.assert_called_once()
 
 
 class TestDPDKConfigReady:
diff --git a/tests/unit/test_services.py b/tests/unit/test_services.py
index ed6107f..e772f11 100644
--- a/tests/unit/test_services.py
+++ b/tests/unit/test_services.py
@@ -2,13 +2,16 @@
 # SPDX-License-Identifier: Apache-2.0
 import base64
 import os
+import subprocess
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
 
+import openstack_hypervisor.services as services_module
 from openstack_hypervisor.services import (
     FileTransferService,
+    NeutronOVNMetadataAgentService,
     NovaComputeService,
 )
 
@@ -85,7 +88,7 @@ def test_returns_1_when_tls_invalid_base64(self, snap, config_file):
     @patch("openstack_hypervisor.services.os.set_inheritable")
     @patch("openstack_hypervisor.services.os.lseek")
     @patch("openstack_hypervisor.services.os.write")
-    @patch("openstack_hypervisor.services.os.memfd_create", return_value=[10, 11, 12])
+    @patch("openstack_hypervisor.services.os.memfd_create", create=True, return_value=[10, 11, 12])
     @patch("openstack_hypervisor.services.os.dup2")
     @patch("openstack_hypervisor.services.os.close")
     def test_returns_1_when_config_missing(
@@ -102,7 +105,7 @@ def test_returns_1_when_config_missing(
     @patch("openstack_hypervisor.services.os.set_inheritable")
     @patch("openstack_hypervisor.services.os.lseek")
     @patch("openstack_hypervisor.services.os.write")
-    @patch("openstack_hypervisor.services.os.memfd_create", return_value=[10, 11, 12])
+    @patch("openstack_hypervisor.services.os.memfd_create", create=True, return_value=[10, 11, 12])
     def test_success_path(
         self,
         mock_memfd,
@@ -162,3 +165,119 @@ def test_success_path(
         assert cmd[sep + 1] == str(tls_config.paths.snap / "usr" / "sbin" / "apache2")
         assert cmd[-1] == "-DFOREGROUND"
         assert "/proc/self/fd/6" in cmd
+
+
+class TestNeutronOVNMetadataAgentService:
+    """Tests for NeutronOVNMetadataAgentService."""
+
+    def _write_metadata_config(self, snap, ovsdb_connection):
+        config = snap.paths.common / "etc" / "neutron" / "neutron_ovn_metadata_agent.ini"
+        config.parent.mkdir(parents=True, exist_ok=True)
+        config.write_text(f"[ovs]\novsdb_connection = {ovsdb_connection}\n")
+        return config
+
+    @patch("openstack_hypervisor.services.subprocess.run")
+    def test_waits_for_ovsdb_schema_before_starting_agent(
+        self,
+        mock_run,
+        snap,
+        tmp_path,
+        monkeypatch,
+    ):
+        """Service should verify the local OVSDB schema before starting."""
+        monkeypatch.setattr(services_module, "OVSDB_SCHEMA_TIMEOUT", 0, raising=False)
+        monkeypatch.setattr(services_module, "OVSDB_SCHEMA_CHECK_INTERVAL", 0, raising=False)
+        ovs_socket = tmp_path / "db.sock"
+        ovs_socket.touch()
+        self._write_metadata_config(snap, f"unix:{ovs_socket}")
+        mock_run.side_effect = [
+            MagicMock(returncode=0),
+            MagicMock(returncode=0),
+        ]
+
+        result = NeutronOVNMetadataAgentService().run(snap)
+
+        assert result == 0
+        assert mock_run.call_count == 2
+        mock_run.assert_any_call(
+            [
+                str(snap.paths.snap / "usr" / "bin" / "ovsdb-client"),
+                "get-schema",
+                f"unix:{ovs_socket}",
+                "Open_vSwitch",
+            ],
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        mock_run.assert_any_call(
+            [
+                str(snap.paths.snap / "usr" / "bin" / "neutron-ovn-metadata-agent"),
+                "--config-file",
+                str(snap.paths.common / "etc" / "neutron" / "neutron.conf"),
+                "--config-file",
+                str(snap.paths.common / "etc" / "neutron" / "neutron_ovn_metadata_agent.ini"),
+                "--config-dir",
+                str(snap.paths.common / "etc" / "neutron" / "neutron.conf.d"),
+            ]
+        )
+
+    @patch("openstack_hypervisor.services.subprocess.run")
+    def test_returns_1_when_ovsdb_connection_missing(self, mock_run, snap):
+        """Service should fail fast when ovsdb_connection is not configured."""
+        config = snap.paths.common / "etc" / "neutron" / "neutron_ovn_metadata_agent.ini"
+        config.parent.mkdir(parents=True, exist_ok=True)
+        config.write_text("[ovs]\n")
+
+        result = NeutronOVNMetadataAgentService().run(snap)
+
+        assert result == 1
+        mock_run.assert_not_called()
+
+    @patch("openstack_hypervisor.services.subprocess.run")
+    def test_returns_1_when_unix_socket_missing(
+        self,
+        mock_run,
+        snap,
+        tmp_path,
+        monkeypatch,
+    ):
+        """Service should not start before the configured unix socket exists."""
+        monkeypatch.setattr(services_module, "OVSDB_SCHEMA_TIMEOUT", 0, raising=False)
+        monkeypatch.setattr(services_module, "OVSDB_SCHEMA_CHECK_INTERVAL", 0, raising=False)
+        ovs_socket = tmp_path / "db.sock"
+        self._write_metadata_config(snap, f"unix:{ovs_socket}")
+
+        result = NeutronOVNMetadataAgentService().run(snap)
+
+        assert result == 1
+        mock_run.assert_not_called()
+
+    @patch("openstack_hypervisor.services.subprocess.run")
+    def test_returns_1_when_schema_probe_times_out(
+        self,
+        mock_run,
+        snap,
+        tmp_path,
+        monkeypatch,
+    ):
+        """Service should fail when OVSDB never serves the Open_vSwitch schema."""
+        monkeypatch.setattr(services_module, "OVSDB_SCHEMA_TIMEOUT", 0, raising=False)
+        monkeypatch.setattr(services_module, "OVSDB_SCHEMA_CHECK_INTERVAL", 0, raising=False)
+        ovs_socket = tmp_path / "db.sock"
+        ovs_socket.touch()
+        self._write_metadata_config(snap, f"unix:{ovs_socket}")
+        mock_run.return_value = MagicMock(returncode=1)
+
+        result = NeutronOVNMetadataAgentService().run(snap)
+
+        assert result == 1
+        mock_run.assert_called_once_with(
+            [
+                str(snap.paths.snap / "usr" / "bin" / "ovsdb-client"),
+                "get-schema",
+                f"unix:{ovs_socket}",
+                "Open_vSwitch",
+            ],
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )