fix(domain-removal): stop deleting certificates other live sites are served from

DELETE /api/domain ran, unconditionally, for any ssl_enabled row:

    os.remove(ssl_cert_path)
    certbot delete --cert-name <domain> --non-interactive

`domains.domain` is UNIQUE. `domains.ssl_cert_path` is not, and nothing
anywhere guarded against two rows naming the same file. Sharing is not an
edge case -- it is the normal shape of the table, because
request_ssl_bundle() deliberately creates it: one SAN certificate is issued
as `--cert-name <primary>`, published once to
/etc/haproxy/certs/<primary>.pem, and then EVERY included name's row is
pointed at that same path ("Mark every name in the bundle as ssl_enabled,
all pointing at the same combined .pem").

Measured read-only on the live SQLite in the haproxy-manager containers on
2026-08-23:

  * whp01: 157 domain rows, 150 ssl_enabled, 71 distinct cert paths.
    39 of those paths are referenced by MORE THAN ONE row, covering 118 of
    the 150 SSL-enabled rows. Worst cases: brain-jar.com.pem and
    arclightcourt.com.pem with 10 domains each, hackerpublicradio.org.pem
    with 6, anhonesthost.com.pem with 5.
  * whp02: 33 rows, 31 ssl_enabled, 15 distinct paths, 11 shared across 27
    rows -- including threeworldsoneheart.org.pem, referenced by the apex,
    its www, and mail.threeworldsoneheart.org. A production cleanup of that
    mail.* row was stopped short precisely because removing it would have
    unlinked the PEM the serving site is using.

So removing one domain unlinked a file up to nine other configured domains
were being served from. HAProxy binds the crt directory
(`bind ... ssl crt /etc/haproxy/certs`), so the loss is not noticed until
the next reload or restart, at which point the listener refuses to come up
or those names fall back to the wrong certificate.

`certbot delete` is the worse half. It destroys the lineage's archive, live
symlinks and renewal config; recovery is a fresh, rate-limited ACME order.
The old code passed `--cert-name <domain>`, which is also simply the wrong
lineage for a SAN member: 81 of whp01's 150 SSL-enabled rows have a cert
path whose basename is not their own domain, so for those the call was a
silent no-op -- while for a bundle PRIMARY it deleted the one lineage still
renewing the certificate every other name in the bundle is served with.

The fix refcounts, after the row is deleted so the query answers "who else
still needs this":

  * lineage_name_for_cert_path() -- the lineage is the published bundle's
    basename minus .pem, the same derivation _quarantine_superseded_certs()
    already uses, not the domain being removed.
  * domains_referencing_cert_path() / domains_referencing_lineage() -- the
    remaining rows that name that file, and that lineage.
  * remove_domain() unlinks only when the list is empty, `certbot delete`s
    only when the list is empty, logs the retained names explicitly when it
    skips, and reports them as certificate_retained_for /
    lineage_retained_for in the API response.

ssl_enabled is deliberately not filtered on in the refcount: the two
mistakes are not symmetric. A stale PEM left in the crt directory costs a
few kilobytes; an unlinked live one is HTTPS down for every name it serves.
Cleanup is deferred, not cancelled -- removing the last name on a bundle
still unlinks the file and deletes the lineage.

No row on either production host has ssl_enabled=1 with an empty
ssl_cert_path, so the "no path, no attributable lineage" branch changes
nothing on the current fleet.

Tests (scripts/test-cert-write-safety.py, +9, suite now 31, all offline):
last reference -> file unlinked and lineage deleted; shared file survives
removal of a SAN member AND of the bundle primary, byte-for-byte, with the
edge still starting; shared lineage is not certbot-deleted; the production
mail.* shape; removing every name eventually cleans up; an unrelated
bundle is never collateral damage. Assertions are on os.path.exists, file
contents and the recorded certbot argv, never on which branch ran.

Mutation-tested, all five mutants killed:
  1. guard absent entirely (suite run with HAPROXY_MANAGER_DIR pointed at
     main) -> 6 failures, incl. "example.com is still configured and still
     served from this file".
  2. refcount taken before the row is deleted -> 5 failures, incl. "the
     last reference is gone - now it may be removed".
  3. certbot guard removed, file guard kept -> 3 failures, incl.
     [] != ['delete --cert-name example.com --non-interactive'].
  4. lineage taken from the domain name instead of the cert path -> 3
     failures, incl. 'delete --cert-name example.com' != 'delete
     --cert-name www.example.com'.
  5. file-unlink guard removed, certbot guard kept -> 4 failures, incl.
     "two sites are still served from this bundle".

Other suites unchanged and green: test-config-rollback, test-cert-scripts,
test-stick-table-contract, test-runtime-map-contract.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

.shared-memory-project

@@ -1 +0,0 @@
-cloud-hosting-platform/haproxy-manager-base

CLAUDE.md

@@ -8,6 +8,8 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
 - **API Testing**: `./scripts/test-api.sh` - Tests all API endpoints with optional authentication
 - **Certificate Request Testing**: `./scripts/test-certificate-request.sh` - Tests certificate generation endpoints
 - **Stick-table contract**: `python3 scripts/test-stick-table-contract.py` - offline; holds the templates' `store` clauses, `STICK_TABLE_FIELD_CONTRACT`, and every consumer to each other. Run it after touching any `stick-table` line.
+- **Runtime-map contract**: `python3 scripts/test-runtime-map-contract.py` - offline; asserts the runtime map commands are `@1`-prefixed, reference the map by FILE PATH (never `#<id>`), carry the value `1`, and that every captured rejection is classified as a failure. Run it after touching any `add map`/`del map`/`clear map` path.
+- **Certificate destruction safety**: `python3 scripts/test-cert-write-safety.py` - offline; asserts a live `.pem` is never truncated, removed, or its certbot lineage deleted while any configured domain still references it (one bundle serves many names, so `ssl_cert_path` is routinely shared). Run it after touching any `os.remove`/`certbot delete`/PEM-write path.
 - **Manual Testing**: Run `curl` commands against `http://localhost:8000` endpoints as shown in README.md
 
 ### Reading stick tables (and why it is easy to get silently wrong)
@@ -31,6 +33,29 @@ reported "Scan Count"/"BLOCKED" figures parsed from `gpc0`/`gpc1` — fields no
 stick table has ever stored — for their entire existence. See the header of
 `haproxy_tarpit_config.txt` and the contract test.
 
+### Changing a runtime map (`add map` / `del map`)
+
+Same socket, two more ways to fail silently — and both were live in
+`add_ip_to_runtime_map()`/`remove_ip_from_runtime_map()` for their whole
+existence:
+
+* **Reference the map by FILE PATH, never `#<id>`.** Ids are assigned at
+  config-parse time and move on every config regeneration (on whp01
+  `blocked_ips.map` is 37, `trusted_ips.map` is 10 — there is no id 0). Use
+  `add map /etc/haproxy/blocked_ips.map <ip> 1`.
+* **A mutation answers NOTHING on success**, so an empty body is the only
+  success — any output at all is a rejection. Worse, `@1 add map #0 <ip> 1`
+  *also* answers nothing and adds nothing, so the body cannot prove an add
+  worked. **Read it back** with `@1 get map <path> <key>`.
+* Entries must carry the value `1`; haproxy.cfg matches with
+  `map_ip(...,0) -m int gt 0`, so a valueless entry does not block.
+
+In Python use `haproxy_cli(cmd, worker=True, expect_empty=True)` for mutations
+and `runtime_map_lookup()` / `runtime_map_keys()` to verify. The runtime map is
+only a fast path: `/etc/haproxy/blocked_ips.map` is authoritative and HAProxy
+re-reads it on reload, so a failed runtime command must degrade to
+"enforced on reload" and be reported, never swallowed.
+
 ### Running the Application
 - **Docker Build**: `docker build -t haproxy-manager .`
 - **Local Development**: `python haproxy_manager.py` (requires HAProxy, certbot, and dependencies installed)

IP_BLOCKING_API.md

@@ -508,20 +508,45 @@ curl -X POST http://localhost:8000/api/blocked-ips/sync \
 
 For advanced users, you can interact directly with HAProxy's runtime API:
 
+Three things about these commands are easy to get wrong, and each one fails
+**silently** (socat exits 0 either way — the rejection, if any, is only in the
+response body):
+
+* `/tmp/haproxy-cli` is HAProxy's **master** CLI socket. Map commands are
+  worker commands and need the `@1` prefix. Without it the reply is
+  `Unknown command: 'add', ...`.
+* Reference the map by its **file path**, never by `#<id>`. Ids are assigned at
+  config-parse time and move on every config regeneration (on a live edge,
+  `blocked_ips.map` is id 37, `trusted_ips.map` is 10 — there is no id 0).
+  Worse, `@1 add map #0 <ip> 1` returns an **empty** reply and adds nothing.
+* Entries must carry the value `1`. `haproxy.cfg` matches with
+  `map_ip(...,0) -m int gt 0`, so a valueless entry does not block. (`add map`
+  with no value is rejected: `'add map' expects three parameters ...`.)
+
 ```bash
+MAP=/etc/haproxy/blocked_ips.map
+
 # Add IP to runtime (immediate effect)
-echo "add map #0 192.168.1.100" | socat stdio /var/run/haproxy.sock
+echo "@1 add map $MAP 192.168.1.100 1" | socat stdio /tmp/haproxy-cli
 
 # Remove IP from runtime
-echo "del map #0 192.168.1.100" | socat stdio /var/run/haproxy.sock
+echo "@1 del map $MAP 192.168.1.100" | socat stdio /tmp/haproxy-cli
+
+# Confirm what actually happened (do not trust the exit status)
+echo "@1 get map $MAP 192.168.1.100" | socat stdio /tmp/haproxy-cli
 
 # Clear all blocked IPs from runtime
-echo "clear map #0" | socat stdio /var/run/haproxy.sock
+echo "@1 clear map $MAP" | socat stdio /tmp/haproxy-cli
 
-# Show all runtime map entries
-echo "show map #0" | socat stdio /var/run/haproxy.sock
+# Show all runtime map entries, and the map ids currently in use
+echo "@1 show map $MAP" | socat stdio /tmp/haproxy-cli
+echo "@1 show map" | socat stdio /tmp/haproxy-cli
 ```
 
+The runtime map is a **fast path only**. `/etc/haproxy/blocked_ips.map` is
+authoritative: HAProxy re-reads it on reload, so a failed runtime command
+delays a block until the next reload rather than losing it.
+
 ## Migration from ACL Method
 
 If you're upgrading from the old ACL-based method:

MIGRATION_GUIDE.md

@@ -50,14 +50,22 @@ http-request deny status 403 if { src -f /etc/haproxy/blocked_ips.map }
 - **Graceful error handling**
 
 ### 2. Runtime IP Management
+Map commands go to a **worker** (`@1`), reference the map by **file path**
+(ids move between config regenerations, and `#0` silently adds nothing), and
+carry the value `1` that `map_ip(...,0) -m int gt 0` matches on:
+
 ```bash
 # Add IP without reload (immediate effect)
-echo "add map #0 192.168.1.100" | socat stdio /var/run/haproxy.sock
+echo "@1 add map /etc/haproxy/blocked_ips.map 192.168.1.100 1" | socat stdio /tmp/haproxy-cli
 
 # Remove IP without reload
-echo "del map #0 192.168.1.100" | socat stdio /var/run/haproxy.sock
+echo "@1 del map /etc/haproxy/blocked_ips.map 192.168.1.100" | socat stdio /tmp/haproxy-cli
 ```
 
+socat exits 0 even when HAProxy rejects the command, so read the response body
+(or read the entry back with `@1 get map ...`) rather than the exit status.
+See IP_BLOCKING_API.md for the full set.
+
 ### 3. New API Endpoints
 
 #### Safe Config Reload

VERSION

@@ -1 +1 @@
-2026.09.1
+2026.08.11

haproxy_manager.py

@@ -1624,6 +1624,61 @@ def request_certificates():
     else:
         return jsonify(response), 500  # All failed
 
+def lineage_name_for_cert_path(cert_path):
+    """The certbot lineage name implied by a published bundle path.
+
+    Every writer in this module publishes to ``{SSL_CERTS_DIR}/<name>.pem`` and
+    issues that lineage with ``--cert-name <name>`` (see request_ssl() and
+    request_ssl_bundle()); _quarantine_superseded_certs() derives the lineage
+    from the filename the same way. So the basename minus ``.pem`` IS the
+    lineage, and it is NOT necessarily the domain being removed: a bundle
+    issued as ``--cert-name example.com`` also serves www.example.com and any
+    other SAN, all of whose DB rows point at ``/etc/haproxy/certs/example.com.pem``.
+    """
+    if not cert_path:
+        return None
+    base = os.path.basename(cert_path)
+    return base[:-len('.pem')] if base.endswith('.pem') else base
+
+
+def domains_referencing_cert_path(cursor, cert_path):
+    """Domains (still in the table) whose ssl_cert_path is exactly `cert_path`.
+
+    Call this AFTER the row being removed has been deleted, so the answer is
+    "who else still needs this file".
+
+    ssl_enabled is deliberately NOT filtered on. HAProxy binds the whole crt
+    directory, so a file is load-bearing for any row that names it; and the
+    asymmetry of the two mistakes is total - keeping a stale PEM costs nothing,
+    unlinking a live one is HTTPS down for every name it serves.
+    """
+    if not cert_path:
+        return []
+    cursor.execute(
+        'SELECT domain FROM domains WHERE ssl_cert_path = ? ORDER BY domain',
+        (cert_path,))
+    return [row[0] for row in cursor.fetchall()]
+
+
+def domains_referencing_lineage(cursor, lineage):
+    """Domains (still in the table) served by certbot lineage `lineage`.
+
+    Same shape as domains_referencing_cert_path(), one level further back:
+    `certbot delete --cert-name X` destroys the archive, live symlinks and
+    renewal config for X. If any remaining domain is served by a bundle
+    published from that lineage, deleting it means the next renewal silently
+    stops happening for all of them and the material cannot be recovered
+    without a fresh, rate-limited ACME order.
+    """
+    if not lineage:
+        return []
+    cursor.execute(
+        "SELECT domain, ssl_cert_path FROM domains "
+        "WHERE ssl_cert_path IS NOT NULL AND ssl_cert_path != ''")
+    return sorted({domain for domain, path in cursor.fetchall()
+                   if lineage_name_for_cert_path(path) == lineage})
+
+
 @app.route('/api/domain', methods=['DELETE'])
 @require_api_key
 def remove_domain():
@@ -1662,33 +1717,74 @@ def remove_domain():
             # Delete domain
             cursor.execute('DELETE FROM domains WHERE id = ?', (domain_id,))
 
+            # Refcount the certificate BEFORE anything is unlinked or deleted,
+            # with the row above already gone so the query answers "who else
+            # still needs this". One .pem serves many names: request_ssl_bundle()
+            # issues a single SAN cert and points EVERY included domain's row at
+            # the same /etc/haproxy/certs/<primary>.pem. Removing one of those
+            # names used to os.remove() that file and `certbot delete` its
+            # lineage unconditionally - taking HTTPS down for every other name
+            # in the bundle, and destroying the only recoverable copy with it.
+            cert_path_users = domains_referencing_cert_path(cursor, ssl_cert_path)
+            lineage = lineage_name_for_cert_path(ssl_cert_path) if ssl_cert_path else None
+            # The lineage to delete is the one this domain's bundle came from,
+            # not the domain's own name - for a SAN member those differ.
+            lineage_users = domains_referencing_lineage(cursor, lineage)
+
+        cert_retained_for = []
+        lineage_retained_for = []
+
         # Delete SSL certificate from HAProxy certs directory
         if ssl_enabled and ssl_cert_path:
-            try:
-                os.remove(ssl_cert_path)
-                logger.info(f"Removed HAProxy certificate file: {ssl_cert_path}")
-            except OSError as e:
-                logger.warning(f"Failed to remove certificate file {ssl_cert_path}: {e}")
+            if cert_path_users:
+                cert_retained_for = cert_path_users
+                logger.info(
+                    "Kept HAProxy certificate file %s while removing %s: still "
+                    "referenced by %d other domain(s): %s",
+                    ssl_cert_path, domain, len(cert_path_users),
+                    ', '.join(cert_path_users))
+            else:
+                try:
+                    os.remove(ssl_cert_path)
+                    logger.info(f"Removed HAProxy certificate file: {ssl_cert_path}")
+                except OSError as e:
+                    logger.warning(f"Failed to remove certificate file {ssl_cert_path}: {e}")
 
         # Remove certificate from certbot
         if ssl_enabled:
-            try:
-                result = subprocess.run(
-                    ['certbot', 'delete', '--cert-name', domain, '--non-interactive'],
-                    capture_output=True, text=True
-                )
-                if result.returncode == 0:
-                    logger.info(f"Removed Let's Encrypt certificate for {domain}")
-                else:
-                    logger.warning(f"Failed to remove Let's Encrypt certificate for {domain}: {result.stderr}")
-            except Exception as e:
-                logger.warning(f"Error removing Let's Encrypt certificate for {domain}: {e}")
+            if not lineage:
+                logger.info(
+                    "Skipping certbot delete for %s: no certificate path on the "
+                    "removed row, so no lineage can be attributed to it", domain)
+            elif lineage_users:
+                lineage_retained_for = lineage_users
+                logger.info(
+                    "Kept Let's Encrypt lineage %s while removing %s: still "
+                    "serving %d other domain(s): %s",
+                    lineage, domain, len(lineage_users), ', '.join(lineage_users))
+            else:
+                try:
+                    result = subprocess.run(
+                        ['certbot', 'delete', '--cert-name', lineage, '--non-interactive'],
+                        capture_output=True, text=True
+                    )
+                    if result.returncode == 0:
+                        logger.info(f"Removed Let's Encrypt certificate {lineage} for {domain}")
+                    else:
+                        logger.warning(f"Failed to remove Let's Encrypt certificate {lineage} for {domain}: {result.stderr}")
+                except Exception as e:
+                    logger.warning(f"Error removing Let's Encrypt certificate {lineage} for {domain}: {e}")
 
         # Regenerate HAProxy config
         generate_config()
 
         log_operation('remove_domain', True, f'Domain {domain} removed successfully')
-        return jsonify({'status': 'success', 'message': 'Domain configuration removed'})
+        return jsonify({
+            'status': 'success',
+            'message': 'Domain configuration removed',
+            'certificate_retained_for': cert_retained_for,
+            'lineage_retained_for': lineage_retained_for,
+        })
 
     except Exception as e:
         log_operation('remove_domain', False, str(e))
@@ -1735,8 +1831,10 @@ def add_blocked_ip():
             log_operation('add_blocked_ip', False, f'Failed to update map file for {ip_address}')
             return jsonify({'status': 'error', 'message': 'Failed to update blocked IPs map file'}), 500
         
-        # Add to runtime map for immediate effect
-        add_ip_to_runtime_map(ip_address)
+        # Add to runtime map for immediate effect. The map FILE above is what
+        # actually enforces the block once HAProxy re-reads it; this is only
+        # the fast path, so a False here is reported, not fatal.
+        runtime_ok = add_ip_to_runtime_map(ip_address)
         
         # Reload HAProxy to ensure consistency
         try:
@@ -1753,8 +1851,18 @@ def add_blocked_ip():
         except Exception as e:
             logger.warning(f"Error reloading HAProxy after blocking IP {ip_address}: {e}")
         
-        log_operation('add_blocked_ip', True, f'IP {ip_address} blocked successfully')
-        return jsonify({'status': 'success', 'blocked_ip_id': blocked_ip_id, 'message': f'IP {ip_address} has been blocked'})
+        log_operation('add_blocked_ip', True,
+                      f'IP {ip_address} blocked successfully '
+                      f'(runtime map fast path: {"ok" if runtime_ok else "FAILED, enforced on reload"})')
+        return jsonify({
+            'status': 'success',
+            'blocked_ip_id': blocked_ip_id,
+            'message': f'IP {ip_address} has been blocked',
+            # False = the block is enforced from the map file at reload rather
+            # than instantly. The caller can tell the difference; it used to be
+            # reported as instant unconditionally.
+            'runtime_map_updated': runtime_ok
+        })
     except sqlite3.IntegrityError:
         log_operation('add_blocked_ip', False, f'IP {ip_address} is already blocked')
         return jsonify({'status': 'error', 'message': 'IP address is already blocked'}), 409
@@ -1790,8 +1898,9 @@ def remove_blocked_ip():
             log_operation('remove_blocked_ip', False, f'Failed to update map file for {ip_address}')
             return jsonify({'status': 'error', 'message': 'Failed to update blocked IPs map file'}), 500
         
-        # Remove from runtime map for immediate effect
-        remove_ip_from_runtime_map(ip_address)
+        # Remove from runtime map for immediate effect. As with blocking, the
+        # map file is authoritative and the reload below picks it up.
+        runtime_ok = remove_ip_from_runtime_map(ip_address)
         
         # Reload HAProxy to ensure consistency
         try:
@@ -1808,8 +1917,14 @@ def remove_blocked_ip():
         except Exception as e:
             logger.warning(f"Error reloading HAProxy after unblocking IP {ip_address}: {e}")
         
-        log_operation('remove_blocked_ip', True, f'IP {ip_address} unblocked successfully')
-        return jsonify({'status': 'success', 'message': f'IP {ip_address} has been unblocked'})
+        log_operation('remove_blocked_ip', True,
+                      f'IP {ip_address} unblocked successfully '
+                      f'(runtime map fast path: {"ok" if runtime_ok else "FAILED, applied on reload"})')
+        return jsonify({
+            'status': 'success',
+            'message': f'IP {ip_address} has been unblocked',
+            'runtime_map_updated': runtime_ok
+        })
     except Exception as e:
         log_operation('remove_blocked_ip', False, str(e))
         return jsonify({'status': 'error', 'message': str(e)}), 500
@@ -1843,31 +1958,67 @@ def sync_blocked_ips():
             cursor.execute('SELECT ip_address FROM blocked_ips ORDER BY ip_address')
             blocked_ips = [row[0] for row in cursor.fetchall()]
         
-        # Try to clear all entries from runtime map (might fail if empty, that's ok)
+        # Clear the runtime map and re-add every blocked IP. The map is
+        # referenced by FILE PATH: `clear map #0` (what this used to send, with
+        # no `@1` either) was answered by the master socket with "Unknown
+        # command: 'clear'" and socat exited 0, so this whole block was a no-op
+        # that reported a full sync.
         try:
-            if os.path.exists(HAPROXY_SOCKET_PATH):
-                socket_path = HAPROXY_SOCKET_PATH
-            else:
-                socket_path = '/tmp/haproxy-cli'
-            
-            subprocess.run(f'echo "clear map #0" | socat stdio {socket_path}', 
-                         shell=True, capture_output=True)
-        except:
-            pass  # Clear might fail if map is empty
-        
-        # Add all IPs to runtime map
-        success_count = 0
+            haproxy_cli('clear map %s' % BLOCKED_IPS_MAP_PATH,
+                        worker=True, expect_empty=True)
+        except HaproxyCliError as e:
+            log_operation('sync_blocked_ips', False, f'Failed to clear runtime map: {e}')
+            logger.warning("Failed to clear runtime map: %s. The map file is "
+                           "still authoritative and correct.", e)
+            return jsonify({
+                'status': 'error',
+                'message': f'Failed to clear runtime map: {e}',
+                'map_file_updated': True,
+                'runtime_map_synced': False,
+                'total_ips': len(blocked_ips)
+            }), 500
+
+        # verify=False here: one `show map` read-back below costs a single
+        # round trip instead of one per IP, and answers the same question for
+        # the whole set.
+        accepted = 0
         for ip in blocked_ips:
-            if add_ip_to_runtime_map(ip):
-                success_count += 1
-        
-        log_operation('sync_blocked_ips', True, f'Synced {success_count}/{len(blocked_ips)} IPs to runtime map')
+            if add_ip_to_runtime_map(ip, verify=False):
+                accepted += 1
+
+        # Ground truth, not a count of commands that did not visibly complain.
+        try:
+            present = runtime_map_keys(BLOCKED_IPS_MAP_PATH)
+        except HaproxyCliError as e:
+            log_operation('sync_blocked_ips', False, f'Could not read back runtime map: {e}')
+            return jsonify({
+                'status': 'error',
+                'message': f'Could not read back the runtime map to verify the sync: {e}',
+                'map_file_updated': True,
+                'runtime_map_synced': False,
+                'total_ips': len(blocked_ips)
+            }), 500
+
+        missing = [ip for ip in blocked_ips if ip not in present]
+        synced = len(blocked_ips) - len(missing)
+        ok = not missing
+        if missing:
+            logger.warning(
+                "Runtime map sync incomplete: %d/%d IPs are not in %s (first "
+                "few: %s). They remain blocked via the map file on reload.",
+                len(missing), len(blocked_ips), BLOCKED_IPS_MAP_PATH, missing[:5])
+
+        log_operation('sync_blocked_ips', ok,
+                      f'Verified {synced}/{len(blocked_ips)} IPs present in the runtime map')
         return jsonify({
-            'status': 'success', 
-            'message': f'Synced {success_count}/{len(blocked_ips)} IPs to runtime map',
+            'status': 'success' if ok else 'partial',
+            'message': f'Verified {synced}/{len(blocked_ips)} IPs present in the runtime map',
             'total_ips': len(blocked_ips),
-            'synced_ips': success_count
-        })
+            'synced_ips': synced,
+            'accepted_commands': accepted,
+            'missing_ips': missing[:50],
+            'runtime_map_synced': ok
+        }), (200 if ok else 207)
     except Exception as e:
         log_operation('sync_blocked_ips', False, str(e))
         return jsonify({'status': 'error', 'message': str(e)}), 500
