Enhance resource management: add close methods for S3Client and HECClient, and improve IMAP connection handling during IDLE. Update CHANGELOG.md for config reload improvements and bug fixes.

* Initial plan

* Address review feedback: Kafka SSL context, SIGHUP handler safety, test formatting

Co-authored-by: seanthegeek <44679+seanthegeek@users.noreply.github.com>
Agent-Logs-Url: https://github.com/domainaware/parsedmarc/sessions/8f2fd48f-32a4-4258-9a89-06f7c7ac29bf

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: seanthegeek <44679+seanthegeek@users.noreply.github.com>

.claude/settings.json

@@ -0,0 +1,17 @@
+{
+  "permissions": {
+    "allow": [
+      "Bash(python -c \"import py_compile; py_compile.compile\\(''parsedmarc/cli.py'', doraise=True\\)\")",
+      "Bash(ruff check:*)",
+      "Bash(ruff format:*)",
+      "Bash(GITHUB_ACTIONS=true pytest --cov tests.py)",
+      "Bash(ls tests*)",
+      "Bash(GITHUB_ACTIONS=true python -m pytest --cov tests.py -x)",
+      "Bash(GITHUB_ACTIONS=true python -m pytest tests.py -x -v)",
+      "Bash(python -m pytest tests.py --no-header -q)"
+    ],
+    "additionalDirectories": [
+      "/tmp"
+    ]
+  }
+}

.gitignore

@@ -137,7 +137,7 @@ samples/private
 *.html
 *.sqlite-journal
 
-parsedmarc.ini
+parsedmarc*.ini
 scratch.py
 
 parsedmarc/resources/maps/base_reverse_dns.csv

CHANGELOG.md

@@ -1,5 +1,45 @@
 # Changelog
 
+## 9.3.0
+
+### Added
+
+- SIGHUP-based configuration reload for watch mode — update output
+  destinations, DNS/GeoIP settings, processing flags, and log level
+  without restarting the service or interrupting in-progress report
+  processing. Use `systemctl reload parsedmarc` when running under
+  systemd. On a successful reload, old output clients are closed and
+  recreated. On a failed reload, the previous configuration remains
+  fully active.
+- `close()` methods on GelfClient, KafkaClient, SyslogClient,
+  WebhookClient, HECClient, and S3Client for clean resource teardown
+  on reload.
+- `should_reload` parameter on all `MailboxConnection.watch()`
+  implementations and `watch_inbox()` to ensure SIGHUP never triggers
+  a new email batch mid-reload.
+- Elasticsearch and OpenSearch connections are now tracked and cleaned
+  up on reload via `_close_output_clients()`.
+- Extracted `_parse_config_file()` and `_init_output_clients()` from
+  `_main()` in `cli.py` to support config reload and reduce code
+  duplication.
+
+### Fixed
+
+- `get_index_prefix()` crashed on forensic reports with `TypeError`
+  due to `report()` instead of `report[]` dict access.
+- Missing `exit(1)` after IMAP user/password validation failure
+  allowed execution to continue with `None` credentials.
+- IMAP `watch()` leaked a connection on every IDLE cycle by not
+  closing the old `IMAPClient` before replacing it.
+- Resource leak in `_init_output_clients()` when Splunk HEC
+  configuration is invalid — the partially-constructed HEC client
+  is now cleaned up on error.
+- Elasticsearch/OpenSearch `set_hosts()` global state was not
+  rollback-safe on reload failure — init now runs last so other
+  client failures don't leave stale global connections.
+- `active_log_file` was not initialized at startup, causing the
+  first reload to unnecessarily remove and re-add the FileHandler.
+
 ## 9.2.1
 
 ### Added

docker-compose.dashboard-dev.yml

