diff --git a/poetry.lock b/poetry.lock
index e8098b5a4e..a0bb056c8c 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -1865,37 +1865,41 @@ testing = ["coverage", "pytest", "pytest-benchmark"]
 
 [[package]]
 name = "postgresql-charms-single-kernel"
-version = "16.3.1"
+version = "16.3.3"
 description = "Shared and reusable code for PostgreSQL-related charms"
 optional = false
-python-versions = "<4.0,>=3.8"
+python-versions = ">=3.8,<4.0"
 groups = ["main", "integration"]
 files = [
-    {file = "postgresql_charms_single_kernel-16.3.1-py3-none-any.whl", hash = "sha256:a1c3f32fb396e69421471d64f54a43968f134fe5538708fabe37139a9fae832e"},
-    {file = "postgresql_charms_single_kernel-16.3.1.tar.gz", hash = "sha256:48e028e873b00c877fd36a25ff6c1913a40d559cb4124aae888edd1262cdce3a"},
+    {file = "5a9fa05.zip", hash = "sha256:0081d9b0a413bb8122d788054c175aa459611b817232542d4c01d6aaf7e8c5e3"},
 ]
 
 [package.dependencies]
-charm-refresh = {version = "*", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
-charmlibs-interfaces-tls-certificates = {version = ">=1.8.3", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
-charmlibs-pathops = {version = ">=1.0.1", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
-charmlibs-rollingops = {version = ">=1.1.1", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
-charmlibs-snap = {version = ">=1.0.1", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"vm\""}
-data-platform-helpers = {version = ">=0.1.7", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
-httpx = {version = "*", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
-jinja2 = {version = ">=3.1.6", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
+charm-refresh = {version = "*", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
+charmlibs-interfaces-tls-certificates = {version = ">=1.8.3", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
+charmlibs-pathops = {version = ">=1.0.1", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
+charmlibs-rollingops = {version = ">=1.1.1", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
+charmlibs-snap = {version = ">=1.0.1", optional = true, markers = "python_version >= \"3.12\" and extra == \"vm\""}
+charmlibs-systemd = {version = ">=1.0.0.post0", optional = true, markers = "python_version >= \"3.12\" and extra == \"vm\""}
+data-platform-helpers = {version = ">=0.1.7", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
+httpx = {version = "*", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
+jinja2 = {version = ">=3.1.6", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
 ops = ">=2.0.0"
-psutil = {version = ">=7.2.2", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"vm\""}
-pydantic = {version = ">=2.0", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
-pysyncobj = {version = ">=0.3.15", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"vm\""}
-requests = {version = "*", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
+psutil = {version = ">=7.2.2", optional = true, markers = "python_version >= \"3.12\" and extra == \"vm\""}
+pydantic = {version = ">=2.0", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
+pysyncobj = {version = ">=0.3.15", optional = true, markers = "python_version >= \"3.12\" and extra == \"vm\""}
+requests = {version = "*", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
 tenacity = ">=9.0.0"
-tomli = {version = "*", optional = true, markers = "python_full_version >= \"3.12.0\" and extra == \"postgresql\""}
+tomli = {version = "*", optional = true, markers = "python_version >= \"3.12\" and extra == \"postgresql\""}
 
 [package.extras]
 db-driver = ["psycopg2 (>=2.9.10)"]
-postgresql = ["charm-refresh ; python_full_version >= \"3.12.0\"", "charmlibs-interfaces-tls-certificates (>=1.8.3) ; python_full_version >= \"3.12.0\"", "charmlibs-pathops (>=1.0.1) ; python_full_version >= \"3.12.0\"", "charmlibs-rollingops (>=1.1.1) ; python_full_version >= \"3.12.0\"", "data-platform-helpers (>=0.1.7) ; python_full_version >= \"3.12.0\"", "httpx ; python_full_version >= \"3.12.0\"", "jinja2 (>=3.1.6) ; python_full_version >= \"3.12.0\"", "pydantic (>=2.0) ; python_full_version >= \"3.12.0\"", "requests ; python_full_version >= \"3.12.0\"", "tomli ; python_full_version >= \"3.12.0\""]
-vm = ["charmlibs-snap (>=1.0.1) ; python_full_version >= \"3.12.0\"", "psutil (>=7.2.2) ; python_full_version >= \"3.12.0\"", "pysyncobj (>=0.3.15) ; python_full_version >= \"3.12.0\""]
+postgresql = ["charm-refresh ; python_version >= \"3.12\"", "charmlibs-interfaces-tls-certificates (>=1.8.3) ; python_version >= \"3.12\"", "charmlibs-pathops (>=1.0.1) ; python_version >= \"3.12\"", "charmlibs-rollingops (>=1.1.1) ; python_version >= \"3.12\"", "data-platform-helpers (>=0.1.7) ; python_version >= \"3.12\"", "httpx ; python_version >= \"3.12\"", "jinja2 (>=3.1.6) ; python_version >= \"3.12\"", "pydantic (>=2.0) ; python_version >= \"3.12\"", "requests ; python_version >= \"3.12\"", "tomli ; python_version >= \"3.12\""]
+vm = ["charmlibs-snap (>=1.0.1) ; python_version >= \"3.12\"", "charmlibs-systemd (>=1.0.0.post0) ; python_version >= \"3.12\"", "psutil (>=7.2.2) ; python_version >= \"3.12\"", "pysyncobj (>=0.3.15) ; python_version >= \"3.12\""]
+
+[package.source]
+type = "url"
+url = "https://github.com/canonical/postgresql-single-kernel-library/archive/5a9fa05.zip"
 
 [[package]]
 name = "prompt-toolkit"
@@ -3193,4 +3197,4 @@ h11 = ">=0.16.0,<1"
 [metadata]
 lock-version = "2.1"
 python-versions = ">=3.12,<4.0"
-content-hash = "71742651a72b3bcf766b893c729952fb30c097d13a1e08e4ccf9f86d111a9d0c"
+content-hash = "b84f2901f3732eba7616a6fe7c249d00b6de70efb92630990ca53e4e22b17a3d"
diff --git a/pyproject.toml b/pyproject.toml
index 441969572b..521c86338d 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -21,7 +21,7 @@ charm-refresh = "^3.1.0.2"
 charmlibs-snap = "^1.0.1"
 charmlibs-systemd = "^1.0.0.post0"
 charmlibs-interfaces-tls-certificates = "^1.8.3"
-postgresql-charms-single-kernel = {extras = ["postgresql", "vm"], version = "16.3.1"}
+postgresql-charms-single-kernel = {url = "https://github.com/canonical/postgresql-single-kernel-library/archive/5a9fa05.zip", extras = ["postgresql", "vm"]}
 
 [tool.poetry.group.charm-libs.dependencies]
 # data_platform_libs/v0/data_interfaces.py
diff --git a/src/charm.py b/src/charm.py
index 0b9626453f..7151466c71 100755
--- a/src/charm.py
+++ b/src/charm.py
@@ -109,10 +109,8 @@
     SNAP_USER,
     SPI_MODULE,
     SYSTEM_USERS,
-    TLS_CA_BUNDLE_FILE,
-    TLS_CA_FILE,
-    TLS_CERT_FILE,
-    TLS_KEY_FILE,
+    TLS_CLIENT_RELATION,
+    TLS_PEER_RELATION,
     TRACING_PROTOCOL,
     TRACING_RELATION_NAME,
     UNIT_SCOPE,
@@ -121,12 +119,13 @@
 )
 from single_kernel_postgresql.core.config import CharmConfig
 from single_kernel_postgresql.core.state import CharmState
+from single_kernel_postgresql.events.tls import TLS
 from single_kernel_postgresql.events.tls_transfer import TLSTransfer
 from single_kernel_postgresql.managers.cluster import ClusterManager
 from single_kernel_postgresql.managers.config import ConfigManager
 from single_kernel_postgresql.managers.patroni import PatroniManager
 from single_kernel_postgresql.managers.tls import TLSManager
-from single_kernel_postgresql.utils import label2name, new_password, render_file
+from single_kernel_postgresql.utils import label2name, new_password
 from single_kernel_postgresql.utils.postgresql import (
     ACCESS_GROUP_IDENTITY,
     ACCESS_GROUPS,
@@ -155,7 +154,6 @@
 )
 from constants import (
     MONITORING_SNAP_SERVICE,
-    PATRONI_CONF_PATH,
     PGBACKREST_MONITORING_SNAP_SERVICE,
     POSTGRESQL_DATA_DIR,
     RAFT_PARTNER_PREFIX,
@@ -167,7 +165,6 @@
 from ldap import PostgreSQLLDAP
 from relations.async_replication import PostgreSQLAsyncReplication
 from relations.postgresql_provider import PostgreSQLProvider
-from relations.tls import TLS
 from relations.watcher import PostgreSQLWatcherRelation
 from rotate_logs import RotateLogs
 
@@ -379,11 +376,10 @@ def __init__(self, *args):
 
         # TODO switch to the abstract class base
         # State
-        self.state = CharmState(charm=self, substrate=self.substrate)  # type: ignore
+        self.state = CharmState(charm=self, substrate=self.substrate)
 
         # Managers
         self.patroni_manager = PatroniManager(state=self.state, workload=self.workload)
-        self.tls_manager = TLSManager(state=self.state, workload=self.workload)
         self.cluster_manager = ClusterManager(state=self.state, workload=self.workload)
         self.config_manager = ConfigManager(state=self.state, workload=self.workload)
 
@@ -421,7 +417,31 @@ def __init__(self, *args):
         self.postgresql_client_relation = PostgreSQLProvider(self)
         self.backup = PostgreSQLBackups(self, "s3-parameters")
         self.ldap = PostgreSQLLDAP(self, "ldap")
-        self.tls = TLS(self, PEER_RELATION)
+        # TLS events handler owns the two cert requirers; build it before the TLS
+        # manager so the manager can constructor-inject them for its live-fetch getters.
+        self.tls = TLS(self, self.state)
+        self.tls_manager = TLSManager(
+            state=self.state,
+            workload=self.workload,
+            client_certificate=self.tls.client_certificate,
+            peer_certificate=self.tls.peer_certificate,
+        )
+        # Bridge the lib TLS handler's requirer events back into a PostgreSQL reload:
+        # the lib handler stores+pushes certs on certificate_available, then we reload.
+        # Also fires on relation_broken so detaching the TLS operator re-renders Patroni
+        # with TLS disabled.
+        self.framework.observe(
+            self.tls.client_certificate.on.certificate_available, self._reload_tls_after_push
+        )
+        self.framework.observe(
+            self.tls.peer_certificate.on.certificate_available, self._reload_tls_after_push
+        )
+        self.framework.observe(
+            self.on[TLS_CLIENT_RELATION].relation_broken, self._reload_tls_after_push
+        )
+        self.framework.observe(
+            self.on[TLS_PEER_RELATION].relation_broken, self._reload_tls_after_push
+        )
         self.tls_transfer = TLSTransfer(self, PEER_RELATION)
         self.async_replication = PostgreSQLAsyncReplication(self)
         self.watcher_offer = PostgreSQLWatcherRelation(self)
@@ -498,6 +518,49 @@ def substrate(self) -> Substrates:
         """
         return Substrates.VM
 
+    def _reload_tls_after_push(self, event) -> None:
+        """Reload PostgreSQL after the lib TLS handler stores+pushes certs.
+
+        Also fires on the TLS relations' relation_broken so detaching the TLS
+        operator re-renders Patroni with TLS disabled and reloads.
+
+        Mirror the handler's readiness guard: when the internal CA is absent the
+        handler defers its push (no files on disk), so skip the reload to avoid
+        rendering ssl:on against missing TLS files on an already-running unit.
+
+        A transient config-apply failure (Patroni API unreachable, member not
+        started) defers and retries rather than leaving stale TLS state or failing
+        the hook.
+        """
+        if not self.get_secret(APP_SCOPE, "internal-ca"):
+            return
+        # Don't enable TLS in the config until the lib has written the cert files to
+        # disk (its push can defer while this local render would still succeed, which
+        # would start Patroni ssl:on against missing files).
+        if self.is_tls_enabled and not self.tls_manager.client_tls_files_on_disk():
+            event.defer()
+            return
+        try:
+            if not self.update_config():
+                event.defer()
+        except Exception:
+            logger.exception("TLS reload (update_config) failed; deferring")
+            event.defer()
+
+    def _regenerate_internal_cert(self, *, reload: bool = True) -> None:
+        """Generate the internal peer cert, push it to the workload, and (optionally) reload.
+
+        reload=False is used at cluster bootstrap: the leader renders patroni.yml on
+        leader-elected and each replica renders it in _on_peer_relation_changed just
+        before starting Patroni, so a reload here would be redundant. The internal peer
+        cert does not toggle ssl in the config -- only the operator/client cert does, via
+        is_tls_enabled -- so skipping the reload cannot leave a stale ssl setting.
+        """
+        self.tls_manager.generate_internal_peer_cert()
+        self.tls_manager.push_tls_files()
+        if reload:
+            self.update_config()
+
     def _check_and_update_internal_cert(self) -> None:
         """Check if the internal cert CN matches the unit IP and regenerate if needed."""
         try:
@@ -509,7 +572,7 @@ def _check_and_update_internal_cert(self) -> None:
                     != self.state.unit_ip
                 )
             ):
-                self.tls.generate_internal_peer_cert()
+                self._regenerate_internal_cert()
         except Exception:
             logger.exception("Unable to check or update internal cert")
 
@@ -1548,7 +1611,7 @@ def is_standby_cluster(self) -> bool:
     @property
     def is_tls_enabled(self) -> bool:
         """Return whether TLS is enabled."""
-        return all(self.tls.get_client_tls_files())
+        return all(self.tls_manager.get_client_tls_files())
 
     @property
     def _peer_members_ips(self) -> set[str]:
@@ -1759,7 +1822,7 @@ def _on_leader_elected(self, event: LeaderElectedEvent) -> None:  # noqa: C901
             logger.debug("On leader elected failed to reconfigure cluster.")
 
         if not self.get_secret(APP_SCOPE, "internal-ca"):
-            self.tls.generate_internal_peer_ca()
+            self.tls_manager.generate_internal_peer_ca()
         self.update_config()
 
         # Don't update connection endpoints in the first time this event run for
@@ -1946,7 +2009,7 @@ def _on_start(self, event: StartEvent) -> None:
             event.defer()
             return
         if not self.get_secret(UNIT_SCOPE, "internal-cert"):
-            self.tls.generate_internal_peer_cert()
+            self._regenerate_internal_cert(reload=False)
 
         self.unit_peer_data.update({"ip": self.state.unit_ip})
         self._ensure_storage_layout()
@@ -2497,10 +2560,12 @@ def _update_certificate(self) -> None:
         # Request the certificate only if there is already one. If there isn't,
         # the certificate will be generated in the relation joined event when
         # relating to the TLS Certificates Operator.
-        if all(self.tls.get_client_tls_files()) or all(self.tls.get_peer_tls_files()):
+        if all(self.tls_manager.get_client_tls_files()) or all(
+            self.tls_manager.get_peer_tls_files()
+        ):
             self.tls.refresh_tls_certificates_event.emit()
         if self.get_secret(UNIT_SCOPE, "internal-cert"):
-            self.tls.generate_internal_peer_cert()
+            self._regenerate_internal_cert()
 
     @property
     def is_blocked(self) -> bool:
@@ -2582,37 +2647,6 @@ def _peers(self) -> Relation | None:
         """
         return self.model.get_relation(PEER_RELATION)
 
-    def push_tls_files_to_workload(self) -> bool:
-        """Move TLS files to the PostgreSQL storage path and enable TLS."""
-        key, ca, cert = self.tls.get_client_tls_files()
-        if key is not None:
-            render_file(Substrates.VM, f"{PATRONI_CONF_PATH}/{TLS_KEY_FILE}", key, 0o600)
-        if ca is not None:
-            render_file(Substrates.VM, f"{PATRONI_CONF_PATH}/{TLS_CA_FILE}", ca, 0o600)
-        if cert is not None:
-            render_file(Substrates.VM, f"{PATRONI_CONF_PATH}/{TLS_CERT_FILE}", cert, 0o600)
-
-        key, ca, cert = self.tls.get_peer_tls_files()
-        if key is not None:
-            render_file(Substrates.VM, f"{PATRONI_CONF_PATH}/peer_{TLS_KEY_FILE}", key, 0o600)
-        if ca is not None:
-            render_file(Substrates.VM, f"{PATRONI_CONF_PATH}/peer_{TLS_CA_FILE}", ca, 0o600)
-        if cert is not None:
-            render_file(Substrates.VM, f"{PATRONI_CONF_PATH}/peer_{TLS_CERT_FILE}", cert, 0o600)
-
-        render_file(
-            Substrates.VM,
-            f"{PATRONI_CONF_PATH}/{TLS_CA_BUNDLE_FILE}",
-            self.tls.get_peer_ca_bundle(),
-            0o600,
-        )
-
-        try:
-            return self.update_config()
-        except Exception:
-            logger.exception("TLS files failed to push. Error in config update")
-            return False
-
     def push_ca_file_into_workload(self, secret_name: str) -> bool:
         """Move CA certificates file into the PostgreSQL storage path."""
         certs = self.get_secret(UNIT_SCOPE, secret_name)
diff --git a/src/relations/postgresql_provider.py b/src/relations/postgresql_provider.py
index 7b5f0bd699..a0e47d79aa 100644
--- a/src/relations/postgresql_provider.py
+++ b/src/relations/postgresql_provider.py
@@ -485,7 +485,7 @@ def update_endpoints(self, event: DatabaseRequestedEvent | None = None) -> None:
         tls = "True" if self.charm.is_tls_enabled else "False"
         ca = None
         if tls == "True":
-            _, ca, _ = self.charm.tls.get_client_tls_files()
+            _, ca, _ = self.charm.tls_manager.get_client_tls_files()
         if not ca:
             ca = ""
 
diff --git a/src/relations/tls.py b/src/relations/tls.py
deleted file mode 100644
index 2c3bc92b5a..0000000000
--- a/src/relations/tls.py
+++ /dev/null
@@ -1,259 +0,0 @@
-# Copyright 2022 Canonical Ltd.
-# See LICENSE file for licensing details.
-
-"""TLS Handler."""
-
-import logging
-import socket
-from datetime import timedelta
-from typing import TYPE_CHECKING
-
-from charmlibs.interfaces.tls_certificates import (
-    Certificate,
-    CertificateRequestAttributes,
-    PrivateKey,
-    TLSCertificatesRequiresV4,
-    generate_ca,
-    generate_certificate,
-    generate_csr,
-    generate_private_key,
-)
-from ops import (
-    EventSource,
-)
-from ops.framework import EventBase, Object
-from ops.pebble import ConnectionError as PebbleConnectionError
-from ops.pebble import PathError, ProtocolError
-from single_kernel_postgresql.config.literals import (
-    APP_SCOPE,
-    UNIT_SCOPE,
-)
-from tenacity import RetryError
-
-if TYPE_CHECKING:
-    from charm import PostgresqlOperatorCharm
-
-logger = logging.getLogger(__name__)
-SCOPE = "unit"
-TLS_CLIENT_RELATION = "client-certificates"
-TLS_PEER_RELATION = "peer-certificates"
-
-
-class RefreshTLSCertificatesEvent(EventBase):
-    """Event for refreshing TLS certificates."""
-
-
-class TlsError(Exception):
-    """TLS implementation internal exception."""
-
-
-class TLS(Object):
-    """In this class we manage certificates relation."""
-
-    refresh_tls_certificates_event = EventSource(RefreshTLSCertificatesEvent)
-
-    def _get_client_addrs(self) -> set[str]:
-        client_addrs = set()
-        if addr := self.charm.unit_peer_data.get("database-address"):
-            client_addrs.add(addr)
-        return client_addrs
-
-    def _get_peer_addrs(self) -> set[str]:
-        peer_addrs = set()
-        if addr := self.charm.unit_peer_data.get("database-peers-address"):
-            peer_addrs.add(addr)
-        if addr := self.charm.unit_peer_data.get("replication-address"):
-            peer_addrs.add(addr)
-        if addr := self.charm.unit_peer_data.get("replication-offer-address"):
-            peer_addrs.add(addr)
-        if addr := self.charm.unit_peer_data.get(
-            "ip", self.charm.unit_peer_data.get("private-address")
-        ):
-            peer_addrs.add(addr)
-        return peer_addrs
-
-    def _get_common_name(self) -> str:
-        return self.charm.unit_peer_data.get("database-address") or self.host
-
-    def _get_peer_common_name(self) -> str:
-        return self.charm.unit_peer_data.get("database-peers-address") or self.host
-
-    def __init__(self, charm: "PostgresqlOperatorCharm", peer_relation: str):
-        super().__init__(charm, "client-relations")
-        self.charm = charm
-        self.peer_relation = peer_relation
-        unit_id = self.charm.unit.name.split("/")[1]
-        self.host = f"{self.charm.app.name}-{unit_id}"
-        if self.charm.unit_peer_data:
-            client_addresses = self._get_client_addrs()
-            peer_addresses = self._get_peer_addrs()
-        else:
-            client_addresses = set()
-            peer_addresses = set()
-        self.common_hosts = {self.host}
-        if fqdn := socket.getfqdn():
-            self.common_hosts.add(fqdn)
-
-        self.client_certificate = TLSCertificatesRequiresV4(
-            self.charm,
-            TLS_CLIENT_RELATION,
-            certificate_requests=[
-                CertificateRequestAttributes(
-                    common_name=self._get_common_name(),
-                    sans_ip=frozenset(client_addresses),
-                    sans_dns=frozenset({
-                        *self.common_hosts,
-                        # IP address need to be part of the DNS SANs list due to
-                        # https://github.com/pgbackrest/pgbackrest/issues/1977.
-                        *client_addresses,
-                    }),
-                ),
-            ],
-            refresh_events=[self.refresh_tls_certificates_event],
-        )
-        self.peer_certificate = TLSCertificatesRequiresV4(
-            self.charm,
-            TLS_PEER_RELATION,
-            certificate_requests=[
-                CertificateRequestAttributes(
-                    common_name=self._get_peer_common_name(),
-                    sans_ip=frozenset(self._get_peer_addrs()),
-                    sans_dns=frozenset({
-                        *self.common_hosts,
-                        # IP address need to be part of the DNS SANs list due to
-                        # https://github.com/pgbackrest/pgbackrest/issues/1977.
-                        *peer_addresses,
-                    }),
-                ),
-            ],
-            refresh_events=[self.refresh_tls_certificates_event],
-        )
-
-        self.framework.observe(
-            self.client_certificate.on.certificate_available, self._on_certificate_available
-        )
-        self.framework.observe(
-            self.peer_certificate.on.certificate_available, self._on_peer_certificate_available
-        )
-
-        self.framework.observe(
-            self.charm.on[TLS_CLIENT_RELATION].relation_broken, self._on_certificate_available
-        )
-        self.framework.observe(
-            self.charm.on[TLS_PEER_RELATION].relation_broken, self._on_peer_certificate_available
-        )
-
-    def _on_peer_certificate_available(self, event: EventBase) -> None:
-        certs, _ = self.peer_certificate.get_assigned_certificates()
-        new_ca = str(certs[0].ca) if certs else None
-        current_ca = self.charm.get_secret(UNIT_SCOPE, "current-ca")
-        # Stash the CAs in case of rotation
-        if new_ca != current_ca:
-            self.charm.set_secret(UNIT_SCOPE, "current-ca", new_ca)
-            self.charm.set_secret(UNIT_SCOPE, "old-ca", current_ca)
-        self._on_certificate_available(event)
-
-    def _on_certificate_available(self, event: EventBase) -> None:
-        if not self.charm.get_secret(APP_SCOPE, "internal-ca"):
-            logger.debug("Charm not ready yet")
-            event.defer()
-            return
-        try:
-            if not self.charm.push_tls_files_to_workload():
-                logger.debug("Cannot push TLS certificates at this moment")
-                event.defer()
-                return
-        except (PebbleConnectionError, PathError, ProtocolError, RetryError) as e:
-            logger.error("Cannot push TLS certificates: %r", e)
-            event.defer()
-            return
-
-    def get_client_tls_files(self) -> tuple[str | None, str | None, str | None]:
-        """Prepare TLS files in special PostgreSQL way.
-
-        PostgreSQL needs three files:
-        — CA file should have a full chain.
-        — Key file should have private key.
-        — Certificate file should have certificate without certificate chain.
-        """
-        ca_file = None
-        cert = None
-        key = None
-        certs, private_key = self.client_certificate.get_assigned_certificates()
-        if private_key:
-            key = str(private_key)
-        if certs:
-            cert = str(certs[0].certificate)
-            ca_file = str(certs[0].ca)
-        return key, ca_file, cert
-
-    def get_peer_tls_files(self) -> tuple[str | None, str | None, str | None]:
-        """Prepare TLS files in special PostgreSQL way.
-
-        PostgreSQL needs three files:
-        — CA file should have a full chain.
-        — Key file should have private key.
-        — Certificate file should have certificate without certificate chain.
-        """
-        ca_file = None
-        cert = None
-        key = None
-        certs, private_key = self.peer_certificate.get_assigned_certificates()
-        if private_key:
-            key = str(private_key)
-        if certs:
-            cert = str(certs[0].certificate)
-            ca_file = str(certs[0].ca)
-        if not all((key, ca_file, cert)):
-            key = self.charm.get_secret(UNIT_SCOPE, "internal-key")
-            cert = self.charm.get_secret(UNIT_SCOPE, "internal-cert")
-            ca_file = self.charm.get_secret(APP_SCOPE, "internal-ca")
-        return key, ca_file, cert
-
-    def get_peer_ca_bundle(self) -> str:
-        """Get bundled CA certs."""
-        certs, _ = self.peer_certificate.get_assigned_certificates()
-        operator_ca = str(certs[0].ca) if certs else ""
-        old_operator_ca = self.charm.get_secret(UNIT_SCOPE, "old-ca") or ""
-        internal_ca = self.charm.get_secret(APP_SCOPE, "internal-ca") or ""
-        return "\n".join((operator_ca, old_operator_ca, internal_ca)).strip()
-
-    def generate_internal_peer_ca(self) -> None:
-        """Generate internal peer CA using the tls lib."""
-        private_key = generate_private_key()
-        ca = generate_ca(
-            private_key,
-            common_name=f"{self.charm.app.name}-{self.charm.model.uuid}",
-            validity=timedelta(days=7300),
-        )
-        logger.warning("Internal peer CA generated. Please use a proper TLS operator if possible.")
-        self.charm.set_secret(APP_SCOPE, "internal-ca-key", str(private_key))
-        self.charm.set_secret(APP_SCOPE, "internal-ca", str(ca))
-
-    def generate_internal_peer_cert(self) -> None:
-        """Generate internal peer certificate using the tls lib."""
-        if not (ca_key_secret := self.charm.get_secret(APP_SCOPE, "internal-ca-key")):
-            raise TlsError("No CA key content.")
-        ca_key = PrivateKey.from_string(ca_key_secret)
-        if not (ca_secret := self.charm.get_secret(APP_SCOPE, "internal-ca")):
-            raise TlsError("No CA cert content.")
-        ca = Certificate.from_string(ca_secret)
-        private_key = generate_private_key()
-        csr = generate_csr(
-            private_key,
-            common_name=self._get_peer_common_name(),
-            sans_ip=frozenset(self._get_peer_addrs()),
-            sans_dns=frozenset({
-                *self.common_hosts,
-                # IP address need to be part of the DNS SANs list due to
-                # https://github.com/pgbackrest/pgbackrest/issues/1977.
-                *self._get_peer_addrs(),
-            }),
-        )
-        cert = generate_certificate(csr, ca, ca_key, validity=timedelta(days=7300))
-        self.charm.set_secret(UNIT_SCOPE, "internal-key", str(private_key))
-        self.charm.set_secret(UNIT_SCOPE, "internal-cert", str(cert))
-        self.charm.push_tls_files_to_workload()
-        logger.info(
-            "Internal peer certificate generated. Please use a proper TLS operator if possible."
-        )
diff --git a/src/relations/watcher.py b/src/relations/watcher.py
index 9cc92a3bef..4f46e5f760 100644
--- a/src/relations/watcher.py
+++ b/src/relations/watcher.py
@@ -419,7 +419,7 @@ def _update_relation_data(self, relation: Relation) -> None:
             "raft-secret-id": secret_id,
             "raft-partner-addrs": json.dumps(pg_endpoints),
             "raft-port": str(RAFT_PORT),
-            "patroni-cas": self.charm.tls.get_peer_ca_bundle(),
+            "patroni-cas": self.charm.tls_manager.get_peer_ca_bundle(),
             "standby-clusters": json.dumps(self._get_standby_clusters()),
             "tls-enabled": "true" if self.charm.is_tls_enabled else "false",
         })
diff --git a/tests/unit/test_charm.py b/tests/unit/test_charm.py
index 7a9db219e5..9e6c31b7a8 100644
--- a/tests/unit/test_charm.py
+++ b/tests/unit/test_charm.py
@@ -188,7 +188,7 @@ def test_on_leader_elected(harness):
         ) as _primary_endpoint,
         patch("charm.PostgresqlOperatorCharm.update_config") as _update_config,
         patch("charm.PostgresqlOperatorCharm._reconfigure_cluster"),
-        patch("charm.TLS.generate_internal_peer_cert"),
+        patch("charm.TLSManager.generate_internal_peer_cert"),
     ):
         # Assert that there is no password in the peer relation.
         assert harness.charm._peers.data[harness.charm.app].get("operator-password", None) is None
@@ -444,7 +444,7 @@ def test_on_start_bootstrap_failure(harness):
         patch("charm.PostgresqlOperatorCharm._ensure_storage_layout"),
         patch("charm.PostgresqlOperatorCharm._check_detached_storage"),
         patch("charm.PostgresqlOperatorCharm.get_secret"),
-        patch("charm.TLS.generate_internal_peer_cert"),
+        patch("charm.TLSManager.generate_internal_peer_cert"),
         patch("charm.PatroniManager.bootstrap_cluster") as _bootstrap_cluster,
         patch("charm.PostgresqlOperatorCharm.update_config"),
         patch("charm.start_raft_observer"),
@@ -475,7 +475,7 @@ def test_on_start_create_user_error(harness):
         patch("charm.PostgresqlOperatorCharm._ensure_storage_layout"),
         patch("charm.PostgresqlOperatorCharm._check_detached_storage"),
         patch("charm.PostgresqlOperatorCharm.get_secret"),
-        patch("charm.TLS.generate_internal_peer_cert"),
+        patch("charm.TLSManager.generate_internal_peer_cert"),
         patch("charm.PatroniManager.bootstrap_cluster") as _bootstrap_cluster,
         patch(
             "charm.PatroniManager.member_started",
@@ -532,7 +532,7 @@ def test_on_start_success(harness):
         patch("charm.PostgresqlOperatorCharm._ensure_storage_layout"),
         patch("charm.PostgresqlOperatorCharm._check_detached_storage"),
         patch("charm.PostgresqlOperatorCharm.get_secret"),
-        patch("charm.TLS.generate_internal_peer_cert"),
+        patch("charm.TLSManager.generate_internal_peer_cert"),
         patch("charm.PatroniManager.bootstrap_cluster") as _bootstrap_cluster,
         patch(
             "charm.PatroniManager.member_started",
@@ -661,7 +661,7 @@ def test_on_start_replica(harness):
             return_value=True,
         ) as _is_storage_attached,
         patch("charm.PostgresqlOperatorCharm.get_secret"),
-        patch("charm.TLS.generate_internal_peer_cert"),
+        patch("charm.TLSManager.generate_internal_peer_cert"),
         patch("charm.start_raft_observer"),
     ):
         _get_postgresql_version.return_value = "16.6"
@@ -720,7 +720,7 @@ def test_on_start_no_patroni_member(harness):
         ) as _is_storage_attached,
         patch("charm.PostgresqlOperatorCharm.get_available_memory") as _get_available_memory,
         patch("charm.PostgresqlOperatorCharm.get_secret"),
-        patch("charm.TLS.generate_internal_peer_cert"),
+        patch("charm.TLSManager.generate_internal_peer_cert"),
         patch("charm.PostgreSQLProvider.get_username_mapping", return_value={}),
         patch("charm.PostgreSQLProvider.get_databases_prefix_mapping", return_value={}),
         patch("charm.start_raft_observer"),
@@ -2344,7 +2344,7 @@ def test_reconfigure_cluster(harness):
 
 def test_update_certificate(harness):
     with (
-        patch("charm.TLS.get_client_tls_files") as _get_client_tls_files,
+        patch("charm.TLSManager.get_client_tls_files") as _get_client_tls_files,
         patch("charm.TLS.refresh_tls_certificates_event") as _refresh_tls_certificates_event,
     ):
         # If there is no current TLS files, _request_certificate should be called
@@ -2402,41 +2402,6 @@ def test_update_member_ip(harness):
         _update_certificate.assert_called_once()
 
 
-def test_push_tls_files_to_workload(harness):
-    with (
-        patch("charm.PostgresqlOperatorCharm.update_config") as _update_config,
-        patch("charm.render_file") as _render_file,
-        patch("charm.TLS.get_client_tls_files") as _get_client_tls_files,
-        patch("charm.TLS.get_peer_tls_files") as _get_peer_tls_files,
-        patch(
-            "charm.PostgresqlOperatorCharm.get_secret", return_value="internal_ca"
-        ) as _get_secret,
-    ):
-        _get_client_tls_files.side_effect = [
-            ("key", "ca", "cert"),
-            ("key", "ca", None),
-            ("key", None, "cert"),
-            (None, "ca", "cert"),
-        ]
-        _get_peer_tls_files.side_effect = [
-            ("key", "ca", "cert"),
-            ("key", "ca", None),
-            ("key", None, "cert"),
-            (None, "ca", "cert"),
-        ]
-        _update_config.side_effect = [True, False, False, False]
-
-        # Test when all TLS files are available.
-        assert harness.charm.push_tls_files_to_workload()
-        assert _render_file.call_count == 7
-
-        # Test when not all TLS files are available.
-        for _ in range(3):
-            _render_file.reset_mock()
-            assert not (harness.charm.push_tls_files_to_workload())
-            assert _render_file.call_count == 5
-
-
 def test_push_ca_file_into_workload(harness):
     with (
         patch("charm.PostgresqlOperatorCharm.update_config") as _update_config,
diff --git a/tests/unit/test_tls.py b/tests/unit/test_tls.py
index c8ce927080..0b38b69b07 100644
--- a/tests/unit/test_tls.py
+++ b/tests/unit/test_tls.py
@@ -1,172 +1,226 @@
 # Copyright 2022 Canonical Ltd.
 # See LICENSE file for licensing details.
-from datetime import timedelta
-from unittest.mock import Mock, patch, sentinel
+"""TLS wiring tests for the single-kernel (lib) TLS handler.
+
+The charm consumes the library TLS handler (single_kernel_postgresql.events.tls.TLS)
+and TLS manager (single_kernel_postgresql.managers.tls.TLSManager) instead of its
+own removed src/relations/tls.py. These tests exercise the lib-backed wiring:
+state-backed cert storage via TLSManager, and the charm's reload bridge that calls
+update_config after the handler stores+pushes certificates.
+"""
+
+from unittest.mock import Mock, PropertyMock, patch
 
 import pytest
 from ops.testing import Harness
-from single_kernel_postgresql.config.literals import PEER_RELATION
+from single_kernel_postgresql.config.literals import (
+    PEER_RELATION,
+    TLS_CLIENT_RELATION,
+    TLS_PEER_RELATION,
+)
+from single_kernel_postgresql.events.tls import TLS
+from single_kernel_postgresql.managers.tls import TLSManager
 
 from charm import PostgresqlOperatorCharm
 
 
 @pytest.fixture(autouse=True)
-def harness():
-    with patch("relations.tls.socket") as _socket:
-        _socket.getfqdn.return_value = "fqdn"
-        harness = Harness(PostgresqlOperatorCharm)
+def _platform_machine(monkeypatch):
+    # refresh_versions.toml keys snap revisions by x86_64/aarch64; ensure the charm
+    # constructs regardless of the host arch (e.g. arm64 dev laptops).
+    monkeypatch.setattr("platform.machine", lambda: "x86_64")
 
-        # Set up the initial relation and hooks.
-        peer_rel_id = harness.add_relation(PEER_RELATION, "postgresql")
-        harness.add_relation_unit(peer_rel_id, "postgresql/0")
-        harness.begin()
-        yield harness
-        harness.cleanup()
 
-
-def test_generate_internal_peer_cert(harness):
-    with (
-        patch(
-            "relations.tls.generate_private_key", return_value=sentinel.cert_key
-        ) as _generate_private_key,
-        patch("relations.tls.generate_csr", return_value=sentinel.cert_csr) as _generate_csr,
-        patch(
-            "relations.tls.generate_certificate", return_value=sentinel.cert
-        ) as _generate_certificate,
-        patch("relations.tls.PrivateKey") as _private_key,
-        patch("relations.tls.Certificate") as _certificate,
-        patch("charm.PostgresqlOperatorCharm.set_secret") as _set_secret,
-        patch("charm.PostgresqlOperatorCharm.get_secret", return_value="secret value"),
-        patch(
-            "charm.PostgresqlOperatorCharm.push_tls_files_to_workload"
-        ) as _push_tls_files_to_workload,
+@pytest.fixture(autouse=True)
+def harness(_platform_machine):
+    harness = Harness(PostgresqlOperatorCharm)
+    peer_rel_id = harness.add_relation(PEER_RELATION, "postgresql")
+    harness.add_relation_unit(peer_rel_id, "postgresql/0")
+    harness.begin()
+    yield harness
+    harness.cleanup()
+
+
+def test_tls_handler_is_lib_backed(harness):
+    """The charm wires the lib TLS handler + manager (not the removed relations.tls)."""
+    charm = harness.charm
+    assert isinstance(charm.tls, TLS)
+    assert isinstance(charm.tls_manager, TLSManager)
+    # The handler owns the operator client/peer requirers and the refresh event.
+    assert hasattr(charm.tls, "client_certificate")
+    assert hasattr(charm.tls, "peer_certificate")
+    assert hasattr(charm.tls, "refresh_tls_certificates_event")
+    # The removed method must not resurface anywhere.
+    assert not hasattr(charm, "push_tls_files_to_workload")
+
+
+def test_is_tls_enabled_reflects_tls_manager(harness):
+    """is_tls_enabled is driven by TLSManager.get_client_tls_files(), not the handler."""
+    with patch("charm.TLSManager.get_client_tls_files") as _get_client_tls_files:
+        _get_client_tls_files.return_value = (None, None, None)
+        assert harness.charm.is_tls_enabled is False
+
+        _get_client_tls_files.return_value = ("key", "ca", "cert")
+        assert harness.charm.is_tls_enabled is True
+
+
+def _observers_for(harness, bound_event):
+    """Return method names observing the given bound event, in registration order."""
+    emitter_path = bound_event.emitter.handle.path
+    event_kind = bound_event.event_kind
+    return [
+        method
+        for (_obs_path, method, e_path, e_kind) in harness.framework._observers
+        if e_path == emitter_path and e_kind == event_kind
+    ]
+
+
+def test_reload_bridge_wired_after_handler_on_client_certificate(harness):
+    """The reload bridge observes the same certificate_available event as the handler.
+
+    The lib handler's store+push observer must be registered BEFORE the charm's
+    reload bridge so that, when the event fires, certs are stored+pushed first and
+    the reload (update_config) runs afterwards (ops calls observers in order).
+    """
+    methods = _observers_for(
+        harness, harness.charm.tls.client_certificate.on.certificate_available
+    )
+    assert "_on_certificate_available" in methods, methods
+    assert "_reload_tls_after_push" in methods, methods
+    # Handler (store+push) before bridge (reload).
+    assert methods.index("_on_certificate_available") < methods.index("_reload_tls_after_push")
+
+
+def test_reload_bridge_wired_after_handler_on_peer_certificate(harness):
+    """The reload bridge also observes the peer certificate_available event."""
+    methods = _observers_for(harness, harness.charm.tls.peer_certificate.on.certificate_available)
+    assert "_on_peer_certificate_available" in methods, methods
+    assert "_reload_tls_after_push" in methods, methods
+    assert methods.index("_on_peer_certificate_available") < methods.index(
+        "_reload_tls_after_push"
+    )
+
+
+def test_reload_bridge_calls_update_config(harness):
+    """_reload_tls_after_push delegates to update_config when internal-ca is present."""
+    with harness.hooks_disabled():
+        harness.set_leader(True)
+        harness.charm.set_secret("app", "internal-ca", "ca-content")
+
+    with patch("charm.PostgresqlOperatorCharm.update_config") as _update_config:
+        harness.charm._reload_tls_after_push(Mock())
+        _update_config.assert_called_once_with()
+
+
+def test_reload_bridge_skips_update_config_without_internal_ca(harness):
+    """_reload_tls_after_push is a no-op when internal-ca is absent (defer path).
+
+    The lib TLS handler defers its push when the internal CA isn't present yet
+    (no files written to disk).  The bridge must not call update_config in that
+    case, or it would render ssl:on against TLS files that don't exist yet.
+    """
+    with patch("charm.PostgresqlOperatorCharm.update_config") as _update_config:
+        harness.charm._reload_tls_after_push(Mock())
+        _update_config.assert_not_called()
+
+
+def test_reload_bridge_wired_on_tls_relation_broken(harness):
+    """Detaching the TLS operator (relation_broken) must trigger the reload bridge.
+
+    Regression test: the VM charm previously observed only certificate_available, so
+    removing a TLS relation left Patroni ssl:on against cleared certs.  Both TLS
+    relations' relation_broken must reach the reload bridge (parity with K8s).
+
+    The ordering is also asserted: unlike the certificate_available wiring, these
+    registrations reference only ``self.on[...]``, so moving them above the lib
+    handler's construction in __init__ would silently reverse the order and run
+    the reload before the lib clears state and pushes files.
+    """
+    for relation, lib_handler in (
+        (TLS_CLIENT_RELATION, "_on_certificate_available"),
+        (TLS_PEER_RELATION, "_on_peer_certificate_available"),
     ):
-        _private_key.from_string.return_value = sentinel.ca_key
-        _certificate.from_string.return_value = sentinel.ca_cert
-
-        harness.charm.tls.generate_internal_peer_cert()
-
-        _generate_csr.assert_called_once_with(
-            sentinel.cert_key,
-            common_name="postgresql-0",
-            sans_ip=frozenset(),
-            sans_dns=frozenset({"postgresql-0", "fqdn"}),
+        methods = _observers_for(harness, harness.charm.on[relation].relation_broken)
+        assert lib_handler in methods, (relation, methods)
+        assert "_reload_tls_after_push" in methods, (relation, methods)
+        # Lib handler (clear+push) before charm bridge (reload).
+        assert methods.index(lib_handler) < methods.index("_reload_tls_after_push"), (
+            relation,
+            methods,
         )
-        _generate_certificate.assert_called_once_with(
-            sentinel.cert_csr, sentinel.ca_cert, sentinel.ca_key, validity=timedelta(days=7300)
-        )
-        assert _set_secret.call_count == 2
-        _set_secret.assert_any_call("unit", "internal-key", str(sentinel.cert_key))
-        _set_secret.assert_any_call("unit", "internal-cert", str(sentinel.cert))
-        _push_tls_files_to_workload.assert_called_once_with()
 
 
-def test_get_client_tls_files(harness):
-    with patch(
-        "relations.tls.TLSCertificatesRequiresV4.get_assigned_certificates"
-    ) as _get_assigned_certificates:
-        cert_mock = Mock()
-        cert_mock.certificate = sentinel.certificate
-        cert_mock.ca = sentinel.ca
-        _get_assigned_certificates.return_value = ([cert_mock], sentinel.private_key)
-
-        assert harness.charm.tls.get_client_tls_files() == (
-            "sentinel.private_key",
-            "sentinel.ca",
-            "sentinel.certificate",
-        )
+def test_reload_bridge_defers_when_update_config_not_ready(harness):
+    """A transient config-apply failure must defer (retry), not leave stale TLS state."""
+    with harness.hooks_disabled():
+        harness.set_leader(True)
+        harness.charm.set_secret("app", "internal-ca", "ca-content")
 
-        _get_assigned_certificates.return_value = (None, None)
-        assert harness.charm.tls.get_client_tls_files() == (None, None, None)
+    event = Mock()
+    with patch("charm.PostgresqlOperatorCharm.update_config", return_value=False):
+        harness.charm._reload_tls_after_push(event)
+    event.defer.assert_called_once()
 
 
-def test_get_peer_tls_files(harness):
+def test_reload_bridge_defers_when_tls_files_not_yet_on_disk(harness):
+    """ssl:on must not be rendered until the lib has pushed the cert files to disk."""
+    with harness.hooks_disabled():
+        harness.set_leader(True)
+        harness.charm.set_secret("app", "internal-ca", "ca-content")
+    event = Mock()
     with (
-        patch(
-            "relations.tls.TLSCertificatesRequiresV4.get_assigned_certificates"
-        ) as _get_assigned_certificates,
-        patch(
-            "charm.PostgresqlOperatorCharm.get_secret", return_value=sentinel.secret
-        ) as _get_secret,
+        patch("charm.PostgresqlOperatorCharm.update_config") as _uc,
+        # is_tls_enabled -> True via the live-fetch getter
+        patch("charm.TLSManager.get_client_tls_files", return_value=("K", "CA", "C")),
+        patch.object(harness.charm.tls_manager, "client_tls_files_on_disk", return_value=False),
     ):
-        cert_mock = Mock()
-        cert_mock.certificate = sentinel.certificate
-        cert_mock.ca = sentinel.ca
-        _get_assigned_certificates.return_value = ([cert_mock], sentinel.private_key)
-
-        assert harness.charm.tls.get_peer_tls_files() == (
-            "sentinel.private_key",
-            "sentinel.ca",
-            "sentinel.certificate",
-        )
-        assert not _get_secret.called
-
-        _get_assigned_certificates.return_value = (None, None)
-        assert harness.charm.tls.get_peer_tls_files() == (
-            sentinel.secret,
-            sentinel.secret,
-            sentinel.secret,
-        )
-        assert _get_secret.call_count == 3
-        _get_secret.assert_any_call("unit", "internal-key")
-        _get_secret.assert_any_call("unit", "internal-cert")
-        _get_secret.assert_any_call("app", "internal-ca")
-
-
-def test_on_client_certificate_available(harness):
-    with (
-        patch("charm.PostgresqlOperatorCharm.get_secret") as _get_secret,
-        patch(
-            "charm.PostgresqlOperatorCharm.push_tls_files_to_workload"
-        ) as _push_tls_files_to_workload,
-    ):
-        # Defers if internal CA is not ready yet
-        _get_secret.return_value = None
-        event_mock = Mock()
-
-        harness.charm.tls._on_certificate_available(event_mock)
+        harness.charm._reload_tls_after_push(event)
+    event.defer.assert_called_once()
+    _uc.assert_not_called()
 
-        event_mock.defer.assert_called_once_with()
-        assert not _push_tls_files_to_workload.called
-        event_mock.reset_mock()
 
-        # Defers if can't push
-        _get_secret.return_value = sentinel.secret
-        _push_tls_files_to_workload.return_value = False
+def test_reload_bridge_defers_when_update_config_raises(harness):
+    """A raising update_config must be caught and deferred, not fail the hook.
 
-        harness.charm.tls._on_certificate_available(event_mock)
+    The bridge mirrors the original charm's broad push-failure guard: a transient
+    Patroni/render failure defers and retries rather than propagating out of the
+    observer.
+    """
+    with harness.hooks_disabled():
+        harness.set_leader(True)
+        harness.charm.set_secret("app", "internal-ca", "ca-content")
 
-        event_mock.defer.assert_called_once_with()
-        _push_tls_files_to_workload.assert_called_once_with()
-        event_mock.reset_mock()
+    event = Mock()
+    with patch(
+        "charm.PostgresqlOperatorCharm.update_config",
+        side_effect=RuntimeError("patroni render failed"),
+    ):
+        # Must not raise.
+        harness.charm._reload_tls_after_push(event)
+    event.defer.assert_called_once_with()
 
 
-def test_on_peer_certificate_available(harness):
+def test_internal_cert_path_pushes_and_reloads(harness):
+    """_check_and_update_internal_cert generates, pushes, and reloads on CN mismatch."""
     with (
+        patch("charm.CharmState.unit_ip", new_callable=PropertyMock) as _unit_ip,
         patch(
-            "relations.tls.TLSCertificatesRequiresV4.get_assigned_certificates"
-        ) as _get_assigned_certificates,
-        patch("relations.tls.TLS._on_certificate_available") as _on_certificate_available,
-        patch("charm.PostgresqlOperatorCharm.get_secret") as _get_secret,
-        patch("charm.PostgresqlOperatorCharm.set_secret") as _set_secret,
+            "charm.PostgresqlOperatorCharm.get_secret",
+            return_value="-----BEGIN CERTIFICATE-----",
+        ),
+        patch("charm.load_pem_x509_certificate") as _load_cert,
+        patch("charm.TLSManager.generate_internal_peer_cert") as _generate,
+        patch("charm.TLSManager.push_tls_files") as _push,
+        patch("charm.PostgresqlOperatorCharm.update_config") as _update_config,
     ):
-        # Same ca
-        cert_mock = Mock()
-        cert_mock.ca = sentinel.ca
-        _get_assigned_certificates.return_value = ([cert_mock], None)
-        _get_secret.return_value = "sentinel.ca"
-        event_mock = Mock()
-
-        harness.charm.tls._on_peer_certificate_available(event_mock)
-
-        _on_certificate_available.assert_called_once_with(event_mock)
-        assert not _set_secret.called
-
-        # Different ca
-        _get_secret.return_value = sentinel.old_ca
+        _unit_ip.return_value = "1.2.3.4"
+        # Make the cert CN differ from the unit IP to force regeneration.
+        attr = Mock()
+        attr.value = "9.9.9.9"
+        _load_cert.return_value.subject.get_attributes_for_oid.return_value = [attr]
 
-        harness.charm.tls._on_peer_certificate_available(event_mock)
+        harness.charm._check_and_update_internal_cert()
 
-        assert _set_secret.call_count == 2
-        _set_secret.assert_any_call("unit", "current-ca", "sentinel.ca")
-        _set_secret.assert_any_call("unit", "old-ca", sentinel.old_ca)
+        _generate.assert_called_once_with()
+        _push.assert_called_once_with()
+        _update_config.assert_called_once_with()