Don't archive/delete mailbox messages until output destinations confirm the save (#863)

* Don't archive/delete mailbox messages until output destinations confirm the save (#242)

get_dmarc_reports_from_mailbox() gains a keyword-only save_callback,
invoked once per fetched batch before any message is deleted or moved.
A False return (or an exception) marks the batch unsaved: its messages
stay in the reports folder for retry, and the aggregate-report dedup
keys staged for that batch are dropped so the retry reparses instead of
skipping. watch_inbox() passes its callback through as save_callback,
so watch mode gets the same protection.

To bound duplicate delivery to destinations that don't deduplicate, a
new [mailbox] max_unsaved_retries option (default 2) caps retries: a
message whose batch has failed the initial attempt plus that many
retries moves to {archive_folder}/Unsaved -- never deleted, whatever
the delete options say. Counts are process-local, so the cap applies
across watch-mode checks; one-shot runs retry indefinitely, which is
the safe direction. A raising callback (the CLI's, under
fail_on_output_error) is counted against the cap the same way before
the exception is re-raised, since mailsuite's IMAP and Maildir watch
loops swallow exceptions and keep checking.

CLI: process_reports() now returns its output-error list and the
mailbox_save_callback adapter feeds that verdict to the library;
save_output() failures (the one uncaught destination) are recorded like
every other sink's; file/mbox-derived reports are saved in a separate
pass from mailbox batches so nothing is saved twice; the combined
results fed to email_results() are filtered by index_prefix_domain_map
explicitly, restoring the SMTP TLS filtering the emailed summary lost
when saving moved into per-batch callbacks.

Credit to @mkilijanek for the original approach in #823.

Fixes #242

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Address automated review feedback on #863

- Keep a message's failure counter until its move to Unsaved actually
  succeeds. The counter was popped when the message was classified
  over-cap, before the move was attempted, so a failed move handed the
  still-in-place message a fresh set of under-cap retries and duplicate
  deliveries; now the next failed save classifies it over-cap again and
  re-attempts the move. Regression-tested at cap 1, where the reset
  would observably leave the message in the INBOX instead of moving it.
- Use sys.exit(1) instead of the site-dependent exit() built-in in the
  two new ParserError handlers.
- Close the sample files opened by the new Maildir tests via context
  managers.
- Keep the docs' section-link text on one line and fix the adjacent
  pre-existing "a IMAP" typo.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Address Copilot re-review feedback on #863

- Document mailbox_save_callback's raise path: with fail_on_output_error
  it raises ParserError via process_reports() instead of returning
  False, and the library counts that as an unsaved batch too.
- Use str(error_) instead of error_.__str__() in the new File output
  handler.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Reject negative max_unsaved_retries with a ValueError

The option is user-configurable (INI/env/kwarg); a negative value
silently behaved like 0. Both get_dmarc_reports_from_mailbox() and
watch_inbox() now validate it at the door alongside the existing
test/delete guard. watch_inbox() validates before entering the watch
loop, because a ValueError raised inside a check would be swallowed
and endlessly retried by the IMAP and Maildir backends' per-check
exception handling instead of surfacing.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Fix two pre-existing bugs surfaced by review of #863

- MAGIC_JSON was b"\7b" -- the octal escape \7 (BEL) plus a literal
  "b", not 0x7B, the "{" every RFC 8259 JSON object begins with -- so
  extract_report() rejected plain uncompressed JSON. Every in-tree
  caller pre-guarded with its own zip/gzip or "{" check, masking the
  dead branch; the practical impact was extract_report() as a public
  API and application/tlsrpt+gzip attachments whose payload is really
  uncompressed JSON. Now b"\x7b".

- get_index_prefix() unconditionally indexed policies[0], while
  parse_smtp_tls_report_json() accepts a report whose policies list is
  empty -- an IndexError crash whenever index_prefix_domain_map was
  configured. An empty-policies report has no domain to map, so it is
  now treated as unmappable and excluded from prefix-mapped output.
  This code was moved into filter_smtp_tls_reports_for_index_prefix()
  by this PR, which widened its exposure to the email_results() path.

Both regression tests were verified to fail against the unfixed code.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Rewrap the max_unsaved_retries docstring paragraph

The ValueError sentence was inserted without reflowing, leaving a
112-character line in a docstring wrapped at ~76; ruff formats code,
not docstring prose, so it slipped through lint.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

CHANGELOG.md