@@ -0,0 +1,45 @@
+name: parsedmarc-dashboards
+
+include:
+  - docker-compose.yml
+
+services:
+  kibana:
+    image: docker.elastic.co/kibana/kibana:8.19.7
+    environment:
+      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
+    ports:
+      - "127.0.0.1:5601:5601"
+    depends_on:
+      elasticsearch:
+        condition: service_healthy
+
+  opensearch-dashboards:
+    image: opensearchproject/opensearch-dashboards:2
+    environment:
+      - OPENSEARCH_HOSTS=["https://opensearch:9200"]
+    ports:
+      - "127.0.0.1:5602:5601"
+    depends_on:
+      opensearch:
+        condition: service_healthy
+
+  grafana:
+    image: grafana/grafana:latest
+    environment:
+      - GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD}
+      - GF_INSTALL_PLUGINS=grafana-piechart-panel,grafana-worldmap-panel
+    ports:
+      - "127.0.0.1:3000:3000"
+    depends_on:
+      elasticsearch:
+        condition: service_healthy
+
+  splunk:
+    image: splunk/splunk:latest
+    environment:
+      - SPLUNK_START_ARGS=--accept-license
+      - "SPLUNK_GENERAL_TERMS=--accept-sgt-current-at-splunk-com"
+      - SPLUNK_PASSWORD=${SPLUNK_PASSWORD}
+    ports:
+      - "127.0.0.1:8000:8000"

docker-compose.yml

@@ -48,7 +48,7 @@ services:
       test:
         [
           "CMD-SHELL",
-          "curl -s -XGET http://localhost:9201/_cluster/health?pretty | grep status | grep -q '\\(green\\|yellow\\)'"
+          "curl -sk -u admin:${OPENSEARCH_INITIAL_ADMIN_PASSWORD} -XGET https://localhost:9200/_cluster/health?pretty | grep status | grep -q '\\(green\\|yellow\\)'"
         ]
       interval: 10s
       timeout: 10s

docs/source/usage.md

@@ -404,6 +404,7 @@ The full set of configuration options are:
   retry_attempts = 3
   retry_delay = 5
   ```
+
 - `gmail_api`
   - `credentials_file` - str: Path to file containing the
       credentials, None to disable (Default: `None`)
@@ -442,7 +443,7 @@ The full set of configuration options are:
   - `dcr_smtp_tls_stream` - str: The stream name for the SMTP TLS reports in the DCR
 
   :::{note}
-    Information regarding the setup of the Data Collection Rule can be found [here](https://learn.microsoft.com/en-us/azure/azure-monitor/logs/tutorial-logs-ingestion-portal).
+    Information regarding the setup of the Data Collection Rule can be found [in the Azure documentation](https://learn.microsoft.com/en-us/azure/azure-monitor/logs/tutorial-logs-ingestion-portal).
     :::
 - `gelf`
   - `host` - str: The GELF server name or IP address
@@ -602,6 +603,7 @@ After=network.target network-online.target elasticsearch.service
 
 [Service]
 ExecStart=/opt/parsedmarc/venv/bin/parsedmarc -c /etc/parsedmarc.ini
+ExecReload=/bin/kill -HUP $MAINPID
 User=parsedmarc
 Group=parsedmarc
 Restart=always
@@ -634,6 +636,51 @@ sudo service parsedmarc restart
 
 :::
 
+### Reloading configuration without restarting
+
+When running in watch mode, `parsedmarc` supports reloading its
+configuration file without restarting the service or interrupting
+report processing that is already in progress. Send a `SIGHUP` signal
+to the process, or use `systemctl reload` if the unit file includes
+the `ExecReload` line shown above:
+
+```bash
+sudo systemctl reload parsedmarc
+```
+
+The reload takes effect after the current batch of reports finishes
+processing and all output operations (Elasticsearch, Kafka, S3, etc.)
+for that batch have completed. The following settings are reloaded:
+
+- All output destinations (Elasticsearch, OpenSearch, Kafka, S3,
+  Splunk, syslog, GELF, webhooks, Log Analytics)
+- Multi-tenant index prefix domain map (`index_prefix_domain_map` —
+  the referenced YAML file is re-read on reload)
+- DNS and GeoIP settings (`nameservers`, `dns_timeout`, `ip_db_path`,
+  `offline`, etc.)
+- Processing flags (`strip_attachment_payloads`, `batch_size`,
+  `check_timeout`, etc.)
+- Log level (`debug`, `verbose`, `warnings`, `silent`)
+
+Mailbox connection settings (IMAP host/credentials, Microsoft Graph,
+Gmail API, Maildir path) are **not** reloaded — changing those still
+requires a full restart.
+
+On a **successful** reload, existing output client connections are
+closed and new ones are created from the updated configuration. The
+service then resumes watching with the new settings.
+
+If the new configuration file contains errors (missing required
+settings, unreachable output destinations, etc.), the **entire reload
+is aborted** — no output clients are replaced and the previous
+configuration remains fully active. This means a typo in one section
+will not take down an otherwise working setup. Check the logs for
+details:
+
+```bash
+journalctl -u parsedmarc.service -r
+```
+
 To check the status of the service, run:
 
 ```bash