@@ -1930,6 +2081,10 @@ STICK_TABLE_ENTRY_META = ('key', 'use', 'exp', 'shard')
 _HAPROXY_CLI_ERROR_MARKERS = (
     'Unknown command',
     'No such table',
+    # `add map #0 ...` / `get map /etc/haproxy/nope.map ...` -- captured
+    # verbatim from HAProxy 3.0.11 on the live edge.
+    'Unknown map identifier',
+    'Key not found',
     'Permission denied',
     "Can't find the specified process",
     'unknown process',
@@ -1947,8 +2102,15 @@ class HaproxyCliError(RuntimeError):
 
     Exists so a rejected command cannot be mistaken for an empty result. That
     distinction is the whole point of this module's stick-table code.
+
+    `.responses` holds the raw, stripped response body of every attempt, so a
+    caller can tell one rejection apart from another (`del map` answering
+    "Key not found." is a no-op, not a failure) without regex-matching the
+    formatted message.
     """
 
+    responses = ()
+
 
 def _haproxy_socket_path():
     return HAPROXY_SOCKET_PATH if os.path.exists(HAPROXY_SOCKET_PATH) else '/tmp/haproxy-cli'
@@ -1972,7 +2134,7 @@ def _cli_send(command, socket_path, timeout):
     return proc.stdout
 
 
-def haproxy_cli(command, worker=False, timeout=None):
+def haproxy_cli(command, worker=False, timeout=None, expect_empty=False):
     """Send one runtime-API command and return its response, or raise.
 
     `worker=True` marks a command that only the WORKER answers (show table,
@@ -1981,24 +2143,41 @@ def haproxy_cli(command, worker=False, timeout=None):
     have one. Rather than guessing from configuration that can change under us,
     try the prefixed form and fall back -- and raise if BOTH are rejected,
     instead of returning HAProxy's help text as if it were data.
+
+    `expect_empty=True` is for MUTATING commands (add/del/clear map, set map).
+    HAProxy answers those with nothing at all on success, so the rule inverts:
+    an empty body is the success, and ANY non-empty body is a rejection. That
+    is deliberately stricter than matching _HAPROXY_CLI_ERROR_MARKERS -- the
+    marker list can only ever recognise the rejections someone has already
+    seen, and a mutation that prints anything has not done what was asked. Two
+    real examples this catches that the marker list did not:
+    `'add map' expects three parameters ...` and `Unknown map identifier.`
     """
     socket_path = _haproxy_socket_path()
     timeout = timeout if timeout is not None else DEFAULT_SUBPROCESS_TIMEOUT
     attempts = (['@1 ' + command, command] if worker else [command])
     failures = []
+    bodies = []
     for attempt in attempts:
         try:
             out = _cli_send(attempt, socket_path, timeout)
         except subprocess.TimeoutExpired:
             raise HaproxyCliError('timed out after %ss running %r on %s'
                                   % (timeout, attempt, socket_path))
-        if out.strip() and not _cli_response_is_error(out):
+        body = out.strip()
+        if expect_empty:
+            if not body:
+                return out
+        elif body and not _cli_response_is_error(out):
             return out
-        failures.append('%r -> %r' % (attempt, out.strip()[:200] or '<empty response>'))
-    raise HaproxyCliError(
+        bodies.append(body)
+        failures.append('%r -> %r' % (attempt, body[:200] or '<empty response>'))
+    error = HaproxyCliError(
         'HAProxy rejected %r on %s (socat exited 0 -- the rejection is in the '
         'response body, which is exactly why this is checked): %s'
         % (command, socket_path, '; '.join(failures)))
+    error.responses = tuple(bodies)
+    raise error
 
 
 def parse_stick_table_entry(line):
@@ -2185,13 +2364,16 @@ def temporary_block():
         if not update_blocked_ips_map():
             return jsonify({'status': 'error', 'message': 'Failed to update map file'}), 500
 
-        add_ip_to_runtime_map(ip_address)
+        runtime_ok = add_ip_to_runtime_map(ip_address)
 
-        log_operation('temporary_block', True, f'Temporarily blocked {ip_address} for {duration_minutes} minutes')
+        log_operation('temporary_block', True,
+                      f'Temporarily blocked {ip_address} for {duration_minutes} minutes '
+                      f'(runtime map fast path: {"ok" if runtime_ok else "FAILED, enforced on reload"})')
         return jsonify({
             'status': 'success',
             'message': f'IP {ip_address} temporarily blocked for {duration_minutes} minutes',
-            'expires_at': expiry_time.isoformat()
+            'expires_at': expiry_time.isoformat(),
+            'runtime_map_updated': runtime_ok
         })
     except Exception as e:
         log_operation('temporary_block', False, str(e))
@@ -2204,6 +2386,10 @@ def clear_expired_blocks():
     try:
         current_time = datetime.now()
         expired_ips = []
+        # IPs the runtime fast path could not drop. They still come out of the
+        # map file below, so they unblock on the next reload -- but silently
+        # reporting them as cleared is what this whole change is about.
+        runtime_failures = []
 
         with sqlite3.connect(DB_FILE) as conn:
             cursor = conn.cursor()
@@ -2224,17 +2410,21 @@ def clear_expired_blocks():
                 # Remove expired IPs
                 for ip in expired_ips:
                     cursor.execute('DELETE FROM blocked_ips WHERE ip_address = ?', (ip,))
-                    remove_ip_from_runtime_map(ip)
+                    if not remove_ip_from_runtime_map(ip):
+                        runtime_failures.append(ip)
 
         # Update map file if any IPs were removed
         if expired_ips:
             update_blocked_ips_map()
 
-        log_operation('clear_expired_blocks', True, f'Cleared {len(expired_ips)} expired IP blocks')
+        log_operation('clear_expired_blocks', not runtime_failures,
+                      f'Cleared {len(expired_ips)} expired IP blocks '
+                      f'({len(runtime_failures)} not removed from the runtime map)')
         return jsonify({
-            'status': 'success',
+            'status': 'success' if not runtime_failures else 'partial',
             'message': f'Cleared {len(expired_ips)} expired IP blocks',
-            'cleared_ips': expired_ips
+            'cleared_ips': expired_ips,
+            'runtime_map_failures': runtime_failures
         })
     except Exception as e:
         log_operation('clear_expired_blocks', False, str(e))
@@ -3273,50 +3463,167 @@ def update_blocked_ips_map(promote_backup=True):
         logger.error(f"Failed to update blocked IPs map: {e}")
         return False
 
-def add_ip_to_runtime_map(ip_address):
-    """Add IP to HAProxy runtime map without reload"""
+# ---------------------------------------------------------------------------
+# Runtime map fast path (blocked IPs)
+#
+# WHAT WAS WRONG, AND WHY NOBODY NOTICED FOR ITS ENTIRE EXISTENCE
+# ---------------------------------------------------------------
+# add_ip_to_runtime_map()/remove_ip_from_runtime_map() sent
+# `add map #0 <ip> 1` / `del map #0 <ip>` to /tmp/haproxy-cli and returned True
+# whenever socat exited 0. Two independent defects, three silences:
+#
+#   1. NO `@1` PREFIX. /tmp/haproxy-cli is HAProxy's MASTER CLI socket. Map
+#      commands are worker commands. The master answers
+#      `Unknown command: 'add', but maybe one of the following ones is a better
+#      match: ...` -- and socat still exits 0, so `result.returncode == 0` was
+#      true and the function logged "Added IP x to runtime map".
+#   2. `#0` IS NOT A VALID MAP ID. Ids are assigned at config-parse time and
+#      move whenever the config is regenerated; on whp01 blocked_ips.map is
+#      id 37 and trusted_ips.map is 10. There is no id 0. Hardcoding ANY number
+#      is wrong -- reference the map by its FILE PATH, which is stable because
+#      it is what haproxy.cfg names in `map_ip(/etc/haproxy/blocked_ips.map,0)`.
+#   3. `add map #0 ...` fails SILENTLY EVEN WITH `@1`. Captured on HAProxy
+#      3.0.11: `@1 add map #0 192.0.2.88 1` returns an EMPTY body, exit 0, and
+#      adds nothing to any map -- while `@1 del map #0 <ip>` and
+#      `@1 show map #0` both answer `Unknown map identifier.`. So a
+#      response-body check alone cannot catch defect 2 on the add path. That is
+#      why every mutation here is READ BACK with `get map` instead of trusting
+#      either the exit status or the (empty) reply.
+#
+# The blocking itself never depended on this: update_blocked_ips_map() rewrites
+# /etc/haproxy/blocked_ips.map and the callers reload HAProxy, which re-reads
+# the file. The FILE IS AUTHORITATIVE; this is only the no-reload fast path.
+# Every function below therefore returns a bool the caller can report, and
+# never raises into a request handler -- a runtime-map failure must degrade to
+# "enforced on reload", not to "not blocked" and not to a 500.
+#
+# scripts/test-runtime-map-contract.py holds the command strings and the
+# classification of every captured response to these rules.
+# ---------------------------------------------------------------------------
+
+# The value every blocked_ips.map entry must carry. haproxy.cfg matches with
+# `map_ip(/etc/haproxy/blocked_ips.map,0) -m int gt 0`, so a keyed entry with
+# no value evaluates to 0 and is NOT blocked. Runtime map and file must agree.
+BLOCKED_IPS_MAP_VALUE = '1'
+
+_GET_MAP_FOUND_RE = re.compile(r'\bfound=(yes|no)\b')
+_GET_MAP_VALUE_RE = re.compile(r'\bvalue="([^"]*)"')
+
+
+def runtime_map_lookup(map_path, key):
+    """(found, value) for one key in a runtime map, or raise HaproxyCliError.
+
+    Reads back what `add map`/`del map` actually did. `get map` answers
+    `type=ip, case=sensitive, found=yes, idx=tree, key="1.2.3.4", value="1",
+    type="str"` or `type=ip, case=sensitive, found=no`; an unusable map
+    reference answers `Unknown map identifier.`, which haproxy_cli() rejects.
+    """
+    out = haproxy_cli('get map %s %s' % (map_path, key), worker=True)
+    match = _GET_MAP_FOUND_RE.search(out)
+    if not match:
+        raise HaproxyCliError(
+            'unparseable `get map %s %s` response (no found=yes/no): %r'
+            % (map_path, key, out.strip()[:200]))
+    if match.group(1) == 'no':
+        return (False, None)
+    value = _GET_MAP_VALUE_RE.search(out)
+    return (True, value.group(1) if value else None)
+
+
+def runtime_map_keys(map_path):
+    """The set of keys currently in a runtime map, or raise HaproxyCliError.
+
+    `show map <file>` emits `<0x-pointer> <key> <value>` per line. An empty map
+    legitimately emits nothing, which is why this does not go through the
+    non-empty check.
+    """
     try:
-        if os.path.exists(HAPROXY_SOCKET_PATH):
-            socket_path = HAPROXY_SOCKET_PATH
-        else:
-            socket_path = '/tmp/haproxy-cli'
+        out = haproxy_cli('show map %s' % map_path, worker=True)
+    except HaproxyCliError as e:
+        # An empty map is a real, distinguishable state -- not a rejection.
+        if e.responses and all(body == '' for body in e.responses):
+            return set()
+        raise
+    keys = set()
+    for line in out.splitlines():
+        parts = line.split()
+        if len(parts) >= 2 and parts[0].startswith('0x'):
+            keys.add(parts[1])
+    return keys
 
-        # Add to runtime map (map file ID 0 for blocked IPs)
-        # Format: add map #<id> <key> <value>
-        # For IP blocking, value is always "1"
-        cmd = f'echo "add map #0 {ip_address} 1" | socat stdio {socket_path}'
-        result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
 
-        if result.returncode == 0:
-            logger.info(f"Added IP {ip_address} to runtime map")
-            return True
-        else:
-            logger.warning(f"Failed to add IP to runtime map: {result.stderr}")
-            return False
+def add_ip_to_runtime_map(ip_address, verify=True):
+    """Add IP to the running HAProxy's blocked map without a reload.
+
+    Returns True only when the entry is verifiably present with the value the
+    config matches on. False means the FILE + reload path is what will enforce
+    this block -- which it does regardless; see the section comment above.
+    """
+    try:
+        haproxy_cli(
+            'add map %s %s %s' % (BLOCKED_IPS_MAP_PATH, ip_address, BLOCKED_IPS_MAP_VALUE),
+            worker=True, expect_empty=True)
+        if verify:
+            found, value = runtime_map_lookup(BLOCKED_IPS_MAP_PATH, ip_address)
+            if not found:
+                raise HaproxyCliError(
+                    '`add map` was accepted but %s is not in %s afterwards -- '
+                    'the command did nothing (this is exactly how `#0` failed)'
+                    % (ip_address, BLOCKED_IPS_MAP_PATH))
+            if value != BLOCKED_IPS_MAP_VALUE:
+                raise HaproxyCliError(
+                    '%s is in %s with value %r, not %r -- haproxy.cfg matches '
+                    'with `-m int gt 0`, so this entry does NOT block'
+                    % (ip_address, BLOCKED_IPS_MAP_PATH, value, BLOCKED_IPS_MAP_VALUE))
+        logger.info(f"Added IP {ip_address} to runtime map (verified={verify})")
+        return True
+    except HaproxyCliError as e:
+        logger.warning(
+            "Runtime map fast path FAILED for %s: %s. The block is NOT lost -- "
+            "%s was rewritten and HAProxy re-reads it on reload -- but it does "
+            "not take effect until that reload completes.",
+            ip_address, e, BLOCKED_IPS_MAP_PATH)
+        return False
     except Exception as e:
-        logger.error(f"Error adding IP to runtime map: {e}")
+        logger.error(f"Error adding IP {ip_address} to runtime map: {e}")
         return False
 
-def remove_ip_from_runtime_map(ip_address):
-    """Remove IP from HAProxy runtime map without reload"""
+
+def remove_ip_from_runtime_map(ip_address, verify=True):
+    """Remove IP from the running HAProxy's blocked map without a reload.
+
+    Returns True only when the key is verifiably gone. `Key not found.` means
+    the runtime map never had it, which is the requested end state, so that is
+    a success -- but it is logged, because it also means the runtime map and
+    the file had drifted apart.
+    """
     try:
-        if os.path.exists(HAPROXY_SOCKET_PATH):
-            socket_path = HAPROXY_SOCKET_PATH
-        else:
-            socket_path = '/tmp/haproxy-cli'
-        
-        # Remove from runtime map (map file ID 0 for blocked IPs)
-        cmd = f'echo "del map #0 {ip_address}" | socat stdio {socket_path}'
-        result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
-        
-        if result.returncode == 0:
-            logger.info(f"Removed IP {ip_address} from runtime map")
-            return True
-        else:
-            logger.warning(f"Failed to remove IP from runtime map: {result.stderr}")
-            return False
+        try:
+            haproxy_cli('del map %s %s' % (BLOCKED_IPS_MAP_PATH, ip_address),
+                        worker=True, expect_empty=True)
+        except HaproxyCliError as e:
+            if not any(body.startswith('Key not found') for body in e.responses):
+                raise
+            logger.info(
+                "Runtime map had no entry for %s to remove (`Key not found.`); "
+                "the file and the runtime map had drifted", ip_address)
+        if verify:
+            found, _ = runtime_map_lookup(BLOCKED_IPS_MAP_PATH, ip_address)
+            if found:
+                raise HaproxyCliError(
+                    '`del map` was accepted but %s is STILL in %s'
+                    % (ip_address, BLOCKED_IPS_MAP_PATH))
+        logger.info(f"Removed IP {ip_address} from runtime map (verified={verify})")
+        return True
+    except HaproxyCliError as e:
+        logger.warning(
+            "Runtime map fast path FAILED for %s: %s. The unblock is NOT lost -- "
+            "%s was rewritten and HAProxy re-reads it on reload -- but the IP "
+            "stays blocked until that reload completes.",
+            ip_address, e, BLOCKED_IPS_MAP_PATH)
+        return False
     except Exception as e:
-        logger.error(f"Error removing IP from runtime map: {e}")
+        logger.error(f"Error removing IP {ip_address} from runtime map: {e}")
         return False
 
 def start_haproxy():

scripts/test-cert-write-safety.py

@@ -27,6 +27,8 @@ These tests pin the invariants:
   * nothing is published that is not a complete, validated cert+key pair;
   * no old certificate file is removed and no lineage deleted until the
     replacement is validated, in place, and actually loaded by HAProxy;
+  * removing ONE domain never unlinks a .pem, or deletes a certbot lineage,
+    that other still-configured domains are being served from;
   * only final .pem files ever exist in the crt directory.
 
 Running
@@ -894,6 +896,184 @@ class TestBundleValidation(CertPublishTestCase):
                          'a private key must never be group/world writable')
 
 
+class TestSharedCertificateSurvivesDomainRemoval(CertPublishTestCase):
+    """Bug 5: DELETE /api/domain unlinked a PEM other live sites were served from.
+
+    `domains.domain` is UNIQUE; `domains.ssl_cert_path` is not, and nothing
+    ever made it so. request_ssl_bundle() issues one SAN certificate and points
+    every included name's row at the same /etc/haproxy/certs/<primary>.pem, so
+    sharing is not an edge case - it is the normal shape of the table. Measured
+    on production the day this was written: 39 of 71 distinct cert paths on one
+    host were referenced by more than one domain row (118 of 150 SSL-enabled
+    rows), and one shared path was
+    /etc/haproxy/certs/threeworldsoneheart.org.pem, referenced by the live
+    apex, its www, and a mail.* alias.
+
+    remove_domain() did, unconditionally:
+
+        os.remove(ssl_cert_path)
+        certbot delete --cert-name <domain>
+
+    Removing the mail.* alias would therefore have deleted the PEM the apex was
+    serving on, and (had the alias been the bundle primary) the lineage behind
+    it - HTTPS down for every other name in the bundle, with no local copy and
+    only a fresh, rate-limited ACME order to recover from.
+
+    These tests assert file-system and certbot-invocation outcomes, not which
+    branch was taken.
+    """
+
+    BUNDLE = '/etc/haproxy/certs'  # documentation only; SSL_CERTS_DIR is stubbed
+
+    def _cert_for(self, primary):
+        """Publish a bundle for `primary` and return its path."""
+        return self.publish_live_bundle(primary)
+
+    def remove(self, domain):
+        return self.client.delete('/api/domain', json={'domain': domain})
+
+    # -- last reference: the cleanup must still happen --------------------
+
+    def test_last_reference_removal_unlinks_the_pem(self):
+        cert = self._cert_for('example.com')
+        self.add_domain('example.com', 'be_example', ssl_cert_path=cert)
+
+        resp = self.remove('example.com')
+
+        self.assertEqual(200, resp.status_code, resp.data)
+        self.assertFalse(
+            os.path.exists(cert),
+            'nothing else referenced this bundle - it must be cleaned up, or '
+            'the crt directory accumulates certs for domains that are gone')
+
+    def test_last_reference_removal_deletes_the_lineage(self):
+        cert = self._cert_for('example.com')
+        self.add_domain('example.com', 'be_example', ssl_cert_path=cert)
+
+        self.remove('example.com')
+
+        self.assertEqual(
+            ['delete --cert-name example.com --non-interactive'],
+            self.certbot_deletes(),
+            'the last name on a lineage went away - the lineage should go too')
+
+    def test_lineage_deleted_is_the_bundles_not_the_domains_own_name(self):
+        """Removing a SAN member must target the lineage that issued the file."""
+        cert = self._cert_for('example.com')
+        self.add_domain('www.example.com', 'be_www', ssl_cert_path=cert)
+
+        self.remove('www.example.com')
+
+        self.assertEqual(
+            ['delete --cert-name example.com --non-interactive'],
+            self.certbot_deletes(),
+            'the lineage is named after the bundle primary (--cert-name), not '
+            'after whichever SAN happened to be removed last')
+
+    # -- shared reference: nothing may be destroyed -----------------------
+
+    def test_shared_pem_survives_removal_of_one_name(self):
+        cert = self._cert_for('example.com')
+        before = self.read(cert)
+        self.add_domain('example.com', 'be_apex', ssl_cert_path=cert)
+        self.add_domain('www.example.com', 'be_www', ssl_cert_path=cert)
+
+        resp = self.remove('www.example.com')
+
+        self.assertEqual(200, resp.status_code, resp.data)
+        self.assertTrue(
+            os.path.exists(cert),
+            'example.com is still configured and still served from this file')
+        self.assertEqual(before, self.read(cert),
+                         'the surviving bundle must be byte-for-byte intact')
+        self.assertTrue(self.edge_would_start(),
+                        'the edge must still load the crt directory')
+
+    def test_shared_pem_survives_removal_of_the_bundle_primary(self):
+        """The worst shape: the name being removed IS the lineage/file name."""
+        cert = self._cert_for('example.com')
+        before = self.read(cert)
+        self.add_domain('example.com', 'be_apex', ssl_cert_path=cert)
+        self.add_domain('www.example.com', 'be_www', ssl_cert_path=cert)
+
+        self.remove('example.com')
+
+        self.assertTrue(
+            os.path.exists(cert),
+            'www.example.com is still configured and is served from this exact '
+            'file - removing the apex must not unlink it')
+        self.assertEqual(before, self.read(cert))
+        self.assertTrue(self.edge_would_start())
+
+    def test_shared_lineage_is_not_certbot_deleted(self):
+        cert = self._cert_for('example.com')
+        self.add_domain('example.com', 'be_apex', ssl_cert_path=cert)
+        self.add_domain('www.example.com', 'be_www', ssl_cert_path=cert)
+
+        self.remove('example.com')
+
+        self.assertEqual(
+            [], self.certbot_deletes(),
+            'certbot delete destroys archive, live and renewal config for a '
+            'lineage that is still renewing the certificate www.example.com '
+            'is served with')
+
+    def test_production_shape_mail_alias_removal(self):
+        """The exact row that stopped a production cleanup.
+
+        mail.threeworldsoneheart.org carried
+        ssl_cert_path=/etc/haproxy/certs/threeworldsoneheart.org.pem - the live
+        PEM of a different, serving site.
+        """
+        cert = self._cert_for('threeworldsoneheart.org')
+        before = self.read(cert)
+        self.add_domain('threeworldsoneheart.org', 'be_apex', ssl_cert_path=cert)
+        self.add_domain('www.threeworldsoneheart.org', 'be_www', ssl_cert_path=cert)
+        self.add_domain('mail.threeworldsoneheart.org', 'be_mail', ssl_cert_path=cert)
+
+        self.remove('mail.threeworldsoneheart.org')
+
+        self.assertTrue(os.path.exists(cert),
+                        'two sites are still served from this bundle')
+        self.assertEqual(before, self.read(cert))
+        self.assertEqual([], self.certbot_deletes())
+        self.assertTrue(self.edge_would_start())
+
+    def test_removing_every_name_eventually_cleans_up(self):
+        """The guard defers cleanup, it does not cancel it."""
+        cert = self._cert_for('example.com')
+        self.add_domain('example.com', 'be_apex', ssl_cert_path=cert)
+        self.add_domain('www.example.com', 'be_www', ssl_cert_path=cert)
+
+        self.remove('www.example.com')
+        self.assertTrue(os.path.exists(cert), 'apex still needs it')
+
+        self.remove('example.com')
+        self.assertFalse(os.path.exists(cert),
+                         'the last reference is gone - now it may be removed')
+        self.assertEqual(
+            ['delete --cert-name example.com --non-interactive'],
+            self.certbot_deletes())
+
+    def test_unrelated_domains_certificate_is_untouched(self):
+        """Sharing is by exact path; different bundles stay independent."""
+        mine = self._cert_for('example.com')
+        theirs = self._cert_for('other.example')
+        theirs_before = self.read(theirs)
+        self.add_domain('example.com', 'be_mine', ssl_cert_path=mine)
+        self.add_domain('other.example', 'be_theirs', ssl_cert_path=theirs)
+
+        self.remove('example.com')
+
+        self.assertFalse(os.path.exists(mine))
+        self.assertTrue(os.path.exists(theirs),
+                        'a different bundle must not be collateral damage')
+        self.assertEqual(theirs_before, self.read(theirs))
+        self.assertEqual(
+            ['delete --cert-name example.com --non-interactive'],
+            self.certbot_deletes())
+
+
 @unittest.skipIf(TESTING_FOREIGN_TREE,
                  'HAPROXY_MANAGER_DIR points at another tree')
 class TestPublisherApiIsPresent(unittest.TestCase):

scripts/test-runtime-map-contract.py

@@ -0,0 +1,397 @@
+#!/usr/bin/env python3
+"""Contract test: the runtime-map fast path (blocked IPs).
+
+Why this file exists
+--------------------
+`add_ip_to_runtime_map()` and `remove_ip_from_runtime_map()` spent their whole
+existence sending
+
+    add map #0 <ip> 1
+    del map #0 <ip>
+
+to `/tmp/haproxy-cli` and returning True whenever socat exited 0. Neither
+command has ever worked. Two independent defects:
+
+  * **No `@1` prefix.** `/tmp/haproxy-cli` is HAProxy's MASTER CLI socket; map
+    commands are worker commands. The master answers `Unknown command: 'add',
+    but maybe one of the following ones is a better match: ...` -- and **socat
+    still exits 0**, so `result.returncode == 0` was true and the function
+    logged "Added IP x to runtime map".
+  * **`#0` is not a valid map id.** Ids are assigned at config-parse time and
+    move on every config regeneration; on the live edge `blocked_ips.map` is
+    id 37 and `trusted_ips.map` is 10. There is no id 0. Any hardcoded number
+    is wrong -- the map must be referenced by its FILE PATH, which is what
+    haproxy.cfg itself names in `map_ip(/etc/haproxy/blocked_ips.map,0)`.
+
+And a third silence that makes a response-body check alone insufficient:
+`@1 add map #0 <ip> 1` returns an **empty body**, exit 0, and adds nothing
+anywhere -- while `@1 del map #0 <ip>` answers `Unknown map identifier.`. The
+add path can therefore only be trusted after reading the entry back.
+
+IP blocking still worked, because `update_blocked_ips_map()` rewrites
+`/etc/haproxy/blocked_ips.map` and the callers reload HAProxy, which re-reads
+it. The FILE is authoritative; this is the no-reload fast path, and it has
+never once run while reporting that it did.
+
+What it enforces
+----------------
+  1. The command strings actually sent: `@1` prefix first, map referenced by
+     PATH and never by `#<id>`, and the value `1` that
+     `map_ip(...,0) -m int gt 0` requires.
+  2. Every captured rejection is classified as FAILURE (returns False), not
+     success -- including the two that carry no error text at all.
+  3. Success is only reported when the entry reads back in the state asked
+     for. Not the exit status, not an empty reply.
+  4. No source in this repo builds a map command with a `#<id>` reference.
+     Comments may describe the old form; code may not use it.
+
+Runs fully offline: `_cli_send()` is replaced, so no socket, no socat, no
+HAProxy, no network.
+
+Running
+-------
+    python3 scripts/test-runtime-map-contract.py
+"""
+
+import ast
+import io
+import os
+import re
+import sys
+import glob
+import logging
+import tempfile
+import unittest
+
+MODULE_DIR = os.path.abspath(
+    os.environ.get('HAPROXY_MANAGER_DIR',
+                   os.path.join(os.path.dirname(os.path.abspath(__file__)), '..'))
+)
+os.chdir(MODULE_DIR)
+sys.path.insert(0, MODULE_DIR)
+
+_LOG_DIR = tempfile.mkdtemp(prefix='haproxy-mgr-test-logs-')
+_real_file_handler = logging.FileHandler
+logging.FileHandler = (
+    lambda filename, *a, **kw: _real_file_handler(
+        os.path.join(_LOG_DIR, os.path.basename(filename)), *a, **kw)
+)
+
+import haproxy_manager  # noqa: E402
+
+logging.getLogger().setLevel(logging.CRITICAL)
+
+MAP = haproxy_manager.BLOCKED_IPS_MAP_PATH
+IP = '192.0.2.77'
+
+# ---------------------------------------------------------------------------
+# Responses captured VERBATIM from the haproxy-manager container on the live
+# edge (HAProxy 3.0.11, 2026-08-22). socat exited 0 for every single one of
+# them, which is the entire reason none of this is decided on exit status.
+# ---------------------------------------------------------------------------
+
+# What the MASTER socket answers to an unprefixed `add map ...` -- i.e. the
+# reply the old code read as success.
+MASTER_REJECTS_ADD = """\
+Unknown command: 'add', but maybe one of the following ones is a better match:
+  @!<pid>                                 : send a command to the <pid> process
+  @master                                 : send a command to the master process
+  hard-reload                             : achieve a hard-reload (-st) of haproxy
+  reload                                  : achieve a soft-reload (-sf) of haproxy
+  user                                    : lower the level of the current CLI session to user
+  help [<command>]                        : list matching or all commands
+  prompt [timed]                          : toggle interactive mode with prompt
+  quit                                    : disconnect
+"""
+
+MASTER_REJECTS_DEL = MASTER_REJECTS_ADD.replace("'add'", "'del'")
+
+# `@1 del map #0 <ip>` / `@1 get map /etc/haproxy/nope.map <ip>`.
+UNKNOWN_MAP_IDENTIFIER = 'Unknown map identifier. Please use #<id> or <file>.\n'
+
+# `@1 add map <map> <ip>` with the value omitted (as the old docs showed).
+ADD_MAP_MISSING_VALUE = (
+    "'add map' expects three parameters (map identifier, key and value) or one "
+    "parameter (map identifier) and a payload\n")
+
+# `@1 del map <map> <ip>` for a key the runtime map does not hold.
+KEY_NOT_FOUND = 'Key not found.\n'
+
+# A successful mutation. THIS IS THE WHOLE PROBLEM: it is byte-for-byte what
+# `@1 add map #0 <ip> 1` also returns while adding nothing at all.
+MUTATION_OK = ''
+
+GET_FOUND = ('type=ip, case=sensitive, found=yes, idx=tree, key="%s", '
+             'value="1", type="str"\n' % IP)
+GET_NOT_FOUND = 'type=ip, case=sensitive, found=no\n'
+# `@1 get map #0 <ip>` -- "found", but with no value. haproxy.cfg matches with
+# `-m int gt 0`, so an entry like this does NOT block.
+GET_FOUND_NO_VALUE = ('type=ip, case=sensitive, found=yes, idx=tree, key="%s", '
+                      'value=none\n' % IP)
+
+SHOW_MAP_SAMPLE = (
+    '0x7f6e5a788700 101.36.109.130 1\n'
+    '0x7f6e5a788780 101.47.140.218 1\n'
+    '0x7f6e5a1fd500 %s 1\n' % IP)
+
+
+class FakeSocket(object):
+    """Replaces _cli_send(). Records every command, answers from a script.
+
+    `script` is a list of (substring, response) pairs, consulted in order; the
+    first whose substring appears in the command wins. Anything unmatched is
+    an explicit test bug, not a silent default.
+    """
+
+    def __init__(self, script):
+        self.script = script
+        self.sent = []
+
+    def __call__(self, command, socket_path, timeout):
+        self.sent.append(command)
+        for needle, response in self.script:
+            if needle in command:
+                return response
+        raise AssertionError('test script has no response for %r' % command)
+
+
+def with_socket(script):
+    """Install a FakeSocket for the duration of a `with` block."""
+    class _Ctx(object):
+        def __enter__(self):
+            self.fake = FakeSocket(script)
+            self._real = haproxy_manager._cli_send
+            haproxy_manager._cli_send = self.fake
+            return self.fake
+
+        def __exit__(self, *exc):
+            haproxy_manager._cli_send = self._real
+            return False
+    return _Ctx()
+
+
+# Every command the happy path needs, in the order the code issues them.
+HAPPY_ADD = [('get map', GET_FOUND), ('add map', MUTATION_OK)]
+HAPPY_DEL = [('get map', GET_NOT_FOUND), ('del map', MUTATION_OK)]
+
+
+class CommandsAreWellFormed(unittest.TestCase):
+    """Guard 1: the exact bytes on the wire.
+
+    Both original defects are visible here and nowhere else -- a `#0` map
+    reference and a missing `@1` are perfectly ordinary-looking Python.
+    """
+
+    def test_add_sends_worker_prefixed_path_referenced_command(self):
+        with with_socket(HAPPY_ADD) as fake:
+            self.assertTrue(haproxy_manager.add_ip_to_runtime_map(IP))
+        self.assertEqual(fake.sent[0], '@1 add map %s %s 1' % (MAP, IP))
+
+    def test_del_sends_worker_prefixed_path_referenced_command(self):
+        with with_socket(HAPPY_DEL) as fake:
+            self.assertTrue(haproxy_manager.remove_ip_from_runtime_map(IP))
+        self.assertEqual(fake.sent[0], '@1 del map %s %s' % (MAP, IP))
+
+    def test_every_command_is_tried_on_the_worker_first(self):
+        """`/tmp/haproxy-cli` is the MASTER socket; bare map commands 404."""
+        for run in (lambda: haproxy_manager.add_ip_to_runtime_map(IP),
+                    lambda: haproxy_manager.remove_ip_from_runtime_map(IP)):
+            with with_socket(HAPPY_ADD + HAPPY_DEL) as fake:
+                run()
+            for command in fake.sent:
+                self.assertTrue(command.startswith('@1 '),
+                                '%r is missing the @1 worker prefix' % command)
+
+    def test_no_command_references_a_map_by_id(self):
+        """Map ids move on every config regeneration. Path, always."""
+        script = HAPPY_ADD + HAPPY_DEL + [('show map', SHOW_MAP_SAMPLE)]
+        with with_socket(script) as fake:
+            haproxy_manager.add_ip_to_runtime_map(IP)
+            haproxy_manager.remove_ip_from_runtime_map(IP)
+            haproxy_manager.runtime_map_keys(MAP)
+        for command in fake.sent:
+            self.assertNotRegex(
+                command, r'\bmap\s+#',
+                '%r references a map by id; ids are not stable' % command)
+            self.assertIn(MAP, command,
+                          '%r does not name the map file' % command)
+
+    def test_add_carries_the_value_the_config_matches_on(self):
+        """`map_ip(...,0) -m int gt 0`: a valueless entry does not block."""
+        self.assertEqual(haproxy_manager.BLOCKED_IPS_MAP_VALUE, '1')
+        with with_socket(HAPPY_ADD) as fake:
+            haproxy_manager.add_ip_to_runtime_map(IP)
+        self.assertTrue(fake.sent[0].endswith(' %s 1' % IP),
+                        '%r has no value; HAProxy rejects it' % fake.sent[0])
+
+
+class RejectionIsFailure(unittest.TestCase):
+    """Guard 2+3: nothing may report success unless the map really changed.
+
+    Every response below was returned by the live socket with **exit code 0**.
+    The old code returned True for all of them.
+    """
+
+    def test_master_socket_rejection_of_add_is_failure(self):
+        with with_socket([('get map', GET_NOT_FOUND), ('add map', MASTER_REJECTS_ADD)]):
+            self.assertFalse(haproxy_manager.add_ip_to_runtime_map(IP))
+
+    def test_master_socket_rejection_of_del_is_failure(self):
+        with with_socket([('get map', GET_FOUND), ('del map', MASTER_REJECTS_DEL)]):
+            self.assertFalse(haproxy_manager.remove_ip_from_runtime_map(IP))
+
+    def test_unknown_map_identifier_is_failure(self):
+        with with_socket([('get map', GET_NOT_FOUND), ('add map', UNKNOWN_MAP_IDENTIFIER)]):
+            self.assertFalse(haproxy_manager.add_ip_to_runtime_map(IP))
+        with with_socket([('get map', GET_FOUND), ('del map', UNKNOWN_MAP_IDENTIFIER)]):
+            self.assertFalse(haproxy_manager.remove_ip_from_runtime_map(IP))
+
+    def test_missing_value_rejection_is_failure(self):
+        """Carries no marker word at all -- caught by 'a mutation says nothing'."""
+        with with_socket([('get map', GET_NOT_FOUND), ('add map', ADD_MAP_MISSING_VALUE)]):
+            self.assertFalse(haproxy_manager.add_ip_to_runtime_map(IP))
+
+    def test_silent_noop_add_is_failure(self):
+        """The `#0` failure mode: accepted, empty reply, nothing added.
+
+        Nothing in the response distinguishes this from success. Only the
+        read-back does -- which is why the read-back is not optional.
+        """
+        with with_socket([('get map', GET_NOT_FOUND), ('add map', MUTATION_OK)]):
+            self.assertFalse(haproxy_manager.add_ip_to_runtime_map(IP))
+
+    def test_add_that_lands_without_a_value_is_failure(self):
+        with with_socket([('get map', GET_FOUND_NO_VALUE), ('add map', MUTATION_OK)]):
+            self.assertFalse(haproxy_manager.add_ip_to_runtime_map(IP))
+
+    def test_del_that_leaves_the_key_behind_is_failure(self):
+        with with_socket([('get map', GET_FOUND), ('del map', MUTATION_OK)]):
+            self.assertFalse(haproxy_manager.remove_ip_from_runtime_map(IP))
+
+    def test_key_not_found_on_del_is_the_requested_end_state(self):
+        """Not a failure: the runtime map already lacks the key."""
+        with with_socket([('get map', GET_NOT_FOUND), ('del map', KEY_NOT_FOUND)]):
+            self.assertTrue(haproxy_manager.remove_ip_from_runtime_map(IP))
+
+    def test_verified_success_is_reported_as_success(self):
+        with with_socket(HAPPY_ADD):
+            self.assertTrue(haproxy_manager.add_ip_to_runtime_map(IP))
+        with with_socket(HAPPY_DEL):
+            self.assertTrue(haproxy_manager.remove_ip_from_runtime_map(IP))
+
+    def test_a_runtime_failure_never_raises_into_the_request_handler(self):
+        """The map file + reload still enforces the block; degrade, don't 500."""
+        with with_socket([('get map', GET_NOT_FOUND), ('add map', MASTER_REJECTS_ADD)]):
+            self.assertIs(haproxy_manager.add_ip_to_runtime_map(IP), False)
+
+    def test_mutations_are_checked_for_an_empty_body_not_a_marker_list(self):
+        """_HAPROXY_CLI_ERROR_MARKERS can only know rejections already seen."""
+        with self.assertRaises(haproxy_manager.HaproxyCliError):
+            with with_socket([('add map', 'something nobody has ever seen\n')]):
+                haproxy_manager.haproxy_cli('add map %s x 1' % MAP,
+                                            worker=True, expect_empty=True)
+
+    def test_the_new_error_markers_are_recognised(self):
+        for response in (UNKNOWN_MAP_IDENTIFIER, KEY_NOT_FOUND,
+                         MASTER_REJECTS_ADD):
+            self.assertTrue(haproxy_manager._cli_response_is_error(response),
+                            '%r must be classified as an error' % response[:40])
+        for response in (GET_FOUND, GET_NOT_FOUND, SHOW_MAP_SAMPLE):
+            self.assertFalse(haproxy_manager._cli_response_is_error(response),
+                             '%r is data, not an error' % response[:40])
+
+    def test_socat_exit_zero_carries_no_information(self):
+        """FakeSocket never signals failure any other way, and neither did socat."""
+        with with_socket([('get map', GET_NOT_FOUND), ('add map', MASTER_REJECTS_ADD)]) as fake:
+            self.assertFalse(haproxy_manager.add_ip_to_runtime_map(IP))
+        self.assertTrue(fake.sent, 'the command was sent and "succeeded" at the '
+                                   'process level; only the body says otherwise')
+
+
+class ReadBack(unittest.TestCase):
+    """The read-back primitives the guarantees above rest on."""
+
+    def test_lookup_reports_found_with_value(self):
+        with with_socket([('get map', GET_FOUND)]):
+            self.assertEqual(haproxy_manager.runtime_map_lookup(MAP, IP),
+                             (True, '1'))
+
+    def test_lookup_reports_not_found(self):
+        with with_socket([('get map', GET_NOT_FOUND)]):
+            self.assertEqual(haproxy_manager.runtime_map_lookup(MAP, IP),
+                             (False, None))
+
+    def test_lookup_raises_on_a_rejected_reference(self):
+        with with_socket([('get map', UNKNOWN_MAP_IDENTIFIER)]):
+            with self.assertRaises(haproxy_manager.HaproxyCliError):
+                haproxy_manager.runtime_map_lookup(MAP, IP)
+
+    def test_keys_parses_show_map_output(self):
+        with with_socket([('show map', SHOW_MAP_SAMPLE)]):
+            self.assertEqual(
+                haproxy_manager.runtime_map_keys(MAP),
+                {'101.36.109.130', '101.47.140.218', IP})
+
+    def test_empty_map_is_not_a_rejection(self):
+        with with_socket([('show map', '')]):
+            self.assertEqual(haproxy_manager.runtime_map_keys(MAP), set())
+
+    def test_rejected_show_map_still_raises(self):
+        with with_socket([('show map', UNKNOWN_MAP_IDENTIFIER)]):
+            with self.assertRaises(haproxy_manager.HaproxyCliError):
+                haproxy_manager.runtime_map_keys(MAP)
+
+
+MAP_BY_ID_RE = re.compile(r'\b(?:add|del|clear|show|get)\s+map\s+#')
+
+
+class NoSourceBuildsAMapIdCommand(unittest.TestCase):
+    """Guard 4: `map #<id>` may be described in comments, never executed.
+
+    Scanning string literals rather than raw text is deliberate -- the whole
+    reason this bug is documented at length in the source is so the next reader
+    does not reintroduce it, and a plain grep would fail on those comments.
+    """
+
+    def test_no_python_string_literal_builds_a_map_id_command(self):
+        tree = ast.parse(io.open('haproxy_manager.py', encoding='utf-8').read())
+        offenders = [
+            node.value for node in ast.walk(tree)
+            if isinstance(node, ast.Constant) and isinstance(node.value, str)
+            and MAP_BY_ID_RE.search(node.value)
+            and not (ast.get_docstring(tree) == node.value)
+        ]
+        # Docstrings are string literals too; exclude any literal that is a
+        # docstring of a module/class/function.
+        docstrings = set()
+        for node in ast.walk(tree):
+            if isinstance(node, (ast.Module, ast.ClassDef, ast.FunctionDef,
+                                 ast.AsyncFunctionDef)):
+                doc = ast.get_docstring(node, clean=False)
+                if doc:
+                    docstrings.add(doc)
+        offenders = [o for o in offenders if o not in docstrings]
+        self.assertEqual(offenders, [],
+                         'these string literals build a map command with an '
+                         'unstable #<id> reference')
+
+    def test_no_shell_or_template_code_line_uses_a_map_id(self):
+        targets = (glob.glob('scripts/*.sh') + glob.glob('templates/*.tpl')
+                   + ['Dockerfile'])
+        offenders = []
+        for path in targets:
+            if not os.path.exists(path):
+                continue
+            for lineno, line in enumerate(
+                    io.open(path, encoding='utf-8').read().splitlines(), 1):
+                if line.lstrip().startswith('#'):
+                    continue  # a comment describing the old form is fine
+                if MAP_BY_ID_RE.search(line):
+                    offenders.append('%s:%d: %s' % (path, lineno, line.strip()))
+        self.assertEqual(offenders, [],
+                         'map ids are assigned at config-parse time and move; '
+                         'reference the map file by path')
+
+
+if __name__ == '__main__':
+    unittest.main(verbosity=2)

templates/hap_backend.tpl

@@ -4,7 +4,7 @@ backend {{ name }}-backend
     option forwardfor
     # Pass the real client IP to backend (from proxy headers or direct connection)
     # This is crucial for container-level logging and security tools
-    http-request set-header X-CLIENT-IP %[var(txn.real_ip)]
+    http-request add-header X-CLIENT-IP %[var(txn.real_ip)]
     http-request set-header X-Real-IP %[var(txn.real_ip)]
     http-request set-header X-Forwarded-For %[var(txn.real_ip)]
     http-request set-header X-Forwarded-Proto https if { ssl_fc }
@@ -29,7 +29,7 @@ backend {{ name }}-sse-backend
 
     option forwardfor
     # Pass the real client IP to backend (from proxy headers or direct connection)
-    http-request set-header X-CLIENT-IP %[var(txn.real_ip)]
+    http-request add-header X-CLIENT-IP %[var(txn.real_ip)]
     http-request set-header X-Real-IP %[var(txn.real_ip)]
     http-request set-header X-Forwarded-For %[var(txn.real_ip)]
     http-request set-header X-Forwarded-Proto https if { ssl_fc }

templates/hap_backend_http_check.tpl

@@ -4,7 +4,7 @@ backend {{ name }}-backend
     option httpchk
     # Pass the real client IP to backend (from proxy headers or direct connection)
     # This is crucial for container-level logging and security tools
-    http-request set-header X-CLIENT-IP %[var(txn.real_ip)]
+    http-request add-header X-CLIENT-IP %[var(txn.real_ip)]
     http-request set-header X-Real-IP %[var(txn.real_ip)]
     http-request set-header X-Forwarded-For %[var(txn.real_ip)]
     {% if ssl_enabled %}http-request set-header X-Forwarded-Proto https if { ssl_fc }{% endif %}

templates/hap_backend_longlived.tpl

@@ -14,7 +14,7 @@ backend {{ name }}-backend
     timeout tunnel 6h
     timeout http-keep-alive 6h
     option forwardfor
-    http-request set-header X-CLIENT-IP %[var(txn.real_ip)]
+    http-request add-header X-CLIENT-IP %[var(txn.real_ip)]
     http-request set-header X-Real-IP %[var(txn.real_ip)]
     http-request set-header X-Forwarded-For %[var(txn.real_ip)]
     http-request set-header X-Forwarded-Proto https if { ssl_fc }
@@ -31,7 +31,7 @@ backend {{ name }}-sse-backend
     timeout tunnel 6h
     timeout http-keep-alive 6h
     option forwardfor
-    http-request set-header X-CLIENT-IP %[var(txn.real_ip)]
+    http-request add-header X-CLIENT-IP %[var(txn.real_ip)]
     http-request set-header X-Real-IP %[var(txn.real_ip)]
     http-request set-header X-Forwarded-For %[var(txn.real_ip)]
     http-request set-header X-Forwarded-Proto https if { ssl_fc }

templates/hap_backend_websocket.tpl

@@ -7,7 +7,7 @@ backend {{ name }}-backend
     timeout tunnel 6h
     timeout http-keep-alive 6h
     option forwardfor
-    http-request set-header X-CLIENT-IP %[var(txn.real_ip)]
+    http-request add-header X-CLIENT-IP %[var(txn.real_ip)]
     http-request set-header X-Real-IP %[var(txn.real_ip)]
     http-request set-header X-Forwarded-For %[var(txn.real_ip)]
     http-request set-header X-Forwarded-Proto https if { ssl_fc }
@@ -24,7 +24,7 @@ backend {{ name }}-sse-backend
     timeout tunnel 6h
     timeout http-keep-alive 6h
     option forwardfor
-    http-request set-header X-CLIENT-IP %[var(txn.real_ip)]
+    http-request add-header X-CLIENT-IP %[var(txn.real_ip)]
     http-request set-header X-Real-IP %[var(txn.real_ip)]
     http-request set-header X-Forwarded-For %[var(txn.real_ip)]
     http-request set-header X-Forwarded-Proto https if { ssl_fc }

templates/hap_listener.tpl

@@ -582,7 +582,9 @@ frontend web
 
     # IP blocking using map file (manual blocks only)
     # Map file format: /etc/haproxy/blocked_ips.map contains "<ip_or_cidr> 1" per line
-    # Runtime updates: echo "add map #0 IP_ADDRESS 1" | socat stdio /var/run/haproxy.sock
+    # Runtime updates (worker command, map referenced by PATH -- "#<id>" ids
+    # move on every config regeneration and "#0" silently adds nothing):
+    #   echo "@1 add map /etc/haproxy/blocked_ips.map IP_ADDRESS 1" | socat stdio /tmp/haproxy-cli
     # Checks the real client IP (from headers if present, otherwise src)
     # map_ip() converter supports both single IPs and CIDR ranges (e.g., 192.168.1.0/24)
     acl is_blocked_ip var(txn.real_ip),map_ip(/etc/haproxy/blocked_ips.map,0) -m int gt 0