@@ -8,10 +8,22 @@
 - **Added `ParserConfig`, a single frozen dataclass carrying every parsing/enrichment option** ([#503](https://github.com/domainaware/parsedmarc/issues/503)): offline mode, IP database path, reverse DNS map and PSL overrides paths/URLs, DNS nameservers/timeout/retries, `strip_attachment_payloads`, `normalize_timespan_threshold_hours`, and the three caches the parser and enrichment code share across calls (IP address info, seen aggregate report IDs, and the reverse DNS map). `parse_aggregate_report_xml()`, `parse_aggregate_report_file()`, `parse_failure_report()`, `parse_report_email()`, `parse_report_file()`, `get_dmarc_reports_from_mbox()`, `get_dmarc_reports_from_mailbox()`, and `watch_inbox()` all accept a keyword-only `config=` argument; when it's provided, the individual option keyword arguments are ignored in favor of the config's values, and every existing per-option keyword argument continues to work unchanged when `config=` is omitted. Every explicitly constructed `ParserConfig` owns fresh, isolated caches; omitting `config=` falls back to the existing process-wide shared default caches, unchanged and identity-preserved (`parsedmarc.IP_ADDRESS_CACHE`, `parsedmarc.SEEN_AGGREGATE_REPORT_IDS`, `parsedmarc.REVERSE_DNS_MAP`). Caches never cross multiprocessing worker boundaries: `ParserConfig.__getstate__` drops the three cache fields when a config is pickled for a worker, and each worker accumulates its own from that point on. `keep_alive` and `n_procs` remain separate keyword arguments — they control process/worker orchestration, not parsing or enrichment behavior, so they're deliberately not `ParserConfig` fields. See the new "Using parsedmarc as a library" section of the usage docs for an example.
 - **Added a "Domain policy" dropdown filter to the Splunk aggregate DMARC dashboard** ([#854](https://github.com/domainaware/parsedmarc/issues/854)): filters every panel on the published DMARC policy (`published_policy.p`) or subdomain policy (`published_policy.sp`), making it easy to see which domains with a `none` policy are ready to move to `quarantine` or `reject`. It sits immediately before the "Message disposition" dropdown and offers the same choices (`any`/`none`/`quarantine`/`reject`, defaulting to `any`).
 - **New `[mailbox]` options `delete_aggregate`, `delete_failure`, `delete_smtp_tls`, and `delete_invalid`** ([#256](https://github.com/domainaware/parsedmarc/issues/256)): control message deletion per report type instead of all-or-nothing. Each defaults to the value of the overall `delete` option, so existing configurations behave exactly as before; set one to override just that type — for example `delete = True` with `delete_failure = False` deletes processed aggregate and SMTP TLS report messages while archiving failure report messages. `delete_invalid` covers messages that could not be parsed, so they can be kept in the `Invalid` archive subfolder for debugging even when everything else is deleted. `get_dmarc_reports_from_mailbox()` and `watch_inbox()` gained matching `delete_aggregate`, `delete_failure`, `delete_smtp_tls`, and `delete_invalid` keyword arguments, where `None` (the default) means "inherit `delete`". The mutual exclusion between deletion and `test` mode is now evaluated against the effective per-type flags: any type that would be deleted still raises a `ValueError` alongside `test`, but `delete = True` with all four per-type options explicitly `False` is now valid with `test` enabled, since nothing would be deleted.
+- **New `[mailbox]` option `max_unsaved_retries`** (default `2`): how many times a batch of messages whose reports could not be saved is retried before its messages are moved to the `Unsaved` archive subfolder and stop being retried. `0` moves messages on the first failed save; negative values are rejected with a `ValueError`. Failures are counted in memory, per message and per process, so the cap applies across watch-mode checks within one long-running process rather than across separate one-shot runs. See the mailbox-persistence bug fix below for the full behavior; the cap is deliberately low because each retry re-delivers the batch to every output destination that does not deduplicate. `get_dmarc_reports_from_mailbox()` and `watch_inbox()` gained a matching `max_unsaved_retries` keyword argument.
 - **New `[general]` option `archive_directory`**: move successfully processed local report files into a dated archive tree (`<archive_directory>/<year>/<month>/<Aggregate|Failure|SMTP-TLS>/`); files that fail to parse as a report go to `<archive_directory>/Invalid/`, while files that fail for other reasons (e.g. transient I/O errors) are left in place so a later run can retry them; existing files are never overwritten (numeric suffix); files already under the archive directory are excluded from later runs. Applies only to report files passed directly as local path arguments ([#570](https://github.com/domainaware/parsedmarc/issues/570)).
 
+### Changes
+
+- **`get_dmarc_reports_from_mailbox()` gained an optional keyword-only `save_callback` parameter**, invoked once per fetched batch with a `ParsingResults` dict holding only that batch's newly parsed reports, after parsing but before any of the batch's messages are deleted or moved out of `reports_folder`. Returning `False` reports the batch as unsaved: its messages stay in place for retry and the aggregate-report dedup cache is not updated for that batch, so the retry reparses those reports rather than skipping them as duplicates. Raising counts as an unsaved batch too — the same retention and `max_unsaved_retries` bookkeeping runs — and the exception is then re-raised, so a callback that fails by raising (the CLI's does, under `fail_on_output_error`) is still bounded by the retry cap even when the caller's watch loop swallows the exception, as mailsuite's IMAP and Maildir backends do. Any other return value, including `None`, commits the batch exactly as before, which is what `save_callback=None` (the default) does for every batch. As a related, strictly safer side effect of staging the dedup keys per batch, a mid-batch crash now leaves `SEEN_AGGREGATE_REPORT_IDS` unmodified for that batch instead of partially populated.
+- **`watch_inbox()`'s `callback` is now passed through as `save_callback`**, so it runs once per batch *before* that batch is archived or deleted, with only that batch's reports, rather than once afterward with the whole check's accumulated results. It may return `False` to defer a batch to the next check.
+- **A single-shot CLI run configured with both file/mbox inputs and a mailbox connection now emits output in two passes instead of one**: mailbox-derived reports are saved through the new `save_callback` before `get_dmarc_reports_from_mailbox()` returns, and file/mbox-derived reports are saved in a second pass afterward. `append_json`/`append_csv` are already append-safe and watch mode already called its callback repeatedly, so this changes how often output happens in the single-shot case, not what is written. When a mailbox connection is configured and the file/mbox inputs yielded no reports (none given, or none parsed), the second pass is skipped entirely rather than printing an empty JSON document.
+
 ### Bug fixes
 
+- **Mailbox messages are no longer archived or deleted until the output destinations confirm the reports were saved** ([#242](https://github.com/domainaware/parsedmarc/issues/242)). Previously `get_dmarc_reports_from_mailbox()` archived or deleted every successfully parsed message *before* the CLI wrote anything to Elasticsearch/OpenSearch/Splunk/S3/Kafka/PostgreSQL/etc., so an output outage meant the report was never persisted anywhere while its source message was already gone from the reports folder, with no way to recover it. Each batch is now written to every configured destination first; if any destination fails, the whole batch's messages stay in the reports folder and are retried on the next run or watch-mode check. This is all-or-nothing per batch (archiving on partial success would still permanently lose the data for whichever destination failed) and applies independently of `fail_on_output_error`, which only controls the process exit code. Unparseable messages carry no report data and are still filed under `Invalid` (or deleted per `delete_invalid`) as before. To bound re-delivery to destinations that do not deduplicate, a message whose batch has failed `max_unsaved_retries + 1` times (three by default) is moved to `<archive_folder>/Unsaved` and stops being retried — never deleted, whatever the `delete` options say. Retries can produce duplicates in Kafka, Splunk HEC, syslog, GELF, webhooks, Azure Log Analytics, and the `--output` JSON/CSV files; Elasticsearch, OpenSearch, and PostgreSQL dedupe via `AlreadySaved`, and S3 overwrites the same key. Thanks to [@mkilijanek](https://github.com/mkilijanek) for the original implementation in [#823](https://github.com/domainaware/parsedmarc/pull/823).
+- **`extract_report()` now accepts plain uncompressed JSON.** The JSON magic-byte constant was written as `b"\7b"`, which Python reads as the octal escape `\7` (BEL, `0x07`) followed by a literal `b` — not `0x7B`, the `{` every RFC 8259 JSON object begins with — so the JSON branch never matched and plain JSON input was rejected as "Not a valid zip, gzip, json, or xml file". Every in-tree caller pre-guarded with its own zip/gzip or `{` check, which masked the dead branch; the practical impact was on `extract_report()` as a public API and on `application/tlsrpt+gzip` email attachments whose payload is actually uncompressed JSON. Now `b"\x7b"`.
+- **An SMTP TLS report with an empty `policies` list no longer crashes the CLI when `index_prefix_domain_map` is configured.** `parse_smtp_tls_report_json()` accepts an empty list, but `get_index_prefix()` unconditionally indexed `policies[0]`, raising `IndexError`. Such a report has no domain to map, so it is now treated as unmappable — excluded from prefix-mapped output like any other unmapped-domain report — instead of crashing.
+- **A failed `--output` file write no longer escapes uncaught.** `save_output()` was the only output destination the CLI did not wrap in a try/except, so a full disk or an unwritable output path crashed the run instead of being recorded like every other destination's failure. It is now caught and recorded as a `File output` error, which also means it correctly blocks mailbox archiving like any other failure.
+- **The emailed summary (`smtp_host` / Microsoft Graph) once again respects `index_prefix_domain_map` filtering of SMTP TLS reports.** `process_reports()` applies that filter in place to the dict it is given, and it is now given each mailbox batch and the file-derived snapshot rather than the combined results — so the combined dict passed to `email_results()` no longer inherited the filtering. It is filtered explicitly instead, so a configuration combining `index_prefix_domain_map` and an email destination excludes unmapped-domain SMTP TLS reports from the emailed summary, matching what was saved.
 - **Fixed the broken "XML files" link on the Splunk docs page**: it pointed at the old `splunk/` repository path instead of `dashboards/splunk/`, where the dashboards have lived since the `dashboards/` directory was introduced.
 - **A report file whose parsing raised an unexpected non-parser exception no longer hangs the CLI forever.** The old direct-file parallel implementation spawned a fresh child process per file and blocked the parent on a pipe read; a child that crashed with anything other than a `ParserError` never wrote to that pipe, so the parent waited indefinitely. The new implementation returns the error as a value and logs `Failed to parse <path>` instead.
 - **Direct-file parallel parsing no longer stalls a whole batch on one slow file, and no longer spawns a fresh interpreter per file.** The old implementation processed files in hard batches of `n_procs`, so a single slow file delayed every other file in its batch; the new pooled-worker implementation streams files through a reused pool instead.

docs/source/usage.md

@@ -186,6 +186,14 @@ The full set of configuration options are:
   - `fail_on_output_error` - bool: Exit with a non-zero status code if
       any configured output destination fails while saving/publishing
       reports (Default: `False`)
+
+    :::{note}
+    This option only controls the process exit code. Retaining mailbox
+    messages whose reports could not be saved is automatic and happens
+    either way — see
+    [Mailbox messages are only archived once the reports are saved](#mailbox-messages-are-only-archived-once-the-reports-are-saved).
+    :::
+
   - `log_file` - str: Write log messages to a file at this path
   - `n_procs` - int: Number of processes to run in parallel when
       parsing report files passed directly as CLI arguments, messages
@@ -242,9 +250,20 @@ The full set of configuration options are:
   - `test` - bool: Do not move or delete messages
   - `batch_size` - int: Number of messages to read and process
       before saving. Default `10`. Use `0` for no limit.
-  - `check_timeout` - int: Number of seconds to wait for a IMAP
+  - `check_timeout` - int: Number of seconds to wait for an IMAP
       IDLE response or the number of seconds until the next
       mail check (Default: `30`)
+  - `max_unsaved_retries` - int: How many times a batch of messages
+      whose reports could not be saved is retried before its messages
+      are moved to the `Unsaved` archive subfolder instead of being
+      retried again (Default: `2`, i.e. the initial attempt plus two
+      retries). Use `0` to move messages on the first failed save;
+      negative values are rejected.
+      Failures are counted in memory, so the cap applies across watch-mode
+      checks within one long-running process, not across separate one-shot
+      runs. See
+      [Mailbox messages are only archived once the reports are saved](#mailbox-messages-are-only-archived-once-the-reports-are-saved)
+      below.
   - `since` - str: Search for messages since certain time. (Examples: `5m|3h|2d|1w`)
       Acceptable units - {"m":"minutes", "h":"hours", "d":"days", "w":"weeks"}.
       Defaults to `1d` if incorrect value is provided.
@@ -850,6 +869,78 @@ PUT _cluster/settings
 Increasing this value increases resource usage.
 :::
 
+### Mailbox messages are only archived once the reports are saved
+
+parsedmarc processes a mailbox in batches of `batch_size` messages. Each
+batch is written to every configured output destination *before* any of
+that batch's messages are archived or deleted. If any destination reports a
+failure — an Elasticsearch outage, an expired Splunk HEC token, an
+unreachable Kafka broker, a full `--output` disk — the whole batch is left
+in the reports folder and retried on the next run or watch-mode check, so a
+report is never removed from the mailbox while it exists nowhere else
+([issue #242](https://github.com/domainaware/parsedmarc/issues/242)).
+
+This is all-or-nothing per batch: archiving a batch because most
+destinations accepted it would still permanently lose the data for the one
+that didn't. It also applies regardless of `fail_on_output_error`, which
+only controls the process exit code. Messages that could not be parsed at
+all carry no report data, so they are filed under `Invalid` (or deleted per
+`delete_invalid`) as usual.
+
+A destination that is broken rather than briefly unavailable would
+otherwise be retried forever, so retries are capped. Once a message's batch
+has failed `max_unsaved_retries + 1` times (three times by default), that
+message is moved to `<archive_folder>/Unsaved` and stops being retried. **A
+message is never deleted on this path, whatever the `delete` options say.**
+To recover after fixing the output destination, either move the messages
+from `Archive/Unsaved` back into the reports folder, or run parsedmarc once
+with `reports_folder = Archive/Unsaved`.
+
+:::{note}
+The failure counts live in memory, so they are counted per parsedmarc
+process. In watch mode — a long-running process that checks the mailbox
+repeatedly — the cap works as described across checks. A one-shot run
+(`cron`, `systemd` timers) attempts each message exactly once and then
+exits, so its counts start over every time and the default cap is never
+reached: messages simply keep being retried on every run, which is the
+safe direction. Set `max_unsaved_retries = 0` if you want one-shot runs to
+move unsavable messages to `Unsaved` immediately instead.
+:::
+
+:::{warning}
+Retrying a batch means re-sending it. Output destinations that
+deduplicate — Elasticsearch, OpenSearch, and PostgreSQL, which recognize
+an already-saved report — are unaffected, and S3 is idempotent because
+each report is written to an object key built from its type, date, and
+report ID, so a retry overwrites the same object. Kafka,
+Splunk HEC, syslog, GELF, webhooks, Azure Log Analytics, and the
+`--output` JSON/CSV files all append unconditionally, so each retry adds
+another copy of every report in the batch. That is why the default retry
+cap is deliberately low: at most three deliveries per report before its
+message is set aside in `Unsaved`. The summary email covers everything
+parsed in a run, including reports whose batch failed to save, so a report
+retried across runs can also appear in more than one summary email.
+:::
+
+:::{note}
+Not every destination can report a failed delivery. The webhook output
+deliberately logs and swallows its own HTTP and network errors, and the
+syslog and GELF outputs send through Python logging handlers, which
+swallow delivery errors by design — so an unreachable webhook, syslog, or
+GELF endpoint is *not* treated as a failed save and does not hold a
+batch's messages back. Failures in Elasticsearch, OpenSearch, Splunk HEC,
+Kafka, S3, PostgreSQL, Azure Log Analytics, and the `--output` files are
+all detected and do.
+:::
+
+:::{note}
+`since` interacts with retries: a message that ages out of the configured
+`since` window stops being fetched, and therefore stops being retried
+automatically. It is never deleted or moved — it simply stays in the
+reports folder until it is processed by a run with a wider (or no)
+`since` window.
+:::
+
 ## Environment variable configuration
 
 Any configuration option can be set via environment variables using the

parsedmarc/__init__.py

@@ -109,7 +109,31 @@ MAGIC_ZIP = b"\x50\x4b\x03\x04"
 MAGIC_GZIP = b"\x1f\x8b"
 MAGIC_XML = b"\x3c\x3f\x78\x6d\x6c\x20"
 MAGIC_XML_TAG = b"\x3c"  # '<' - XML starting with an element tag (no declaration)
-MAGIC_JSON = b"\7b"
+# 0x7B, "{" -- a JSON text that is an object begins with it (RFC 8259).
+# Previously written as b"\7b", which Python reads as the octal escape
+# \7 (BEL) followed by a literal "b", so the branch never matched real
+# JSON; every in-tree caller happened to pre-guard with its own zip/gzip
+# or "{" check, which masked it.
+MAGIC_JSON = b"\x7b"
+
+# Per-message count of consecutive failed saves, keyed on
+# ``(reports_folder, str(message_uid))``. Populated only when
+# ``get_dmarc_reports_from_mailbox()`` is given a ``save_callback`` that
+# reports a batch as unsaved; a message whose count exceeds
+# ``max_unsaved_retries`` is moved to ``{archive_folder}/Unsaved`` instead
+# of being retried forever, and its entry is dropped. A successful save
+# clears the entries for that batch's messages.
+#
+# Process-local and deliberately not persisted: a restart re-attempts every
+# message still in the reports folder, which is the safe direction (retry
+# rather than shelve). The key carries no connection identity, so a library
+# caller processing two connections that share a folder name (two IMAP
+# servers, both "INBOX") in one process could collide counters if UIDs
+# happen to match; the CLI uses one connection per process. It is not a
+# ``ParserConfig`` field for the same reason ``batch_size`` and the
+# ``delete`` flags are not -- it governs mailbox orchestration, not
+# parsing.
+_FAILED_SAVE_ATTEMPTS: dict[tuple[str, str], int] = {}
 
 EMAIL_SAMPLE_CONTENT_TYPES = (
     "text/rfc822",
@@ -2169,6 +2193,7 @@ def _classify_parsed_email(
     smtp_tls_reports: list[SMTPTLSReport],
     *,
     seen_aggregate_report_ids: ExpiringDict,
+    pending_aggregate_keys: set[str] | None = None,
 ) -> ReportType:
     """Classify a parsed report email, appending it to the matching list.
 
@@ -2181,6 +2206,15 @@ def _classify_parsed_email(
     ``seen_aggregate_report_ids`` cache so dedup state stays scoped to
     whichever ``ParserConfig`` (explicit or module-default) is in effect.
 
+    ``pending_aggregate_keys`` lets a caller *defer* the dedup-cache write:
+    when a set is supplied, a newly seen key is staged in that set instead
+    of being written to ``seen_aggregate_report_ids``, and the dedup check
+    consults both. The caller then folds the staged keys into the cache
+    once the batch is known to have been saved (see
+    ``get_dmarc_reports_from_mailbox``), or drops them so a retry reparses
+    the same reports rather than silently skipping them as duplicates.
+    ``None`` (the default) keeps the immediate-write behavior.
+
     Returns the report type so mailbox callers know which UID list to
     append the source message's UID to.
     """
@@ -2192,8 +2226,14 @@ def _classify_parsed_email(
         report_org = parsed_email["report"]["report_metadata"]["org_name"]
         report_id = parsed_email["report"]["report_metadata"]["report_id"]
         report_key = f"{report_org}_{report_id}"
-        if report_key not in seen_aggregate_report_ids:
-            seen_aggregate_report_ids[report_key] = True
+        already_seen = report_key in seen_aggregate_report_ids or (
+            pending_aggregate_keys is not None and report_key in pending_aggregate_keys
+        )
+        if not already_seen:
+            if pending_aggregate_keys is None:
+                seen_aggregate_report_ids[report_key] = True
+            else:
+                pending_aggregate_keys.add(report_key)
             aggregate_reports.append(parsed_email["report"])
         else:
             logger.debug(
@@ -2423,6 +2463,28 @@ def _migrate_forensic_archive_folder(
         )
 
 
+def _ensure_folder(connection: MailboxConnection, folder: str) -> None:
+    """Best-effort create ``folder`` if the backend says it is missing.
+
+    ``get_dmarc_reports_from_mailbox()`` creates its destination folders up
+    front, but only when ``create_folders`` is set -- watch mode calls it
+    with ``create_folders=False``, and the ``Unsaved`` holding folder is
+    only created up front when a ``save_callback`` was supplied. This is the
+    defensive check immediately before a message is moved there.
+
+    Like ``_migrate_forensic_archive_folder``, it never raises: a backend
+    that cannot report on or create the folder is logged and skipped, and
+    the move that follows either succeeds anyway (the folder already
+    existed) or fails and is logged by its own error handler -- warn, don't
+    crash.
+    """
+    try:
+        if not connection.folder_exists(folder):
+            connection.create_folder(folder)
+    except Exception as error:
+        logger.warning(f"Could not create folder {folder}: {error}")
+
+
 def get_dmarc_reports_from_mailbox(
     connection: MailboxConnection,
     *,
@@ -2449,6 +2511,8 @@ def get_dmarc_reports_from_mailbox(
     create_folders: bool = True,
     normalize_timespan_threshold_hours: float = 24.0,
     n_procs: int = 1,
+    save_callback: Callable[[ParsingResults], bool | None] | None = None,
+    max_unsaved_retries: int = 2,
     config: ParserConfig | None = None,
 ) -> ParsingResults:
     """
@@ -2502,6 +2566,50 @@ def get_dmarc_reports_from_mailbox(
             happens after the parsing phase completes, rather than
             interleaved message-by-message as it is when ``n_procs`` is 1.
             Not part of ``config``; always applies.
+        save_callback: An optional callable invoked once per fetched batch
+            with a ``ParsingResults`` dict holding only that batch's newly
+            parsed reports, after parsing but before any of the batch's
+            messages are deleted or moved out of ``reports_folder``. It
+            tells this function whether the batch was actually persisted:
+
+            * Returning ``False`` means "not saved": the batch's messages
+              are left in ``reports_folder`` for retry on the next run (or
+              moved to ``{archive_folder}/Unsaved`` once they have failed
+              ``max_unsaved_retries`` retries -- see below), and the
+              aggregate-report dedup cache is not updated for that batch, so
+              a retry reparses the same reports instead of skipping them as
+              duplicates.
+            * Raising counts as "not saved" too: the same bookkeeping runs
+              (the batch's messages are held back or moved to ``Unsaved`` at
+              the cap, and the failed attempt counts toward
+              ``max_unsaved_retries``), and the exception is then re-raised
+              to the caller.
+            * Any other return value, including ``None``, commits the batch:
+              the dedup cache is updated and the messages are deleted or
+              archived exactly as they are with no callback.
+
+            ``None`` (the default) commits every batch, preserving the prior
+            behavior. The callback is still invoked when ``test`` is
+            ``True``, so a test run exercises the full pipeline, but neither
+            the mailbox nor the retry counters are touched regardless of
+            what it returns.
+        max_unsaved_retries (int): How many times a message may be *retried*
+            after ``save_callback`` first reported its batch unsaved, before
+            it is moved to the ``Unsaved`` archive subfolder instead of
+            being retried again (default 2, i.e. the initial attempt plus
+            two retries -- at most three deliveries to any output
+            destination that does not deduplicate). ``0`` moves a message on
+            the first failed save; negative values raise ``ValueError``.
+            Messages moved to ``Unsaved`` are never deleted, whatever the
+            ``delete`` options say; recover them by fixing the output
+            destination and moving them back into ``reports_folder``.
+            Counts are kept in memory, per message and
+            per process, are reset by a successful save, and are only kept
+            when a ``save_callback`` is supplied -- so the cap applies
+            across repeated calls within one process (``watch_inbox()``'s
+            checks), not across separate one-shot processes, each of which
+            starts a message's count over. Mailbox orchestration, so like
+            ``batch_size`` it is not part of ``config``.
         config (ParserConfig): a single object carrying all parsing and
             enrichment options plus the caches; when provided, it replaces
             the individual parsing and enrichment option keyword arguments
@@ -2510,8 +2618,8 @@ def get_dmarc_reports_from_mailbox(
             ignored. The remaining keyword arguments control mailbox
             handling and orchestration rather than parsing (the folder
             names, the ``delete`` options, ``test``, ``since``,
-            ``batch_size``); they are not part of ``config`` and always
-            apply.
+            ``batch_size``, ``save_callback``, ``max_unsaved_retries``);
+            they are not part of ``config`` and always apply.
 
     Returns:
         dict: Lists of ``aggregate_reports``, ``failure_reports``, and ``smtp_tls_reports``
@@ -2532,6 +2640,9 @@ def get_dmarc_reports_from_mailbox(
     ):
         raise ValueError("delete options and test are mutually exclusive")
 
+    if max_unsaved_retries < 0:
+        raise ValueError("max_unsaved_retries must be >= 0")
+
     if connection is None:
         raise ValueError("Must supply a connection")
 
@@ -2555,6 +2666,17 @@ def get_dmarc_reports_from_mailbox(
     aggregate_reports: list[AggregateReport] = []
     failure_reports: list[FailureReport] = []
     smtp_tls_reports: list[SMTPTLSReport] = []
+    # This call's own reports, kept separate from the accumulated lists
+    # above (which carry earlier batches' reports in via ``results``) so the
+    # save callback is handed only what this batch parsed.
+    batch_aggregate_reports: list[AggregateReport] = []
+    batch_failure_reports: list[FailureReport] = []
+    batch_smtp_tls_reports: list[SMTPTLSReport] = []
+    # Aggregate dedup keys are staged here and only written to the shared
+    # cache once the batch is known to be saved, so an unsaved batch (or a
+    # mid-batch crash) leaves the cache clean and the reports are reparsed
+    # on the retry instead of being dropped as duplicates.
+    pending_aggregate_keys: set[str] = set()
     aggregate_report_msg_uids = []
     failure_report_msg_uids = []
     smtp_tls_msg_uids = []
@@ -2562,6 +2684,7 @@ def get_dmarc_reports_from_mailbox(
     failure_reports_folder = f"{archive_folder}/Failure"
     smtp_tls_reports_folder = f"{archive_folder}/SMTP-TLS"
     invalid_reports_folder = f"{archive_folder}/Invalid"
+    unsaved_reports_folder = f"{archive_folder}/Unsaved"
 
     if results:
         aggregate_reports = results["aggregate_reports"].copy()
@@ -2575,6 +2698,11 @@ def get_dmarc_reports_from_mailbox(
         connection.create_folder(failure_reports_folder)
         connection.create_folder(smtp_tls_reports_folder)
         connection.create_folder(invalid_reports_folder)
+        if save_callback is not None:
+            # Only reachable when a callback can report a batch unsaved;
+            # without one no message is ever held back, so the folder would
+            # sit empty in every mailbox.
+            connection.create_folder(unsaved_reports_folder)
 
     if since and isinstance(since, str):
         _since = 1440  # default one day
@@ -2672,10 +2800,11 @@ def get_dmarc_reports_from_mailbox(
             else:
                 report_type = _classify_parsed_email(
                     result,
-                    aggregate_reports,
-                    failure_reports,
-                    smtp_tls_reports,
+                    batch_aggregate_reports,
+                    batch_failure_reports,
+                    batch_smtp_tls_reports,
                     seen_aggregate_report_ids=cfg.seen_aggregate_report_ids,
+                    pending_aggregate_keys=pending_aggregate_keys,
                 )
                 if report_type == "aggregate":
                     aggregate_report_msg_uids.append(message_id)
@@ -2707,10 +2836,11 @@ def get_dmarc_reports_from_mailbox(
                 )
                 report_type = _classify_parsed_email(
                     parsed_email,
-                    aggregate_reports,
-                    failure_reports,
-                    smtp_tls_reports,
+                    batch_aggregate_reports,
+                    batch_failure_reports,
+                    batch_smtp_tls_reports,
                     seen_aggregate_report_ids=cfg.seen_aggregate_report_ids,
+                    pending_aggregate_keys=pending_aggregate_keys,
                 )
                 if report_type == "aggregate":
                     aggregate_report_msg_uids.append(message_id)
@@ -2725,7 +2855,102 @@ def get_dmarc_reports_from_mailbox(
                         connection, message_id, delete_invalid, invalid_reports_folder
                     )
 
-    if not test:
+    # Ask the caller whether this batch actually made it to its output
+    # destinations before touching a single message. A callback that says
+    # otherwise (or raises) means the reports exist nowhere else yet, so no
+    # message may be archived or deleted -- only retained for retry, or moved
+    # intact to the Unsaved folder once retries run out (#242).
+    batch_results: ParsingResults = {
+        "aggregate_reports": batch_aggregate_reports,
+        "failure_reports": batch_failure_reports,
+        "smtp_tls_reports": batch_smtp_tls_reports,
+    }
+    persisted = True
+    callback_error: Exception | None = None
+    if save_callback is not None:
+        try:
+            persisted = save_callback(batch_results) is not False
+        except Exception as error:
+            # A raising callback is a failed save too: the failure
+            # bookkeeping below still runs (count the attempt, hold or shelve
+            # the messages) before the exception is re-raised, so a callback
+            # that always raises -- e.g. the CLI's under
+            # ``fail_on_output_error`` -- is still bounded by
+            # ``max_unsaved_retries`` even when the caller's watch loop
+            # swallows the exception and keeps checking, as mailsuite's IMAP
+            # and Maildir backends do.
+            persisted = False
+            callback_error = error
+
+    batch_msg_uids = (
+        aggregate_report_msg_uids + failure_report_msg_uids + smtp_tls_msg_uids
+    )
+
+    if persisted:
+        # Committing: the reports are safely stored elsewhere, so record
+        # their dedup keys and forget any earlier failures for these
+        # messages.
+        for report_key in pending_aggregate_keys:
+            cfg.seen_aggregate_report_ids[report_key] = True
+        if not test:
+            for msg_uid in batch_msg_uids:
+                _FAILED_SAVE_ATTEMPTS.pop((reports_folder, str(msg_uid)), None)
+    elif not test:
+        # Held back for retry. Messages that have now failed the initial
+        # attempt plus ``max_unsaved_retries`` retries stop being retried and
+        # move to the Unsaved folder, bounding duplicate delivery to output
+        # destinations that do not deduplicate; the rest stay put.
+        retained_uids: list[int | str] = []
+        over_cap_uids: list[int | str] = []
+        highest_attempt = 0
+        for msg_uid in batch_msg_uids:
+            attempts_key = (reports_folder, str(msg_uid))
+            attempts = _FAILED_SAVE_ATTEMPTS.get(attempts_key, 0) + 1
+            _FAILED_SAVE_ATTEMPTS[attempts_key] = attempts
+            if attempts > max_unsaved_retries:
+                over_cap_uids.append(msg_uid)
+            else:
+                retained_uids.append(msg_uid)
+                highest_attempt = max(highest_attempt, attempts)
+        if retained_uids:
+            logger.error(
+                f"Reports were not saved: leaving {len(retained_uids)} "
+                f"message(s) in {reports_folder} to retry on the next run "
+                f"or check (failed attempt {highest_attempt} of "
+                f"{max_unsaved_retries + 1})"
+            )
+        if over_cap_uids:
+            logger.error(
+                f"Reports were not saved after {max_unsaved_retries + 1} "
+                f"attempt(s): moving {len(over_cap_uids)} message(s) from "
+                f"{reports_folder} to {unsaved_reports_folder} instead of "
+                "retrying them further. They are never deleted -- fix the "
+                f"output destination, then move them back to {reports_folder}"
+            )
+            _ensure_folder(connection, unsaved_reports_folder)
+            for msg_uid in over_cap_uids:
+                try:
+                    connection.move_message(msg_uid, unsaved_reports_folder)
+                except Exception as e:
+                    e = f"Error moving message UID {msg_uid}: {e}"
+                    logger.error(f"Mailbox error: {e}")
+                else:
+                    # Drop the counter only once the message is actually out
+                    # of the retry loop. Clearing it before a failed move
+                    # would hand the still-in-place message a fresh set of
+                    # under-cap retries (and deliveries); keeping it means
+                    # the next failed save classifies the message over-cap
+                    # again and re-attempts the move instead.
+                    _FAILED_SAVE_ATTEMPTS.pop((reports_folder, str(msg_uid)), None)
+
+    if callback_error is not None:
+        raise callback_error
+
+    aggregate_reports += batch_aggregate_reports
+    failure_reports += batch_failure_reports
+    smtp_tls_reports += batch_smtp_tls_reports
+
+    if persisted and not test:
         # Each report type is disposed of according to its own effective
         # delete flag: deleted outright, or moved to its archive subfolder.
         for msg_uids, delete_type, destination_folder, label in (
@@ -2782,7 +3007,11 @@ def get_dmarc_reports_from_mailbox(
         "smtp_tls_reports": smtp_tls_reports,
     }
 
-    if not test and not batch_size:
+    # An unsaved batch left its messages in ``reports_folder``, so the
+    # re-check below would find them again and immediately reprocess the
+    # very messages that just failed -- burning through the retry cap in one
+    # call. Skip it and let the next run retry them.
+    if persisted and not test and not batch_size:
         if current_time:
             total_messages = len(
                 connection.fetch_messages(reports_folder, since=current_time)
@@ -2807,6 +3036,8 @@ def get_dmarc_reports_from_mailbox(
             results=results,
             since=current_time,
             n_procs=n_procs,
+            save_callback=save_callback,
+            max_unsaved_retries=max_unsaved_retries,
             config=cfg,
         )
 
@@ -2840,6 +3071,7 @@ def watch_inbox(
     normalize_timespan_threshold_hours: float = 24.0,
     config_reloading: Callable | None = None,
     n_procs: int = 1,
+    max_unsaved_retries: int = 2,
     config: ParserConfig | None = None,
 ):
     """
@@ -2848,7 +3080,21 @@ def watch_inbox(
 
     Args:
         mailbox_connection: The mailbox connection object
-        callback: The callback function to receive the parsing results
+        callback: The callback function to receive the parsing results.
+            Passed straight through to ``get_dmarc_reports_from_mailbox()``
+            as its ``save_callback``, so it now runs once per fetched batch,
+            with only that batch's reports, *before* those messages are
+            deleted or moved out of ``reports_folder`` -- rather than once
+            afterward with the whole check's accumulated results. Returning
+            ``False`` reports the batch as unsaved, leaving its messages in
+            place to be retried on the next check instead of archived or
+            deleted (see ``save_callback`` and ``max_unsaved_retries`` on
+            ``get_dmarc_reports_from_mailbox()``). Raising counts as an
+            unsaved batch too -- same retention and retry cap -- before the
+            exception reaches the mailbox backend's watch loop; what happens
+            then is backend-specific: the Microsoft Graph and Gmail backends
+            let it propagate and end the watch, while mailsuite's IMAP and
+            Maildir watch loops log it and keep checking.
         reports_folder (str): The IMAP folder where reports can be found
         archive_folder (str): The folder to move processed mail to
         delete (bool): Delete messages after processing them
@@ -2893,6 +3139,11 @@ def watch_inbox(
         n_procs (int): Number of processes to use for parsing messages in
             parallel. Passed through to ``get_dmarc_reports_from_mailbox``
             on each check. Not part of ``config``; always applies.
+        max_unsaved_retries (int): How many times a message may be retried
+            after ``callback`` first reported its batch unsaved, before it is
+            moved to the ``Unsaved`` archive subfolder (default 2). Passed
+            through to ``get_dmarc_reports_from_mailbox``, where it is
+            documented in full. Not part of ``config``; always applies.
         config (ParserConfig): a single object carrying all parsing and
             enrichment options plus the caches; when provided, it replaces
             the individual parsing and enrichment option keyword arguments
@@ -2901,9 +3152,15 @@ def watch_inbox(
             ignored. The remaining keyword arguments control mailbox
             handling and orchestration rather than parsing (the folder
             names, the ``delete`` options, ``test``, ``since``,
-            ``batch_size``); they are not part of ``config`` and always
-            apply.
+            ``batch_size``, ``max_unsaved_retries``); they are not part of
+            ``config`` and always apply.
     """
+    # Validate before the watch loop starts: raised inside a check, this
+    # would be swallowed and endlessly retried by the IMAP and Maildir
+    # backends' per-check exception handling instead of surfacing.
+    if max_unsaved_retries < 0:
+        raise ValueError("max_unsaved_retries must be >= 0")
+
     cfg = _resolve_config(
         config,
         offline=offline,
@@ -2919,7 +3176,7 @@ def watch_inbox(
     )
 
     def check_callback(connection):
-        res = get_dmarc_reports_from_mailbox(
+        get_dmarc_reports_from_mailbox(
             connection=connection,
             reports_folder=reports_folder,
             archive_folder=archive_folder,
@@ -2933,9 +3190,10 @@ def watch_inbox(
             since=since,
             create_folders=False,
             n_procs=n_procs,
+            save_callback=callback,
+            max_unsaved_retries=max_unsaved_retries,
             config=cfg,
         )
-        callback(res)
 
     watch_kwargs: dict = {
         "check_callback": check_callback,

parsedmarc/cli.py

@@ -805,6 +805,10 @@ def _parse_config(config: ConfigParser, opts):
             opts.mailbox_batch_size = mailbox_config.getint("batch_size")
         if "check_timeout" in mailbox_config:
             opts.mailbox_check_timeout = mailbox_config.getint("check_timeout")
+        if "max_unsaved_retries" in mailbox_config:
+            opts.mailbox_max_unsaved_retries = mailbox_config.getint(
+                "max_unsaved_retries"
+            )
         if "since" in mailbox_config:
             opts.mailbox_since = mailbox_config["since"]
 
@@ -1713,7 +1717,12 @@ def _main():
             domain = report["policy_published"]["domain"]
         elif "reported_domain" in report:
             domain = report["reported_domain"]
-        elif "policies" in report:
+        elif report.get("policies"):
+            # Guarded with .get() truthiness: parse_smtp_tls_report_json()
+            # accepts a report whose policies list is empty, which would
+            # make [0] raise IndexError here. Such a report has no domain
+            # to map, so it falls through to return None like any other
+            # unmappable report.
             domain = report["policies"][0]["policy_domain"]
         if domain:
             domain = get_base_domain(domain)
@@ -1732,7 +1741,44 @@ def _main():
                         return prefix
         return None
 
+    def filter_smtp_tls_reports_for_index_prefix(tls_reports):
+        """Drop SMTP TLS reports whose domain isn't covered by
+        ``index_prefix_domain_map``.
+
+        Shared by ``process_reports()`` (which filters each batch it saves)
+        and by the combined ``parsing_results`` that feeds
+        ``email_results()``. Mailbox batches are saved inside
+        ``get_dmarc_reports_from_mailbox()``, so the dicts
+        ``process_reports()`` filters in place are no longer the same
+        objects as the combined results assembled afterward -- without this,
+        the emailed summary would list SMTP TLS reports that were
+        deliberately excluded from every output destination.
+        """
+        if index_prefix_domain_map is None:
+            return tls_reports
+        filtered_tls = []
+        for report in tls_reports:
+            if get_index_prefix(report) is not None:
+                filtered_tls.append(report)
+            else:
+                domain = "unknown"
+                if "policies" in report and report["policies"]:
+                    domain = report["policies"][0].get("policy_domain", "unknown")
+                logger.debug(
+                    "Ignoring SMTP TLS report for domain not in "
+                    "index_prefix_domain_map: %s",
+                    domain,
+                )
+        return filtered_tls
+
     def process_reports(reports_):
+        """Write ``reports_`` to every configured output destination.
+
+        Returns the list of human-readable output-error messages recorded
+        along the way -- empty when every destination accepted the reports.
+        Callers use that as the "was this batch saved?" signal; see
+        ``mailbox_save_callback()``.
+        """
         output_errors = []
 
         def log_output_error(destination, error):
@@ -1741,20 +1787,9 @@ def _main():
             output_errors.append(message)
 
         if index_prefix_domain_map is not None:
-            filtered_tls = []
-            for report in reports_.get("smtp_tls_reports", []):
-                if get_index_prefix(report) is not None:
-                    filtered_tls.append(report)
-                else:
-                    domain = "unknown"
-                    if "policies" in report and report["policies"]:
-                        domain = report["policies"][0].get("policy_domain", "unknown")
-                    logger.debug(
-                        "Ignoring SMTP TLS report for domain not in "
-                        "index_prefix_domain_map: %s",
-                        domain,
-                    )
-            reports_["smtp_tls_reports"] = filtered_tls
+            reports_["smtp_tls_reports"] = filter_smtp_tls_reports_for_index_prefix(
+                reports_.get("smtp_tls_reports", [])
+            )
 
         indent_value = 2 if opts.prettify_json else None
         output_str = (
@@ -1764,16 +1799,24 @@ def _main():
         if not opts.silent:
             print(output_str)
         if opts.output:
-            save_output(
-                reports_,
-                output_directory=opts.output,
-                aggregate_json_filename=opts.aggregate_json_filename,
-                failure_json_filename=opts.failure_json_filename,
-                smtp_tls_json_filename=opts.smtp_tls_json_filename,
-                aggregate_csv_filename=opts.aggregate_csv_filename,
-                failure_csv_filename=opts.failure_csv_filename,
-                smtp_tls_csv_filename=opts.smtp_tls_csv_filename,
-            )
+            try:
+                save_output(
+                    reports_,
+                    output_directory=opts.output,
+                    aggregate_json_filename=opts.aggregate_json_filename,
+                    failure_json_filename=opts.failure_json_filename,
+                    smtp_tls_json_filename=opts.smtp_tls_json_filename,
+                    aggregate_csv_filename=opts.aggregate_csv_filename,
+                    failure_csv_filename=opts.failure_csv_filename,
+                    smtp_tls_csv_filename=opts.smtp_tls_csv_filename,
+                )
+            except (OSError, ValueError) as error_:
+                # The only output destination that was not already caught:
+                # a full disk or an unwritable directory used to crash the
+                # run outright, and now that a failed save holds mailbox
+                # messages back it also has to be recorded like any other
+                # destination's failure rather than escaping.
+                log_output_error("File output", str(error_))
 
         kafka_client = clients.get("kafka_client")
         s3_client = clients.get("s3_client")
@@ -2096,6 +2139,8 @@ def _main():
                 )
             )
 
+        return output_errors
+
     arg_parser = ArgumentParser(description="Parses DMARC reports")
     arg_parser.add_argument(
         "-c",
@@ -2243,6 +2288,7 @@ def _main():
         mailbox_test=False,
         mailbox_batch_size=10,
         mailbox_check_timeout=30,
+        mailbox_max_unsaved_retries=2,
         mailbox_since=None,
         imap_host=None,
         imap_skip_certificate_verification=False,
@@ -2596,10 +2642,24 @@ def _main():
         failure_reports += reports["failure_reports"]
         smtp_tls_reports += reports["smtp_tls_reports"]
 
+    # Snapshot of the file/mbox-derived reports, taken before the mailbox
+    # block below appends anything fetched from a live mailbox connection.
+    # Mailbox batches are handed to process_reports() by
+    # mailbox_save_callback() before get_dmarc_reports_from_mailbox() even
+    # returns -- that is what lets it decide whether archiving is safe -- so
+    # the final process_reports() call runs on this snapshot only, or the
+    # mailbox-derived reports would be saved twice.
+    file_parsing_results: ParsingResults = {
+        "aggregate_reports": list(aggregate_reports),
+        "failure_reports": list(failure_reports),
+        "smtp_tls_reports": list(smtp_tls_reports),
+    }
+
     mailbox_connection = None
     msgraph_connection: MSGraphConnection | None = None
     mailbox_batch_size_value = 10
     mailbox_check_timeout_value = 30
+    mailbox_max_unsaved_retries_value = 2
 
     if opts.imap_host:
         try:
@@ -2771,6 +2831,27 @@ def _main():
             if opts.mailbox_check_timeout is not None
             else 30
         )
+        mailbox_max_unsaved_retries_value = (
+            int(opts.mailbox_max_unsaved_retries)
+            if opts.mailbox_max_unsaved_retries is not None
+            else 2
+        )
+
+    def mailbox_save_callback(batch: ParsingResults) -> bool:
+        """Save one mailbox batch and report whether it actually landed.
+
+        Passed to ``get_dmarc_reports_from_mailbox()`` as ``save_callback``
+        and to ``watch_inbox()`` as its ``callback``. Returning ``False``
+        when any output destination failed keeps that batch's messages in
+        the mailbox to be retried, instead of archiving or deleting reports
+        that were never persisted anywhere (#242). With
+        ``fail_on_output_error`` enabled it never returns ``False``:
+        ``process_reports()`` raises ``ParserError`` instead, which the
+        library treats as "unsaved" too (same retention and retry-cap
+        bookkeeping) before the exception propagates back out here.
+        """
+        return not process_reports(batch)
+
     if mailbox_connection and not _shutdown_requested:
         try:
             reports = get_dmarc_reports_from_mailbox(
@@ -2787,12 +2868,21 @@ def _main():
                 since=opts.mailbox_since,
                 config=parser_config,
                 n_procs=n_procs,
+                save_callback=mailbox_save_callback,
+                max_unsaved_retries=mailbox_max_unsaved_retries_value,
             )
 
             aggregate_reports += reports["aggregate_reports"]
             failure_reports += reports["failure_reports"]
             smtp_tls_reports += reports["smtp_tls_reports"]
 
+        except ParserError as error:
+            # fail_on_output_error turns a failed batch save into a
+            # ParserError inside mailbox_save_callback; it reaches here
+            # through get_dmarc_reports_from_mailbox, which leaves the
+            # batch's messages in the mailbox on its way out.
+            logger.error(error.__str__())
+            sys.exit(1)
         except (ClientAuthenticationError, APIError, httpx.HTTPError) as error:
             if msgraph_connection is None:
                 logger.exception("Mailbox Error")
@@ -2809,17 +2899,31 @@ def _main():
             logger.exception("Mailbox Error")
             exit(1)
 
+    # Filtered here rather than relying on process_reports()'s in-place
+    # filtering: the dicts it filters are the file snapshot and the mailbox
+    # batches, not this combined dict, which exists only to feed
+    # email_results() / email_results_via_msgraph() below.
     parsing_results: ParsingResults = {
         "aggregate_reports": aggregate_reports,
         "failure_reports": failure_reports,
-        "smtp_tls_reports": smtp_tls_reports,
+        "smtp_tls_reports": filter_smtp_tls_reports_for_index_prefix(smtp_tls_reports),
     }
 
-    try:
-        process_reports(parsing_results)
-    except ParserError as error:
-        logger.error(error.__str__())
-        exit(1)
+    file_results_nonempty = bool(
+        file_parsing_results["aggregate_reports"]
+        or file_parsing_results["failure_reports"]
+        or file_parsing_results["smtp_tls_reports"]
+    )
+    # Mailbox-derived reports were already saved by mailbox_save_callback;
+    # only file/mbox-derived reports are left to save here. With a mailbox
+    # connection and nothing from files, skip the call entirely so the run
+    # doesn't print a second, empty JSON blob.
+    if file_results_nonempty or not mailbox_connection:
+        try:
+            process_reports(file_parsing_results)
+        except ParserError as error:
+            logger.error(error.__str__())
+            sys.exit(1)
 
     smtp_to_value = (
         list(opts.smtp_to)
@@ -2898,7 +3002,7 @@ def _main():
                 # at a safe boundary once the current batch is processed.
                 watch_inbox(
                     mailbox_connection=mailbox_connection,
-                    callback=process_reports,
+                    callback=mailbox_save_callback,
                     reports_folder=opts.mailbox_reports_folder,
                     archive_folder=opts.mailbox_archive_folder,
                     delete=opts.mailbox_delete,
@@ -2913,6 +3017,7 @@ def _main():
                     config=parser_config,
                     config_reloading=lambda: _reload_requested or _shutdown_requested,
                     n_procs=n_procs,
+                    max_unsaved_retries=mailbox_max_unsaved_retries_value,
                 )
             except FileExistsError as error:
                 logger.error(f"{error.__str__()}")
@@ -3012,6 +3117,11 @@ def _main():
                     if opts.mailbox_check_timeout is not None
                     else 30
                 )
+                mailbox_max_unsaved_retries_value = (
+                    int(opts.mailbox_max_unsaved_retries)
+                    if opts.mailbox_max_unsaved_retries is not None
+                    else 2
+                )
 
                 # Update log level
                 logger.setLevel(logging.ERROR)

tests/test_cli.py

@@ -44,6 +44,29 @@ def _sample_aggregate_reports() -> list[AggregateReport]:
     return [cast(AggregateReport, result["report"])]
 
 
+def _fetch_invoking_save_callback(reports, *, batch=None):
+    """Build a ``get_dmarc_reports_from_mailbox`` mock side effect that
+    honors the real contract: every fetched batch is handed to
+    ``save_callback`` *before* the function returns, and the accumulated
+    results are returned afterward.
+
+    A plain ``return_value`` mock would leave ``save_callback`` uncalled, so
+    nothing would ever be written to any output destination -- which is not
+    how the CLI behaves against a real mailbox since #242.
+
+    ``batch`` overrides what the callback receives, for tests that need the
+    batch dict to be a distinct object from the returned results (the real
+    function builds its return value from its own accumulated lists, not
+    from the dict it passed to the callback).
+    """
+
+    def _fetch_and_save(**kwargs):
+        kwargs["save_callback"](reports if batch is None else batch)
+        return reports
+
+    return _fetch_and_save
+
+
 class _BreakLoop(BaseException):
     pass
 
@@ -191,11 +214,13 @@ aws_service = aoss
     ):
         """CLI should exit with code 1 when fail_on_output_error is enabled"""
         mock_imap_connection.return_value = object()
-        mock_get_reports.return_value = {
-            "aggregate_reports": [{"policy_published": {"domain": "example.com"}}],
-            "failure_reports": [],
-            "smtp_tls_reports": [],
-        }
+        mock_get_reports.side_effect = _fetch_invoking_save_callback(
+            {
+                "aggregate_reports": [{"policy_published": {"domain": "example.com"}}],
+                "failure_reports": [],
+                "smtp_tls_reports": [],
+            }
+        )
         mock_save_aggregate.side_effect = parsedmarc.elastic.ElasticsearchError(
             "simulated output failure"
         )
@@ -241,11 +266,13 @@ hosts = localhost
         mock_save_aggregate,
     ):
         mock_imap_connection.return_value = object()
-        mock_get_reports.return_value = {
-            "aggregate_reports": [{"policy_published": {"domain": "example.com"}}],
-            "failure_reports": [],
-            "smtp_tls_reports": [],
-        }
+        mock_get_reports.side_effect = _fetch_invoking_save_callback(
+            {
+                "aggregate_reports": [{"policy_published": {"domain": "example.com"}}],
+                "failure_reports": [],
+                "smtp_tls_reports": [],
+            }
+        )
         mock_save_aggregate.side_effect = parsedmarc.elastic.ElasticsearchError(
             "simulated output failure"
         )
@@ -297,11 +324,13 @@ hosts = localhost
         mock_save_failure_opensearch,
     ):
         mock_imap_connection.return_value = object()
-        mock_get_reports.return_value = {
-            "aggregate_reports": [{"policy_published": {"domain": "example.com"}}],
-            "failure_reports": [{"reported_domain": "example.com"}],
-            "smtp_tls_reports": [],
-        }
+        mock_get_reports.side_effect = _fetch_invoking_save_callback(
+            {
+                "aggregate_reports": [{"policy_published": {"domain": "example.com"}}],
+                "failure_reports": [{"reported_domain": "example.com"}],
+                "smtp_tls_reports": [],
+            }
+        )
         mock_save_aggregate.side_effect = parsedmarc.elastic.ElasticsearchError(
             "aggregate sink failed"
         )
@@ -3959,6 +3988,70 @@ watch = true
         es_client.close.assert_called_once()
 
 
+def _domain_map_tls_reports():
+    """Four SMTP TLS reports for the index_prefix_domain_map tests: two
+    whose policy domains fold to the mapped base domain example.com (one of
+    them mixed-case), one for an unmapped domain, and one with an empty
+    policies list -- parse_smtp_tls_report_json() accepts those, and
+    get_index_prefix() must treat them as unmappable (dropped by the
+    filter) rather than crash on policies[0]."""
+    return [
+        {
+            "organization_name": "Allowed Org",
+            "begin_date": "2024-01-01T00:00:00Z",
+            "end_date": "2024-01-01T23:59:59Z",
+            "report_id": "allowed-1",
+            "contact_info": "tls@allowed.example.com",
+            "policies": [
+                {
+                    "policy_domain": "allowed.example.com",
+                    "policy_type": "sts",
+                    "successful_session_count": 1,
+                    "failed_session_count": 0,
+                }
+            ],
+        },
+        {
+            "organization_name": "Unmapped Org",
+            "begin_date": "2024-01-01T00:00:00Z",
+            "end_date": "2024-01-01T23:59:59Z",
+            "report_id": "unmapped-1",
+            "contact_info": "tls@unmapped.example.net",
+            "policies": [
+                {
+                    "policy_domain": "unmapped.example.net",
+                    "policy_type": "sts",
+                    "successful_session_count": 5,
+                    "failed_session_count": 0,
+                }
+            ],
+        },
+        {
+            "organization_name": "Mixed Case Org",
+            "begin_date": "2024-01-01T00:00:00Z",
+            "end_date": "2024-01-01T23:59:59Z",
+            "report_id": "mixed-case-1",
+            "contact_info": "tls@mixedcase.example.com",
+            "policies": [
+                {
+                    "policy_domain": "MixedCase.Example.Com",
+                    "policy_type": "sts",
+                    "successful_session_count": 2,
+                    "failed_session_count": 0,
+                }
+            ],
+        },
+        {
+            "organization_name": "No Policies Org",
+            "begin_date": "2024-01-01T00:00:00Z",
+            "end_date": "2024-01-01T23:59:59Z",
+            "report_id": "no-policies-1",
+            "contact_info": "tls@nopolicies.example.org",
+            "policies": [],
+        },
+    ]
+
+
 class TestIndexPrefixDomainMapTlsFiltering(unittest.TestCase):
     """Tests that SMTP TLS reports for unmapped domains are filtered out
     when index_prefix_domain_map is configured."""
@@ -3972,57 +4065,13 @@ class TestIndexPrefixDomainMapTlsFiltering(unittest.TestCase):
     ):
         """TLS reports for domains not in the map should be silently dropped."""
         mock_imap_connection.return_value = object()
-        mock_get_reports.return_value = {
-            "aggregate_reports": [],
-            "failure_reports": [],
-            "smtp_tls_reports": [
-                {
-                    "organization_name": "Allowed Org",
-                    "begin_date": "2024-01-01T00:00:00Z",
-                    "end_date": "2024-01-01T23:59:59Z",
-                    "report_id": "allowed-1",
-                    "contact_info": "tls@allowed.example.com",
-                    "policies": [
-                        {
-                            "policy_domain": "allowed.example.com",
-                            "policy_type": "sts",
-                            "successful_session_count": 1,
-                            "failed_session_count": 0,
-                        }
-                    ],
-                },
-                {
-                    "organization_name": "Unmapped Org",
-                    "begin_date": "2024-01-01T00:00:00Z",
-                    "end_date": "2024-01-01T23:59:59Z",
-                    "report_id": "unmapped-1",
-                    "contact_info": "tls@unmapped.example.net",
-                    "policies": [
-                        {
-                            "policy_domain": "unmapped.example.net",
-                            "policy_type": "sts",
-                            "successful_session_count": 5,
-                            "failed_session_count": 0,
-                        }
-                    ],
-                },
-                {
-                    "organization_name": "Mixed Case Org",
-                    "begin_date": "2024-01-01T00:00:00Z",
-                    "end_date": "2024-01-01T23:59:59Z",
-                    "report_id": "mixed-case-1",
-                    "contact_info": "tls@mixedcase.example.com",
-                    "policies": [
-                        {
-                            "policy_domain": "MixedCase.Example.Com",
-                            "policy_type": "sts",
-                            "successful_session_count": 2,
-                            "failed_session_count": 0,
-                        }
-                    ],
-                },
-            ],
-        }
+        mock_get_reports.side_effect = _fetch_invoking_save_callback(
+            {
+                "aggregate_reports": [],
+                "failure_reports": [],
+                "smtp_tls_reports": _domain_map_tls_reports(),
+            }
+        )
 
         domain_map = {"tenant_a": ["example.com"]}
         with NamedTemporaryFile("w", suffix=".yaml", delete=False) as map_file:
@@ -4052,6 +4101,9 @@ password = test-password
             with patch("sys.stdout", captured):
                 parsedmarc.cli._main()
 
+        # A single JSON document, not two: a run whose only reports came
+        # from the mailbox saves them inside save_callback and skips the
+        # second, empty pass rather than printing another blob after it.
         output = json.loads(captured.getvalue())
         tls_reports = output["smtp_tls_reports"]
         self.assertEqual(len(tls_reports), 2)
@@ -4059,6 +4111,341 @@ password = test-password
         self.assertIn("allowed-1", report_ids)
         self.assertIn("mixed-case-1", report_ids)
         self.assertNotIn("unmapped-1", report_ids)
+        # Unmappable (no policies -> no domain), dropped rather than an
+        # IndexError from get_index_prefix() on policies[0].
+        self.assertNotIn("no-policies-1", report_ids)
+
+    @patch("parsedmarc.cli.email_results")
+    @patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
+    @patch("parsedmarc.cli.IMAPConnection")
+    def testTlsReportsFilteredForEmailResults(
+        self,
+        mock_imap_connection,
+        mock_get_reports,
+        mock_email_results,
+    ):
+        """The combined results handed to email_results() are filtered by
+        index_prefix_domain_map exactly like the batches that were saved.
+
+        Regression test: get_dmarc_reports_from_mailbox() builds its return
+        value from its own accumulated lists, not from the batch dict it
+        passes to save_callback, so the in-place filtering process_reports()
+        applies to each batch never reaches the combined dict. The mock
+        mirrors that by handing save_callback a copy while returning the
+        full, unfiltered set."""
+        mock_imap_connection.return_value = object()
+        reports_dict = {
+            "aggregate_reports": [],
+            "failure_reports": [],
+            "smtp_tls_reports": _domain_map_tls_reports(),
+        }
+        mock_get_reports.side_effect = _fetch_invoking_save_callback(
+            reports_dict,
+            batch={
+                "aggregate_reports": [],
+                "failure_reports": [],
+                "smtp_tls_reports": list(reports_dict["smtp_tls_reports"]),
+            },
+        )
+
+        domain_map = {"tenant_a": ["example.com"]}
+        with NamedTemporaryFile("w", suffix=".yaml", delete=False) as map_file:
+            import yaml
+
+            yaml.dump(domain_map, map_file)
+            map_path = map_file.name
+        self.addCleanup(lambda: os.path.exists(map_path) and os.remove(map_path))
+
+        config = f"""[general]
+save_smtp_tls = true
+silent = true
+index_prefix_domain_map = {map_path}
+
+[imap]
+host = imap.example.com
+user = test-user
+password = test-password
+
+[smtp]
+host = smtp.example.com
+user = smtp-user
+password = smtp-password
+from = dmarc@example.com
+to = admin@example.com
+"""
+        with NamedTemporaryFile("w", suffix=".ini", delete=False) as config_file:
+            config_file.write(config)
+            config_path = config_file.name
+        self.addCleanup(lambda: os.path.exists(config_path) and os.remove(config_path))
+
+        with patch.object(sys, "argv", ["parsedmarc", "-c", config_path]):
+            parsedmarc.cli._main()
+
+        mock_email_results.assert_called_once()
+        emailed_results = mock_email_results.call_args.args[0]
+        report_ids = {r["report_id"] for r in emailed_results["smtp_tls_reports"]}
+        self.assertEqual(report_ids, {"allowed-1", "mixed-case-1"})
+
+
+class TestMailboxSaveCallbackWiring(unittest.TestCase):
+    """The CLI's end of the #242 contract: the callback it hands to
+    get_dmarc_reports_from_mailbox() (and to watch_inbox()) reports whether
+    a batch reached every configured output destination, so the library
+    knows whether archiving that batch is safe."""
+
+    def setUp(self):
+        # _main()'s file-parsing loop dedups aggregate reports against this
+        # process-wide cache, so a sample parsed here would be dropped from
+        # a later test's results (and vice versa).
+        parsedmarc.SEEN_AGGREGATE_REPORT_IDS.clear()
+        self.addCleanup(parsedmarc.SEEN_AGGREGATE_REPORT_IDS.clear)
+
+    def _write_config(self, text):
+        with NamedTemporaryFile("w", suffix=".ini", delete=False) as config_file:
+            config_file.write(text)
+            config_path = config_file.name
+        self.addCleanup(lambda: os.path.exists(config_path) and os.remove(config_path))
+        return config_path
+
+    @patch("parsedmarc.cli.elastic.save_aggregate_report_to_elasticsearch")
+    @patch("parsedmarc.cli.elastic.migrate_indexes")
+    @patch("parsedmarc.cli.elastic.set_hosts")
+    @patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
+    @patch("parsedmarc.cli.IMAPConnection")
+    def testSaveCallbackIsFalseOnlyWhenAnOutputDestinationFailed(
+        self,
+        mock_imap_connection,
+        mock_get_reports,
+        _mock_set_hosts,
+        _mock_migrate_indexes,
+        mock_save_aggregate,
+    ):
+        """The save_callback saves the batch it is given and returns a
+        truthy value only when every destination accepted it. That verdict
+        is the whole basis on which get_dmarc_reports_from_mailbox() decides
+        whether to archive or delete the batch's messages."""
+        mock_imap_connection.return_value = object()
+        mock_get_reports.return_value = {
+            "aggregate_reports": [],
+            "failure_reports": [],
+            "smtp_tls_reports": [],
+        }
+        config_path = self._write_config(
+            """[general]
+save_aggregate = true
+silent = true
+
+[imap]
+host = imap.example.com
+user = test-user
+password = test-password
+
+[elasticsearch]
+hosts = localhost
+"""
+        )
+
+        with patch.object(sys, "argv", ["parsedmarc", "-c", config_path]):
+            parsedmarc.cli._main()
+
+        save_callback = mock_get_reports.call_args.kwargs.get("save_callback")
+        self.assertTrue(callable(save_callback))
+
+        report = {"policy_published": {"domain": "example.com"}}
+        batch = {
+            "aggregate_reports": [report],
+            "failure_reports": [],
+            "smtp_tls_reports": [],
+        }
+        self.assertTrue(save_callback(batch))
+        self.assertIs(mock_save_aggregate.call_args.args[0], report)
+
+        mock_save_aggregate.side_effect = parsedmarc.elastic.ElasticsearchError(
+            "simulated output failure"
+        )
+        self.assertFalse(save_callback(batch))
+
+    @patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
+    @patch("parsedmarc.cli.watch_inbox")
+    @patch("parsedmarc.cli.IMAPConnection")
+    def testWatchCallbackIsTheSameAdapterAndRetryCapReachesBothCallSites(
+        self, mock_imap_connection, mock_watch_inbox, mock_get_reports
+    ):
+        """watch_inbox's callback must be the same save-then-report adapter
+        the single-shot fetch gets, not process_reports directly -- watch
+        mode would otherwise still archive before confirming the save. The
+        configured max_unsaved_retries has to reach both call sites too, or
+        one of the two paths would silently use the default cap."""
+        mock_imap_connection.return_value = object()
+        mock_get_reports.return_value = {
+            "aggregate_reports": [],
+            "failure_reports": [],
+            "smtp_tls_reports": [],
+        }
+        mock_watch_inbox.side_effect = FileExistsError("stop-watch-loop")
+        config_path = self._write_config(
+            """[general]
+silent = true
+
+[imap]
+host = imap.example.com
+user = user
+password = pass
+
+[mailbox]
+watch = true
+max_unsaved_retries = 7
+"""
+        )
+
+        with patch.object(sys, "argv", ["parsedmarc", "-c", config_path]):
+            with self.assertRaises(SystemExit) as system_exit:
+                parsedmarc.cli._main()
+
+        self.assertEqual(system_exit.exception.code, 1)
+        single_shot_kwargs = mock_get_reports.call_args.kwargs
+        watch_kwargs = mock_watch_inbox.call_args.kwargs
+        self.assertTrue(callable(single_shot_kwargs.get("save_callback")))
+        self.assertIs(watch_kwargs.get("callback"), single_shot_kwargs["save_callback"])
+        self.assertEqual(single_shot_kwargs.get("max_unsaved_retries"), 7)
+        self.assertEqual(watch_kwargs.get("max_unsaved_retries"), 7)
+
+    @patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
+    @patch("parsedmarc.cli.IMAPConnection")
+    def testFileOutputFailureIsRecordedAndReportsTheBatchUnsaved(
+        self, mock_imap_connection, mock_get_reports
+    ):
+        """A failed --output write is recorded like any other destination's
+        failure instead of escaping uncaught, so it blocks archiving too.
+        Pointing `output` at a regular file makes save_output() raise the
+        real ValueError it raises for a non-directory path."""
+        mock_imap_connection.return_value = object()
+        with NamedTemporaryFile("w", suffix=".txt", delete=False) as not_a_directory:
+            output_path = not_a_directory.name
+        self.addCleanup(lambda: os.path.exists(output_path) and os.remove(output_path))
+
+        reports_dict = {
+            "aggregate_reports": [{"policy_published": {"domain": "example.com"}}],
+            "failure_reports": [],
+            "smtp_tls_reports": [],
+        }
+        verdicts = []
+
+        def _fetch_and_save(**kwargs):
+            verdicts.append(kwargs["save_callback"](reports_dict))
+            return reports_dict
+
+        mock_get_reports.side_effect = _fetch_and_save
+        config_path = self._write_config(
+            f"""[general]
+silent = true
+output = {output_path}
+
+[imap]
+host = imap.example.com
+user = test-user
+password = test-password
+"""
+        )
+
+        with patch.object(sys, "argv", ["parsedmarc", "-c", config_path]):
+            with self.assertLogs("parsedmarc.log", level="ERROR") as cm:
+                parsedmarc.cli._main()
+
+        self.assertEqual(verdicts, [False])
+        self.assertTrue(
+            any("File output Error" in line for line in cm.output), cm.output
+        )
+
+    @patch("parsedmarc.cli.elastic.save_aggregate_report_to_elasticsearch")
+    @patch("parsedmarc.cli.elastic.migrate_indexes")
+    @patch("parsedmarc.cli.elastic.set_hosts")
+    @patch("parsedmarc.cli.get_dmarc_reports_from_mailbox")
+    @patch("parsedmarc.cli.IMAPConnection")
+    def testMailboxAndFileReportsAreEachSavedExactlyOnce(
+        self,
+        mock_imap_connection,
+        mock_get_reports,
+        _mock_set_hosts,
+        _mock_migrate_indexes,
+        mock_save_aggregate,
+    ):
+        """A run combining a file argument and a mailbox saves in two passes:
+        the mailbox batch inside save_callback, the file-derived reports
+        afterward. The final pass must run on the file snapshot alone --
+        running it on the combined results would send every mailbox report
+        to every destination a second time."""
+        mock_imap_connection.return_value = object()
+        mailbox_report = {"policy_published": {"domain": "mailbox.example.com"}}
+        mock_get_reports.side_effect = _fetch_invoking_save_callback(
+            {
+                "aggregate_reports": [mailbox_report],
+                "failure_reports": [],
+                "smtp_tls_reports": [],
+            }
+        )
+        config_path = self._write_config(
+            """[general]
+save_aggregate = true
+silent = true
+offline = true
+
+[imap]
+host = imap.example.com
+user = test-user
+password = test-password
+
+[elasticsearch]
+hosts = localhost
+"""
+        )
+
+        argv = ["parsedmarc", "-c", config_path, SAMPLE_AGGREGATE_REPORT_PATH]
+        with patch.object(sys, "argv", argv):
+            parsedmarc.cli._main()
+
+        saved_domains = [
+            call.args[0]["policy_published"]["domain"]
+            for call in mock_save_aggregate.call_args_list
+        ]
+        self.assertEqual(saved_domains.count("mailbox.example.com"), 1)
+        self.assertEqual(len(saved_domains), 2)
+
+    @patch("parsedmarc.cli.elastic.save_aggregate_report_to_elasticsearch")
+    @patch("parsedmarc.cli.elastic.migrate_indexes")
+    @patch("parsedmarc.cli.elastic.set_hosts")
+    def testFailOnOutputErrorExitsForFileDerivedReports(
+        self,
+        _mock_set_hosts,
+        _mock_migrate_indexes,
+        mock_save_aggregate,
+    ):
+        """fail_on_output_error still exits non-zero for reports that came
+        from a file argument rather than a mailbox: with no mailbox
+        connection there is no save_callback to raise inside, so the failure
+        surfaces from the final pass over the file snapshot instead."""
+        mock_save_aggregate.side_effect = parsedmarc.elastic.ElasticsearchError(
+            "simulated output failure"
+        )
+        config_path = self._write_config(
+            """[general]
+save_aggregate = true
+fail_on_output_error = true
+silent = true
+offline = true
+
+[elasticsearch]
+hosts = localhost
+"""
+        )
+
+        argv = ["parsedmarc", "-c", config_path, SAMPLE_AGGREGATE_REPORT_PATH]
+        with patch.object(sys, "argv", argv):
+            with self.assertRaises(SystemExit) as system_exit:
+                parsedmarc.cli._main()
+
+        self.assertEqual(system_exit.exception.code, 1)
+        mock_save_aggregate.assert_called_once()
 
 
 class TestConfigAliases(unittest.TestCase):
@@ -4655,6 +5042,7 @@ class TestParseConfigMailbox(unittest.TestCase):
                 "test": "false",
                 "batch_size": "25",
                 "check_timeout": "60",
+                "max_unsaved_retries": "5",
                 "since": "3d",
             },
         )
@@ -4672,8 +5060,22 @@ class TestParseConfigMailbox(unittest.TestCase):
         self.assertIs(opts.mailbox_test, False)
         self.assertEqual(opts.mailbox_batch_size, 25)
         self.assertEqual(opts.mailbox_check_timeout, 60)
+        self.assertEqual(opts.mailbox_max_unsaved_retries, 5)
         self.assertEqual(opts.mailbox_since, "3d")
 
+    def test_mailbox_max_unsaved_retries_from_env_var(self):
+        """PARSEDMARC_MAILBOX_MAX_UNSAVED_RETRIES resolves to
+        [mailbox] max_unsaved_retries and parses as an int -- including 0,
+        which is a meaningful value ("never retry"), not an absent one."""
+        from parsedmarc.cli import _load_config, _parse_config
+
+        env = {"PARSEDMARC_MAILBOX_MAX_UNSAVED_RETRIES": "0"}
+        with patch.dict(os.environ, env, clear=False):
+            config = _load_config(None)
+        opts = _opts()
+        _parse_config(config, opts)
+        self.assertEqual(opts.mailbox_max_unsaved_retries, 0)
+
     def test_mailbox_absent_per_type_delete_keys_are_not_set(self):
         from parsedmarc.cli import _parse_config
 
@@ -5547,11 +5949,13 @@ host = db.example.com
             patch("parsedmarc.cli.postgres.PostgreSQLClient") as mock_client_cls,
             patch(
                 "parsedmarc.cli.get_dmarc_reports_from_mailbox",
-                return_value={
-                    "aggregate_reports": [report],
-                    "failure_reports": [],
-                    "smtp_tls_reports": [],
-                },
+                side_effect=_fetch_invoking_save_callback(
+                    {
+                        "aggregate_reports": [report],
+                        "failure_reports": [],
+                        "smtp_tls_reports": [],
+                    }
+                ),
             ),
             patch("parsedmarc.cli.IMAPConnection", return_value=object()),
             patch.object(sys, "argv", ["parsedmarc", "-c", config_path]),
@@ -5593,7 +5997,7 @@ host = db.example.com
             patch("parsedmarc.cli.postgres.PostgreSQLClient") as mock_client_cls,
             patch(
                 "parsedmarc.cli.get_dmarc_reports_from_mailbox",
-                return_value=reports,
+                side_effect=_fetch_invoking_save_callback(reports),
             ),
             patch("parsedmarc.cli.IMAPConnection", return_value=object()),
             patch.object(sys, "argv", ["parsedmarc", "-c", config_path]),

tests/test_init.py

@@ -1774,6 +1774,19 @@ class TestExtractReport(unittest.TestCase):
         result = parsedmarc.extract_report(compressed)
         self.assertIn("<feedback>", result)
 
+    def testExtractReportFromPlainJson(self):
+        """extract_report passes uncompressed JSON bytes through as text.
+
+        Regression test: MAGIC_JSON was written as b"\\7b", which Python
+        reads as the octal escape \\7 (BEL, 0x07) followed by a literal
+        "b" -- not 0x7B, the "{" every RFC 8259 JSON object begins with --
+        so plain JSON was rejected as "Not a valid zip, gzip, json, or
+        xml file". Every in-tree caller pre-guarded with its own zip/gzip
+        or "{" check, which masked the dead branch."""
+        json_bytes = b'{"organization-name": "Example"}'
+        result = parsedmarc.extract_report(json_bytes)
+        self.assertEqual(result, '{"organization-name": "Example"}')
+
     def testExtractReportFromZip(self):
         """extract_report handles zip compressed content"""
         import zipfile
@@ -3004,6 +3017,32 @@ class TestGetDmarcReportsFromMailboxValidation(unittest.TestCase):
             )
         self.assertIn("connection", str(ctx.exception).lower())
 
+    def test_negative_max_unsaved_retries_raises(self):
+        """max_unsaved_retries is user-configurable (INI/env/kwarg), so an
+        invalid negative value is rejected at the door with a clear error
+        instead of silently behaving like 0 (move to Unsaved on the first
+        failed save)."""
+        with self.assertRaises(ValueError) as ctx:
+            parsedmarc.get_dmarc_reports_from_mailbox(
+                connection=MagicMock(), max_unsaved_retries=-1
+            )
+        self.assertIn("max_unsaved_retries", str(ctx.exception))
+
+    def test_watch_inbox_negative_max_unsaved_retries_raises(self):
+        """watch_inbox validates before entering the watch loop: raised
+        inside a check, the ValueError would be swallowed and endlessly
+        retried by the IMAP and Maildir backends' per-check exception
+        handling instead of surfacing to the caller."""
+        conn = MagicMock()
+        with self.assertRaises(ValueError) as ctx:
+            parsedmarc.watch_inbox(
+                mailbox_connection=conn,
+                callback=lambda batch: None,
+                max_unsaved_retries=-1,
+            )
+        self.assertIn("max_unsaved_retries", str(ctx.exception))
+        conn.watch.assert_not_called()
+
 
 class TestMigrateForensicArchiveFolderErrorHandling(unittest.TestCase):
     """The one migration scenario a real on-disk Maildir can't reproduce: a
@@ -3455,6 +3494,480 @@ class TestGetDmarcReportsFromMailboxMaildir(unittest.TestCase):
         self.assertEqual(len(conn.fetch_messages("Archive/SMTP-TLS")), 1)
 
 
+class _UncreatableFolderMaildirConnection(MaildirConnection):
+    """A MaildirConnection whose folder_exists() always fails.
+
+    Real backends can reject a folder listing (permissions, a dropped
+    connection, an IMAP server that dislikes the name). The Unsaved holding
+    folder's defensive pre-move check has to warn and carry on rather than
+    crash the run, which is only observable with a backend that fails it.
+    """
+
+    def folder_exists(self, folder_name: str) -> bool:
+        raise RuntimeError("server said no")
+
+
+class TestGetDmarcReportsFromMailboxMaildirSaveCallback(unittest.TestCase):
+    """The save_callback contract of get_dmarc_reports_from_mailbox, on the
+    same real on-disk Maildir harness as the class above (no mocks): a batch
+    is only archived or deleted once the callback confirms it was saved, and
+    a batch that keeps failing eventually moves to the Unsaved holding
+    folder instead of being retried forever (#242)."""
+
+    AGGREGATE = "samples/aggregate/twilight.eml"
+    FAILURE = "samples/failure/dmarc_ruf_report_linkedin.eml"
+    SMTP_TLS = "samples/smtp_tls/google.com_smtp_tls_report.eml"
+    JUNK = b"From: noise@example.com\nSubject: not a report\n\nplain text\n"
+
+    def setUp(self):
+        self._tmp = mkdtemp()
+        self.addCleanup(rmtree, self._tmp, ignore_errors=True)
+        # Both of these are module-global process state: a report "seen" by
+        # an earlier test would be dropped from this test's results, and a
+        # stale retry count would move messages to Unsaved early.
+        parsedmarc.SEEN_AGGREGATE_REPORT_IDS.clear()
+        parsedmarc._FAILED_SAVE_ATTEMPTS.clear()
+        self.addCleanup(parsedmarc._FAILED_SAVE_ATTEMPTS.clear)
+        self._maildir = os.path.join(self._tmp, "Maildir")
+        self._inbox = mailbox.Maildir(self._maildir, create=True)
+
+    def _deliver(self, source):
+        if isinstance(source, str):
+            with open(source, "rb") as source_file:
+                raw = source_file.read()
+        else:
+            raw = source
+        self._inbox.add(mailbox.MaildirMessage(raw))
+        self._inbox.flush()
+
+    def _run(self, connection=None, **kwargs):
+        conn = connection or MaildirConnection(self._maildir, maildir_create=True)
+        result = parsedmarc.get_dmarc_reports_from_mailbox(
+            connection=conn, offline=True, **kwargs
+        )
+        return conn, result
+
+    @staticmethod
+    def _fail(batch):
+        del batch
+        return False
+
+    def test_failed_save_callback_leaves_messages_in_inbox(self):
+        """A save_callback returning False leaves the batch's messages in the
+        INBOX untouched, while the parsed reports are still returned to the
+        caller."""
+        self._deliver(self.AGGREGATE)
+        self._deliver(self.FAILURE)
+
+        conn, result = self._run(save_callback=self._fail)
+
+        self.assertEqual(len(result["aggregate_reports"]), 1)
+        self.assertEqual(len(result["failure_reports"]), 1)
+        self.assertEqual(len(conn.fetch_messages("INBOX")), 2)
+        self.assertEqual(conn.fetch_messages("Archive/Aggregate"), [])
+        self.assertEqual(conn.fetch_messages("Archive/Failure"), [])
+
+    def test_failed_save_logs_an_error_naming_the_reports_folder(self):
+        """The retry is logged at error level, not debug or warning: an
+        output destination that keeps rejecting reports is something
+        operators alert on."""
+        self._deliver(self.AGGREGATE)
+
+        with self.assertLogs("parsedmarc.log", level="ERROR") as cm:
+            self._run(save_callback=self._fail)
+
+        self.assertTrue(
+            any(
+                "Reports were not saved: leaving 1 message(s) in INBOX" in line
+                for line in cm.output
+            ),
+            cm.output,
+        )
+
+    def test_failed_save_then_retry_is_not_deduplicated(self):
+        """A failed save must not poison the aggregate-report dedup cache:
+        the message stays in the INBOX and, on the next run, the same report
+        is parsed and handed to save_callback again rather than being
+        silently skipped as an already-seen duplicate."""
+        self._deliver(self.AGGREGATE)
+
+        conn, result = self._run(save_callback=self._fail)
+        self.assertEqual(len(result["aggregate_reports"]), 1)
+        self.assertEqual(len(conn.fetch_messages("INBOX")), 1)
+
+        received = []
+        conn, result = self._run(save_callback=received.append)
+
+        self.assertEqual(len(received), 1)
+        self.assertEqual(len(received[0]["aggregate_reports"]), 1)
+        self.assertEqual(len(result["aggregate_reports"]), 1)
+        self.assertEqual(conn.fetch_messages("INBOX"), [])
+        self.assertEqual(len(conn.fetch_messages("Archive/Aggregate")), 1)
+
+    def test_save_callback_exception_leaves_messages(self):
+        """An exception raised by save_callback counts as a failed save --
+        the batch's messages stay in the reports folder and the attempt is
+        recorded against the retry cap -- and is then re-raised to the
+        caller."""
+        self._deliver(self.AGGREGATE)
+
+        def _boom(batch):
+            del batch
+            raise RuntimeError("sink is down")
+
+        with self.assertRaises(RuntimeError):
+            self._run(save_callback=_boom)
+
+        conn = MaildirConnection(self._maildir, maildir_create=True)
+        self.assertEqual(len(conn.fetch_messages("INBOX")), 1)
+        self.assertEqual(len(parsedmarc._FAILED_SAVE_ATTEMPTS), 1)
+        # The dedup keys were staged, not committed, so the next run
+        # reparses the report instead of dropping it as a duplicate.
+        _, result = self._run(save_callback=None)
+        self.assertEqual(len(result["aggregate_reports"]), 1)
+
+    def test_save_callback_exception_still_bounded_by_the_retry_cap(self):
+        """A callback that always raises -- the CLI's does, under
+        fail_on_output_error -- is bounded by max_unsaved_retries exactly
+        like one that returns False: the failure bookkeeping runs before the
+        exception is re-raised, so the message moves to Archive/Unsaved at
+        the cap instead of being re-delivered on every check forever. This
+        matters because mailsuite's IMAP and Maildir watch loops swallow the
+        exception and keep checking."""
+        self._deliver(self.AGGREGATE)
+
+        def _boom(batch):
+            del batch
+            raise RuntimeError("sink is down")
+
+        with self.assertRaises(RuntimeError):
+            self._run(save_callback=_boom, max_unsaved_retries=0)
+
+        conn = MaildirConnection(self._maildir, maildir_create=True)
+        self.assertEqual(conn.fetch_messages("INBOX"), [])
+        self.assertEqual(len(conn.fetch_messages("Archive/Unsaved")), 1)
+        self.assertEqual(parsedmarc._FAILED_SAVE_ATTEMPTS, {})
+
+    def test_successful_save_callback_archives_messages(self):
+        """A save_callback returning None (or any non-False value) commits the
+        batch exactly as when no callback is supplied, and is handed that
+        batch's parsed reports before any archiving happens."""
+        self._deliver(self.AGGREGATE)
+        self._deliver(self.FAILURE)
+        self._deliver(self.SMTP_TLS)
+
+        conn = MaildirConnection(self._maildir, maildir_create=True)
+        received = []
+
+        def _record(batch):
+            # Nothing has been archived yet at this point -- that is the
+            # whole contract: the caller gets to veto the archiving.
+            received.append((batch, len(conn.fetch_messages("INBOX"))))
+
+        _, result = self._run(connection=conn, save_callback=_record)
+
+        self.assertEqual(len(received), 1)
+        batch, inbox_count_at_callback_time = received[0]
+        self.assertEqual(len(batch["aggregate_reports"]), 1)
+        self.assertEqual(len(batch["failure_reports"]), 1)
+        self.assertEqual(len(batch["smtp_tls_reports"]), 1)
+        self.assertEqual(inbox_count_at_callback_time, 3)
+
+        self.assertEqual(len(result["aggregate_reports"]), 1)
+        self.assertEqual(conn.fetch_messages("INBOX"), [])
+        self.assertEqual(len(conn.fetch_messages("Archive/Aggregate")), 1)
+        self.assertEqual(len(conn.fetch_messages("Archive/Failure")), 1)
+        self.assertEqual(len(conn.fetch_messages("Archive/SMTP-TLS")), 1)
+
+    def test_failed_save_callback_with_delete_does_not_delete(self):
+        """delete=True does not bypass the save_callback contract: a failed
+        save leaves the message in place instead of deleting it, since the
+        mailbox is the only remaining copy of the report."""
+        self._deliver(self.FAILURE)
+
+        conn, result = self._run(delete=True, save_callback=self._fail)
+
+        self.assertEqual(len(result["failure_reports"]), 1)
+        self.assertEqual(len(conn.fetch_messages("INBOX")), 1)
+
+    def test_invalid_message_still_filed_when_save_fails(self):
+        """An unparseable message carries no report data, so nothing about it
+        can fail to save: it is filed to Archive/Invalid regardless of the
+        callback's verdict, and only the successfully parsed report's message
+        is held back."""
+        self._deliver(self.JUNK)
+        self._deliver(self.AGGREGATE)
+
+        conn, result = self._run(save_callback=self._fail)
+
+        self.assertEqual(len(result["aggregate_reports"]), 1)
+        self.assertEqual(len(conn.fetch_messages("Archive/Invalid")), 1)
+        self.assertEqual(len(conn.fetch_messages("INBOX")), 1)
+
+    def test_unsaved_folder_created_only_with_a_save_callback(self):
+        """The Unsaved holding folder is only reachable when a callback can
+        report a batch unsaved, so it is only created up front in that case
+        -- it would otherwise sit empty in every user's mailbox. The
+        without-callback half runs in a second, fresh Maildir: the first run
+        already created the folder in the shared one, so its absence is only
+        observable somewhere the callback never touched."""
+        self._deliver(self.AGGREGATE)
+
+        conn, _ = self._run(save_callback=lambda batch: None)
+        self.assertTrue(conn.folder_exists("Archive/Unsaved"))
+
+        other_maildir = os.path.join(self._tmp, "MaildirNoCallback")
+        other_inbox = mailbox.Maildir(other_maildir, create=True)
+        with open(self.FAILURE, "rb") as failure_file:
+            other_inbox.add(mailbox.MaildirMessage(failure_file.read()))
+        other_inbox.flush()
+        other_conn = MaildirConnection(other_maildir, maildir_create=True)
+        parsedmarc.get_dmarc_reports_from_mailbox(connection=other_conn, offline=True)
+        self.assertTrue(other_conn.folder_exists("Archive/Failure"))
+        self.assertFalse(other_conn.folder_exists("Archive/Unsaved"))
+
+    def test_repeated_failures_move_the_message_to_unsaved(self):
+        """With the default max_unsaved_retries of 2, a message stays in the
+        reports folder through its first failed save and the retry after it,
+        and moves to Archive/Unsaved on the third -- the initial attempt plus
+        two retries, so a permanently broken output destination receives the
+        same reports at most three times instead of forever."""
+        self._deliver(self.AGGREGATE)
+
+        for _ in range(2):
+            conn, _ = self._run(save_callback=self._fail)
+            self.assertEqual(len(conn.fetch_messages("INBOX")), 1)
+            self.assertEqual(conn.fetch_messages("Archive/Unsaved"), [])
+
+        with self.assertLogs("parsedmarc.log", level="ERROR") as cm:
+            conn, _ = self._run(save_callback=self._fail)
+
+        self.assertEqual(conn.fetch_messages("INBOX"), [])
+        self.assertEqual(len(conn.fetch_messages("Archive/Unsaved")), 1)
+        self.assertTrue(any("Archive/Unsaved" in line for line in cm.output), cm.output)
+
+    def test_max_unsaved_retries_zero_moves_on_the_first_failure(self):
+        """max_unsaved_retries=0 is valid and means "do not retry": the
+        message moves to Archive/Unsaved the first time its batch fails to
+        save."""
+        self._deliver(self.AGGREGATE)
+
+        conn, _ = self._run(save_callback=self._fail, max_unsaved_retries=0)
+
+        self.assertEqual(conn.fetch_messages("INBOX"), [])
+        self.assertEqual(len(conn.fetch_messages("Archive/Unsaved")), 1)
+
+    def test_unsaved_messages_are_moved_not_deleted(self):
+        """A message that ran out of retries is moved to Unsaved even when
+        every delete flag says delete: it holds the only copy of a report
+        that was never saved anywhere, so deleting it is the one outcome
+        this feature exists to prevent."""
+        self._deliver(self.AGGREGATE)
+
+        conn, _ = self._run(
+            save_callback=self._fail, max_unsaved_retries=0, delete=True
+        )
+
+        self.assertEqual(conn.fetch_messages("INBOX"), [])
+        self.assertEqual(len(conn.fetch_messages("Archive/Unsaved")), 1)
+
+    def test_successful_save_clears_the_retry_counters(self):
+        """A successful save resets the batch's failure counts, so an
+        unrelated failure later starts from zero rather than inheriting a
+        near-cap count. The counters are asserted directly because the
+        messages they were counting are archived on success, taking their
+        own future behavior with them."""
+        self._deliver(self.AGGREGATE)
+
+        conn, _ = self._run(save_callback=self._fail)
+        self.assertEqual(len(parsedmarc._FAILED_SAVE_ATTEMPTS), 1)
+
+        conn, _ = self._run(save_callback=lambda batch: True)
+
+        self.assertEqual(parsedmarc._FAILED_SAVE_ATTEMPTS, {})
+        self.assertEqual(len(conn.fetch_messages("Archive/Aggregate")), 1)
+
+    def test_test_mode_invokes_the_callback_without_touching_anything(self):
+        """test=True still runs the callback, so a test run exercises the
+        whole output pipeline, but neither the mailbox nor the retry
+        counters move regardless of the verdict -- even at
+        max_unsaved_retries=0, which would otherwise file the message under
+        Unsaved immediately."""
+        self._deliver(self.AGGREGATE)
+
+        received = []
+
+        def _record_and_fail(batch):
+            received.append(batch)
+            return False
+
+        conn = None
+        for _ in range(2):
+            conn, _ = self._run(
+                save_callback=_record_and_fail, max_unsaved_retries=0, test=True
+            )
+        assert conn is not None
+
+        self.assertEqual(len(received), 2)
+        self.assertEqual(len(received[0]["aggregate_reports"]), 1)
+        self.assertEqual(len(conn.fetch_messages("INBOX")), 1)
+        self.assertFalse(conn.folder_exists("Archive/Unsaved"))
+        self.assertEqual(parsedmarc._FAILED_SAVE_ATTEMPTS, {})
+
+    def test_unsaved_folder_is_created_when_folder_creation_was_skipped(self):
+        """Watch mode calls in with create_folders=False, so the Unsaved
+        folder may not exist by the time a message needs to go there. The
+        defensive check before the move creates it."""
+        self._deliver(self.AGGREGATE)
+        conn = MaildirConnection(self._maildir, maildir_create=True)
+        conn.create_folder("Archive")
+
+        self._run(
+            connection=conn,
+            save_callback=self._fail,
+            max_unsaved_retries=0,
+            create_folders=False,
+        )
+
+        self.assertEqual(len(conn.fetch_messages("Archive/Unsaved")), 1)
+
+    def test_unsaved_folder_check_failure_is_logged_not_raised(self):
+        """A backend that cannot answer whether the Unsaved folder exists is
+        warned about and skipped rather than crashing the run; the move
+        itself is still attempted."""
+        self._deliver(self.AGGREGATE)
+        conn = _UncreatableFolderMaildirConnection(self._maildir, maildir_create=True)
+        conn.create_folder("Archive")
+
+        with self.assertLogs("parsedmarc.log", level="WARNING") as cm:
+            self._run(
+                connection=conn,
+                save_callback=self._fail,
+                max_unsaved_retries=0,
+                create_folders=False,
+            )
+
+        self.assertTrue(
+            any(
+                "Could not create folder Archive/Unsaved" in line for line in cm.output
+            ),
+            cm.output,
+        )
+        self.assertEqual(len(conn.fetch_messages("Archive/Unsaved")), 1)
+
+    def test_failed_move_to_unsaved_is_logged_and_leaves_the_message(self):
+        """A backend that rejects the move to Unsaved is logged like any
+        other mailbox error, and the message simply stays in the reports
+        folder -- it is never dropped."""
+        self._deliver(self.AGGREGATE)
+        conn = _FailingDisposalMaildirConnection(
+            self._maildir, maildir_create=True, fail_move=True
+        )
+
+        with self.assertLogs("parsedmarc.log", level="ERROR") as cm:
+            self._run(connection=conn, save_callback=self._fail, max_unsaved_retries=0)
+
+        self.assertTrue(
+            any(
+                "Mailbox error: Error moving message UID" in line for line in cm.output
+            ),
+            cm.output,
+        )
+        self.assertEqual(len(conn.fetch_messages("INBOX")), 1)
+        self.assertEqual(conn.fetch_messages("Archive/Unsaved"), [])
+
+    def test_failed_move_keeps_the_counter_so_the_move_is_retried(self):
+        """A failed move to Unsaved must not reset the message's failure
+        counter: the still-in-place message would otherwise get a fresh set
+        of under-cap retries (and duplicate deliveries) each time the move
+        failed. With the counter kept, the next failed save classifies the
+        message over-cap again and re-attempts the move -- which succeeds
+        here once the backend allows it. Cap 1 makes the distinction
+        observable: were the counter reset by run 2's failed move, run 3
+        would count the message back under the cap and leave it in the
+        INBOX instead of moving it."""
+        self._deliver(self.AGGREGATE)
+
+        conn, _ = self._run(save_callback=self._fail, max_unsaved_retries=1)
+        self.assertEqual(len(conn.fetch_messages("INBOX")), 1)
+
+        failing_conn = _FailingDisposalMaildirConnection(
+            self._maildir, maildir_create=True, fail_move=True
+        )
+        with self.assertLogs("parsedmarc.log", level="ERROR"):
+            self._run(
+                connection=failing_conn,
+                save_callback=self._fail,
+                max_unsaved_retries=1,
+            )
+        self.assertEqual(len(failing_conn.fetch_messages("INBOX")), 1)
+        self.assertEqual(len(parsedmarc._FAILED_SAVE_ATTEMPTS), 1)
+
+        conn, _ = self._run(save_callback=self._fail, max_unsaved_retries=1)
+
+        self.assertEqual(conn.fetch_messages("INBOX"), [])
+        self.assertEqual(len(conn.fetch_messages("Archive/Unsaved")), 1)
+        self.assertEqual(parsedmarc._FAILED_SAVE_ATTEMPTS, {})
+
+    def test_failed_save_skips_the_batch_size_zero_re_check(self):
+        """With batch_size=0 the function re-checks the folder and recurses
+        on whatever is still there. After a failed save that would re-fetch
+        the very messages just held back and burn the whole retry budget
+        inside one call, so the re-check is skipped: one run costs a message
+        exactly one retry."""
+        self._deliver(self.AGGREGATE)
+
+        conn, result = self._run(
+            save_callback=self._fail, batch_size=0, max_unsaved_retries=1
+        )
+
+        self.assertEqual(len(result["aggregate_reports"]), 1)
+        self.assertEqual(len(conn.fetch_messages("INBOX")), 1)
+        self.assertEqual(conn.fetch_messages("Archive/Unsaved"), [])
+
+    def test_save_callback_receives_each_batch_separately(self):
+        """batch_size=0 recursion hands each fetched batch to the callback on
+        its own, and each batch is archived before the next is fetched --
+        the callback never sees an earlier batch's reports twice, while the
+        returned results accumulate all of them."""
+        self._deliver(self.AGGREGATE)
+        conn = _MidRunArrivalMaildirConnection(
+            self._maildir, maildir_create=True, extra_source=self.SMTP_TLS
+        )
+        received = []
+
+        _, result = self._run(
+            connection=conn, save_callback=received.append, batch_size=0
+        )
+
+        self.assertEqual(len(received), 2)
+        self.assertEqual(len(received[0]["aggregate_reports"]), 1)
+        self.assertEqual(received[0]["smtp_tls_reports"], [])
+        self.assertEqual(received[1]["aggregate_reports"], [])
+        self.assertEqual(len(received[1]["smtp_tls_reports"]), 1)
+        self.assertEqual(len(result["aggregate_reports"]), 1)
+        self.assertEqual(len(result["smtp_tls_reports"]), 1)
+        self.assertEqual(conn.fetch_messages("INBOX"), [])
+
+    def test_failed_save_callback_parallel(self):
+        """The n_procs>1 parsing branch stages its dedup keys and holds its
+        messages back exactly like the sequential one; only parsing moves to
+        the worker processes."""
+        self._deliver(self.AGGREGATE)
+        self._deliver(self.FAILURE)
+
+        conn, result = self._run(save_callback=self._fail, n_procs=2)
+
+        self.assertEqual(len(result["aggregate_reports"]), 1)
+        self.assertEqual(len(result["failure_reports"]), 1)
+        self.assertEqual(len(conn.fetch_messages("INBOX")), 2)
+
+        # The dedup keys were not committed, so a retry reparses.
+        conn, result = self._run(save_callback=None, n_procs=2)
+        self.assertEqual(len(result["aggregate_reports"]), 1)
+        self.assertEqual(len(conn.fetch_messages("Archive/Aggregate")), 1)
+
+
 class _MidRunArrivalMaildirConnection(MaildirConnection):
     """A MaildirConnection that delivers one extra message into the INBOX
     right after its first fetch_messages() call returns, simulating mail
@@ -3625,6 +4138,57 @@ class TestWatchInboxMaildir(unittest.TestCase):
         # deleted rather than filed in the Invalid subfolder.
         self.assertFalse(conn.folder_exists("Archive/Invalid"))
 
+    def test_watch_inbox_callback_returning_false_defers_the_batch(self):
+        """watch_inbox's callback is the batch's save_callback, not a
+        notification that runs after archiving: returning False leaves the
+        failure report message in the INBOX for the next check. Before this
+        was wired through, watch mode archived every batch first and told the
+        callback afterward, which is the data-loss path of #242."""
+        parsedmarc._FAILED_SAVE_ATTEMPTS.clear()
+        self.addCleanup(parsedmarc._FAILED_SAVE_ATTEMPTS.clear)
+        conn = _SingleCheckMaildirConnection(self._maildir, maildir_create=True)
+        conn.create_folder("Archive")
+        conn.create_folder("Archive/Invalid")
+        received = []
+
+        def _record_and_fail(batch):
+            received.append(batch)
+            return False
+
+        parsedmarc.watch_inbox(
+            mailbox_connection=conn,
+            callback=_record_and_fail,
+            offline=True,
+        )
+
+        self.assertEqual(len(received), 1)
+        self.assertEqual(len(received[0]["failure_reports"]), 1)
+        self.assertEqual(len(conn.fetch_messages("INBOX")), 1)
+        self.assertEqual(conn.fetch_messages("Archive/Failure"), [])
+        # The unparseable message carries no report data and is filed either
+        # way, so only the report message is held back.
+        self.assertEqual(len(conn.fetch_messages("Archive/Invalid")), 1)
+
+    def test_watch_inbox_forwards_max_unsaved_retries(self):
+        """max_unsaved_retries reaches the per-check mailbox call: with 0,
+        the first failed save in watch mode files the message under
+        Archive/Unsaved instead of leaving it for the next check."""
+        parsedmarc._FAILED_SAVE_ATTEMPTS.clear()
+        self.addCleanup(parsedmarc._FAILED_SAVE_ATTEMPTS.clear)
+        conn = _SingleCheckMaildirConnection(self._maildir, maildir_create=True)
+        conn.create_folder("Archive")
+        conn.create_folder("Archive/Invalid")
+
+        parsedmarc.watch_inbox(
+            mailbox_connection=conn,
+            callback=lambda batch: False,
+            offline=True,
+            max_unsaved_retries=0,
+        )
+
+        self.assertEqual(conn.fetch_messages("INBOX"), [])
+        self.assertEqual(len(conn.fetch_messages("Archive/Unsaved")), 1)
+
 
 class TestEmailResultsErrorBranches(unittest.TestCase):
     """email_results requires mail_to to be a list — this is enforced