parsedmarc/__init__.py

@@ -2195,6 +2195,7 @@ def watch_inbox(
     batch_size: int = 10,
     since: Optional[Union[datetime, date, str]] = None,
     normalize_timespan_threshold_hours: float = 24,
+    should_reload: Optional[Callable] = None,
 ):
     """
     Watches the mailbox for new messages and
@@ -2222,6 +2223,8 @@ def watch_inbox(
         batch_size (int): Number of messages to read and process before saving
         since: Search for messages since certain time
         normalize_timespan_threshold_hours (float): Normalize timespans beyond this
+        should_reload: Optional callable that returns True when a config
+            reload has been requested (e.g. via SIGHUP)
     """
 
     def check_callback(connection):
@@ -2246,7 +2249,14 @@ def watch_inbox(
         )
         callback(res)
 
-    mailbox_connection.watch(check_callback=check_callback, check_timeout=check_timeout)
+    watch_kwargs: dict = {
+        "check_callback": check_callback,
+        "check_timeout": check_timeout,
+    }
+    if should_reload is not None:
+        watch_kwargs["should_reload"] = should_reload
+
+    mailbox_connection.watch(**watch_kwargs)
 
 
 def append_json(

parsedmarc/gelf.py

@@ -69,3 +69,8 @@ class GelfClient(object):
         for row in rows:
             log_context_data.parsedmarc = row
             self.logger.info("parsedmarc smtptls report")
+
+    def close(self):
+        """Remove and close the GELF handler, releasing its connection."""
+        self.logger.removeHandler(self.handler)
+        self.handler.close()

parsedmarc/kafkaclient.py

@@ -62,6 +62,10 @@ class KafkaClient(object):
         except NoBrokersAvailable:
             raise KafkaError("No Kafka brokers available")
 
+    def close(self):
+        """Close the Kafka producer, releasing background threads and sockets."""
+        self.producer.close()
+
     @staticmethod
     def strip_metadata(report: dict[str, Any]):
         """

parsedmarc/mail/gmail.py

@@ -175,10 +175,14 @@ class GmailConnection(MailboxConnection):
         # Not needed
         pass
 
-    def watch(self, check_callback, check_timeout):
+    def watch(self, check_callback, check_timeout, should_reload=None):
         """Checks the mailbox for new messages every n seconds"""
         while True:
+            if should_reload and should_reload():
+                return
             sleep(check_timeout)
+            if should_reload and should_reload():
+                return
             check_callback(self)
 
     @lru_cache(maxsize=10)

parsedmarc/mail/graph.py

@@ -278,10 +278,14 @@ class MSGraphConnection(MailboxConnection):
         # Not needed
         pass
 
-    def watch(self, check_callback, check_timeout):
+    def watch(self, check_callback, check_timeout, should_reload=None):
         """Checks the mailbox for new messages every n seconds"""
         while True:
+            if should_reload and should_reload():
+                return
             sleep(check_timeout)
+            if should_reload and should_reload():
+                return
             check_callback(self)
 
     @lru_cache(maxsize=10)

parsedmarc/mail/imap.py

@@ -81,7 +81,7 @@ class IMAPConnection(MailboxConnection):
     def keepalive(self):
         self._client.noop()
 
-    def watch(self, check_callback, check_timeout):
+    def watch(self, check_callback, check_timeout, should_reload=None):
         """
         Use an IDLE IMAP connection to parse incoming emails,
         and pass the results to a callback function
@@ -90,10 +90,17 @@ class IMAPConnection(MailboxConnection):
         # IDLE callback sends IMAPClient object,
         # send back the imap connection object instead
         def idle_callback_wrapper(client: IMAPClient):
+            old_client = self._client
             self._client = client
+            try:
+                old_client.logout()
+            except Exception:
+                pass
             check_callback(self)
 
         while True:
+            if should_reload and should_reload():
+                return
             try:
                 IMAPClient(
                     host=self._client.host,
@@ -111,3 +118,5 @@ class IMAPConnection(MailboxConnection):
             except Exception as e:
                 logger.warning("IMAP connection error. {0}. Reconnecting...".format(e))
                 sleep(check_timeout)
+            if should_reload and should_reload():
+                return

parsedmarc/mail/mailbox_connection.py

@@ -28,5 +28,5 @@ class MailboxConnection(ABC):
     def keepalive(self):
         raise NotImplementedError
 
-    def watch(self, check_callback, check_timeout):
+    def watch(self, check_callback, check_timeout, should_reload=None):
         raise NotImplementedError

parsedmarc/mail/maildir.py

@@ -63,10 +63,14 @@ class MaildirConnection(MailboxConnection):
     def keepalive(self):
         return
 
-    def watch(self, check_callback, check_timeout):
+    def watch(self, check_callback, check_timeout, should_reload=None):
         while True:
+            if should_reload and should_reload():
+                return
             try:
                 check_callback(self)
             except Exception as e:
                 logger.warning("Maildir init error. {0}".format(e))
+            if should_reload and should_reload():
+                return
             sleep(check_timeout)

parsedmarc/s3.py

@@ -93,3 +93,11 @@ class S3Client(object):
         self.bucket.put_object(
             Body=json.dumps(report), Key=object_path, Metadata=object_metadata
         )
+
+    def close(self):
+        """Clean up the boto3 resource."""
+        try:
+            if self.s3.meta is not None:
+                self.s3.meta.client.close()
+        except Exception:
+            pass

parsedmarc/splunk.py

@@ -207,3 +207,7 @@ class HECClient(object):
             raise SplunkError(e.__str__())
         if response["code"] != 0:
             raise SplunkError(response["text"])
+
+    def close(self):
+        """Close the underlying HTTP session."""
+        self.session.close()

parsedmarc/syslog.py

@@ -57,7 +57,7 @@ class SyslogClient(object):
         self.logger.setLevel(logging.INFO)
 
         # Create the appropriate syslog handler based on protocol
-        log_handler = self._create_syslog_handler(
+        self.log_handler = self._create_syslog_handler(
             server_name,
             server_port,
             self.protocol,
@@ -69,7 +69,7 @@ class SyslogClient(object):
             retry_delay,
         )
 
-        self.logger.addHandler(log_handler)
+        self.logger.addHandler(self.log_handler)
 
     def _create_syslog_handler(
         self,
@@ -179,3 +179,8 @@ class SyslogClient(object):
         rows = parsed_smtp_tls_reports_to_csv_rows(smtp_tls_reports)
         for row in rows:
             self.logger.info(json.dumps(row))
+
+    def close(self):
+        """Remove and close the syslog handler, releasing its socket."""
+        self.logger.removeHandler(self.log_handler)
+        self.log_handler.close()

parsedmarc/webhook.py

@@ -63,3 +63,7 @@ class WebhookClient(object):
             self.session.post(webhook_url, data=payload, timeout=self.timeout)
         except Exception as error_:
             logger.error("Webhook Error: {0}".format(error_.__str__()))
+
+    def close(self):
+        """Close the underlying HTTP session."""
+        self.session.close()

tests.py

@@ -4,6 +4,7 @@
 from __future__ import absolute_import, print_function, unicode_literals
 
 import os
+import signal
 import sys
 import tempfile
 import unittest
@@ -1277,7 +1278,7 @@ class TestMailboxWatchSince(unittest.TestCase):
     def testWatchInboxPassesSinceToMailboxFetch(self):
         mailbox_connection = SimpleNamespace()
 
-        def fake_watch(check_callback, check_timeout):
+        def fake_watch(check_callback, check_timeout, should_reload=None):
             check_callback(mailbox_connection)
             raise _BreakLoop()
 
@@ -1445,7 +1446,7 @@ mailbox = shared@example.com
                 parsedmarc.cli._main()
 
         self.assertEqual(system_exit.exception.code, -1)
-        mock_logger.critical.assert_called_once_with(
+        mock_logger.error.assert_called_once_with(
             "certificate_path setting missing from the msgraph config section"
         )
         mock_graph_connection.assert_not_called()
@@ -1517,7 +1518,7 @@ user = owner@example.com
                 parsedmarc.cli._main()
 
         self.assertEqual(system_exit.exception.code, -1)
-        mock_logger.critical.assert_called_once_with(
+        mock_logger.error.assert_called_once_with(
             "password setting missing from the msgraph config section"
         )
         mock_graph_connection.assert_not_called()
@@ -1674,7 +1675,7 @@ mailbox = shared@example.com
                 parsedmarc.cli._main()
 
         self.assertEqual(system_exit.exception.code, -1)
-        mock_logger.critical.assert_called_once_with(
+        mock_logger.error.assert_called_once_with(
             "client_secret setting missing from the msgraph config section"
         )
         mock_graph_connection.assert_not_called()
@@ -1706,7 +1707,7 @@ mailbox = shared@example.com
                 parsedmarc.cli._main()
 
         self.assertEqual(system_exit.exception.code, -1)
-        mock_logger.critical.assert_called_once_with(
+        mock_logger.error.assert_called_once_with(
             "tenant_id setting missing from the msgraph config section"
         )
         mock_graph_connection.assert_not_called()
@@ -1738,7 +1739,7 @@ tenant_id = tenant-id
                 parsedmarc.cli._main()
 
         self.assertEqual(system_exit.exception.code, -1)
-        mock_logger.critical.assert_called_once_with(
+        mock_logger.error.assert_called_once_with(
             "mailbox setting missing from the msgraph config section"
         )
         mock_graph_connection.assert_not_called()
@@ -1808,7 +1809,7 @@ mailbox = shared@example.com
                 parsedmarc.cli._main()
 
         self.assertEqual(system_exit.exception.code, -1)
-        mock_logger.critical.assert_called_once_with(
+        mock_logger.error.assert_called_once_with(
             "tenant_id setting missing from the msgraph config section"
         )
         mock_graph_connection.assert_not_called()
@@ -1839,7 +1840,7 @@ tenant_id = tenant-id
                 parsedmarc.cli._main()
 
         self.assertEqual(system_exit.exception.code, -1)
-        mock_logger.critical.assert_called_once_with(
+        mock_logger.error.assert_called_once_with(
             "mailbox setting missing from the msgraph config section"
         )
         mock_graph_connection.assert_not_called()
@@ -1871,7 +1872,7 @@ certificate_path = /tmp/msgraph-cert.pem
                 parsedmarc.cli._main()
 
         self.assertEqual(system_exit.exception.code, -1)
-        mock_logger.critical.assert_called_once_with(
+        mock_logger.error.assert_called_once_with(
             "tenant_id setting missing from the msgraph config section"
         )
         mock_graph_connection.assert_not_called()
@@ -1903,12 +1904,327 @@ certificate_path = /tmp/msgraph-cert.pem
                 parsedmarc.cli._main()
 
         self.assertEqual(system_exit.exception.code, -1)
-        mock_logger.critical.assert_called_once_with(
+        mock_logger.error.assert_called_once_with(
             "mailbox setting missing from the msgraph config section"
         )
         mock_graph_connection.assert_not_called()
         mock_get_mailbox_reports.assert_not_called()
 
 
+class TestSighupReload(unittest.TestCase):
+    """Tests for SIGHUP-driven configuration reload in watch mode."""
+
+    _BASE_CONFIG = """[general]
+silent = true
+
+[imap]
+host = imap.example.com
+user = user
+password = pass
+
+[mailbox]
+watch = true
+"""
+
+    @unittest.skipUnless(
+        hasattr(signal, "SIGHUP"),
+        "SIGHUP not available on this platform",
+    )
+    @patch("parsedmarc.cli._init_output_clients")
+    @patch("parsedmarc.cli._parse_config_file")
+    @patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
+    @patch("parsedmarc.cli.watch_inbox")
+    @patch("parsedmarc.cli.IMAPConnection")
+    def testSighupTriggersReloadAndWatchRestarts(
+        self,
+        mock_imap,
+        mock_watch,
+        mock_get_reports,
+        mock_parse_config,
+        mock_init_clients,
+    ):
+        """SIGHUP causes watch to return, config is re-parsed, and watch restarts."""
+        import signal as signal_module
+
+        mock_imap.return_value = object()
+        mock_get_reports.return_value = {
+            "aggregate_reports": [],
+            "forensic_reports": [],
+            "smtp_tls_reports": [],
+        }
+
+        def parse_side_effect(config_file, opts):
+            opts.imap_host = "imap.example.com"
+            opts.imap_user = "user"
+            opts.imap_password = "pass"
+            opts.mailbox_watch = True
+            return None
+
+        mock_parse_config.side_effect = parse_side_effect
+        mock_init_clients.return_value = {}
+
+        call_count = [0]
+
+        def watch_side_effect(*args, **kwargs):
+            call_count[0] += 1
+            if call_count[0] == 1:
+                # Simulate SIGHUP arriving while watch is running
+                if hasattr(signal_module, "SIGHUP"):
+                    import os
+
+                    os.kill(os.getpid(), signal_module.SIGHUP)
+                return  # Normal return — reload loop will continue
+            else:
+                raise FileExistsError("stop-watch-loop")
+
+        mock_watch.side_effect = watch_side_effect
+
+        with tempfile.NamedTemporaryFile("w", suffix=".ini", delete=False) as cfg:
+            cfg.write(self._BASE_CONFIG)
+            cfg_path = cfg.name
+        self.addCleanup(lambda: os.path.exists(cfg_path) and os.remove(cfg_path))
+
+        with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path]):
+            with self.assertRaises(SystemExit) as cm:
+                parsedmarc.cli._main()
+
+        # Exited with code 1 (from FileExistsError handler)
+        self.assertEqual(cm.exception.code, 1)
+        # watch_inbox was called twice: initial run + after reload
+        self.assertEqual(mock_watch.call_count, 2)
+        # _parse_config_file called for initial load + reload
+        self.assertGreaterEqual(mock_parse_config.call_count, 2)
+
+    @unittest.skipUnless(
+        hasattr(signal, "SIGHUP"),
+        "SIGHUP not available on this platform",
+    )
+    @patch("parsedmarc.cli._init_output_clients")
+    @patch("parsedmarc.cli._parse_config_file")
+    @patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
+    @patch("parsedmarc.cli.watch_inbox")
+    @patch("parsedmarc.cli.IMAPConnection")
+    def testInvalidConfigOnReloadKeepsPreviousState(
+        self,
+        mock_imap,
+        mock_watch,
+        mock_get_reports,
+        mock_parse_config,
+        mock_init_clients,
+    ):
+        """A failing reload leaves opts and clients unchanged."""
+        import signal as signal_module
+
+        mock_imap.return_value = object()
+        mock_get_reports.return_value = {
+            "aggregate_reports": [],
+            "forensic_reports": [],
+            "smtp_tls_reports": [],
+        }
+
+        # Initial parse sets required opts; reload parse raises
+        initial_map = {"prefix_": ["example.com"]}
+        call_count = [0]
+
+        def parse_side_effect(config_file, opts):
+            call_count[0] += 1
+            opts.imap_host = "imap.example.com"
+            opts.imap_user = "user"
+            opts.imap_password = "pass"
+            opts.mailbox_watch = True
+            if call_count[0] == 1:
+                return initial_map
+            raise RuntimeError("bad config")
+
+        mock_parse_config.side_effect = parse_side_effect
+
+        initial_clients = {"s3_client": MagicMock()}
+        mock_init_clients.return_value = initial_clients
+
+        watch_calls = [0]
+
+        def watch_side_effect(*args, **kwargs):
+            watch_calls[0] += 1
+            if watch_calls[0] == 1:
+                if hasattr(signal_module, "SIGHUP"):
+                    import os
+
+                    os.kill(os.getpid(), signal_module.SIGHUP)
+                return
+            else:
+                raise FileExistsError("stop")
+
+        mock_watch.side_effect = watch_side_effect
+
+        with tempfile.NamedTemporaryFile("w", suffix=".ini", delete=False) as cfg:
+            cfg.write(self._BASE_CONFIG)
+            cfg_path = cfg.name
+        self.addCleanup(lambda: os.path.exists(cfg_path) and os.remove(cfg_path))
+
+        with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path]):
+            with self.assertRaises(SystemExit) as cm:
+                parsedmarc.cli._main()
+
+        self.assertEqual(cm.exception.code, 1)
+        # watch was still called twice (reload loop continued after failed reload)
+        self.assertEqual(mock_watch.call_count, 2)
+        # The failed reload must not have closed the original clients
+        initial_clients["s3_client"].close.assert_not_called()
+
+    @unittest.skipUnless(
+        hasattr(signal, "SIGHUP"),
+        "SIGHUP not available on this platform",
+    )
+    @patch("parsedmarc.cli._init_output_clients")
+    @patch("parsedmarc.cli._parse_config_file")
+    @patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
+    @patch("parsedmarc.cli.watch_inbox")
+    @patch("parsedmarc.cli.IMAPConnection")
+    def testReloadClosesOldClients(
+        self,
+        mock_imap,
+        mock_watch,
+        mock_get_reports,
+        mock_parse_config,
+        mock_init_clients,
+    ):
+        """Successful reload closes the old output clients before replacing them."""
+        import signal as signal_module
+
+        mock_imap.return_value = object()
+        mock_get_reports.return_value = {
+            "aggregate_reports": [],
+            "forensic_reports": [],
+            "smtp_tls_reports": [],
+        }
+
+        def parse_side_effect(config_file, opts):
+            opts.imap_host = "imap.example.com"
+            opts.imap_user = "user"
+            opts.imap_password = "pass"
+            opts.mailbox_watch = True
+            return None
+
+        mock_parse_config.side_effect = parse_side_effect
+
+        old_client = MagicMock()
+        new_client = MagicMock()
+        init_call = [0]
+
+        def init_side_effect(opts):
+            init_call[0] += 1
+            if init_call[0] == 1:
+                return {"kafka_client": old_client}
+            return {"kafka_client": new_client}
+
+        mock_init_clients.side_effect = init_side_effect
+
+        watch_calls = [0]
+
+        def watch_side_effect(*args, **kwargs):
+            watch_calls[0] += 1
+            if watch_calls[0] == 1:
+                if hasattr(signal_module, "SIGHUP"):
+                    import os
+
+                    os.kill(os.getpid(), signal_module.SIGHUP)
+                return
+            else:
+                raise FileExistsError("stop")
+
+        mock_watch.side_effect = watch_side_effect
+
+        with tempfile.NamedTemporaryFile("w", suffix=".ini", delete=False) as cfg:
+            cfg.write(self._BASE_CONFIG)
+            cfg_path = cfg.name
+        self.addCleanup(lambda: os.path.exists(cfg_path) and os.remove(cfg_path))
+
+        with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path]):
+            with self.assertRaises(SystemExit):
+                parsedmarc.cli._main()
+
+        # Old client must have been closed when reload succeeded
+        old_client.close.assert_called_once()
+
+    @unittest.skipUnless(
+        hasattr(signal, "SIGHUP"),
+        "SIGHUP not available on this platform",
+    )
+    @patch("parsedmarc.cli._init_output_clients")
+    @patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
+    @patch("parsedmarc.cli.watch_inbox")
+    @patch("parsedmarc.cli.IMAPConnection")
+    def testRemovedConfigSectionTakesEffectOnReload(
+        self,
+        mock_imap,
+        mock_watch,
+        mock_get_reports,
+        mock_init_clients,
+    ):
+        """Removing a config section on reload resets that option to its default."""
+        import signal as signal_module
+
+        mock_imap.return_value = object()
+        mock_get_reports.return_value = {
+            "aggregate_reports": [],
+            "forensic_reports": [],
+            "smtp_tls_reports": [],
+        }
+        mock_init_clients.return_value = {}
+
+        # First config sets kafka_hosts (with required topics); second removes it.
+        config_v1 = (
+            self._BASE_CONFIG
+            + "\n[kafka]\nhosts = kafka.example.com:9092\n"
+            + "aggregate_topic = dmarc_agg\n"
+            + "forensic_topic = dmarc_forensic\n"
+            + "smtp_tls_topic = smtp_tls\n"
+        )
+        config_v2 = self._BASE_CONFIG  # no [kafka] section
+
+        with tempfile.NamedTemporaryFile("w", suffix=".ini", delete=False) as cfg:
+            cfg.write(config_v1)
+            cfg_path = cfg.name
+        self.addCleanup(lambda: os.path.exists(cfg_path) and os.remove(cfg_path))
+
+        watch_calls = [0]
+
+        def watch_side_effect(*args, **kwargs):
+            watch_calls[0] += 1
+            if watch_calls[0] == 1:
+                # Rewrite config to remove kafka before triggering reload
+                with open(cfg_path, "w") as f:
+                    f.write(config_v2)
+                if hasattr(signal_module, "SIGHUP"):
+                    import os
+
+                    os.kill(os.getpid(), signal_module.SIGHUP)
+                return
+            else:
+                raise FileExistsError("stop")
+
+        mock_watch.side_effect = watch_side_effect
+
+        # Capture opts used on each _init_output_clients call
+        init_opts_captures = []
+
+        def init_side_effect(opts):
+            from argparse import Namespace as NS
+
+            init_opts_captures.append(NS(**vars(opts)))
+            return {}
+
+        mock_init_clients.side_effect = init_side_effect
+
+        with patch.object(sys, "argv", ["parsedmarc", "-c", cfg_path]):
+            with self.assertRaises(SystemExit):
+                parsedmarc.cli._main()
+
+        # First init: kafka_hosts should be set from v1 config
+        self.assertIsNotNone(init_opts_captures[0].kafka_hosts)
+        # Second init (after reload with v2 config): kafka_hosts should be None
+        self.assertIsNone(init_opts_captures[1].kafka_hosts)
+
+
 if __name__ == "__main__":
     unittest.main(verbosity=2)