doc: update to 1.5.0

When a binary is forwarded WITHOUT its paired _ASCII.TXT (because
the TXT wasn't quiescent within the grace period — BW slow to
write, AV scanning, etc.), the old behaviour was to permanently
mark the binary as "done" in the state file, even though the TXT
might land seconds later.  Result: that event lived in SFM forever
with broken-codec peak values and no project info.

Fix: state entries now carry a had_report flag.  Forwards without
a TXT set had_report=False.  On subsequent scans, the watcher
treats had_report=False entries as re-pair candidates — they get
re-forwarded once the TXT appears, and the SFM server's upsert
path (in seismo-relay's insert_events IntegrityError handler)
refreshes the DB row with the report's authoritative values.

Three status states in ForwardState.status(sha256):
  None  — never forwarded.  First-forward path.
  True  — forwarded successfully WITH report (or legacy entry
          without the had_report field).  Permanently done.
  False — forwarded WITHOUT report.  Re-pair if TXT now exists.

Backward compat: legacy state-file entries (no had_report key)
default to True so existing deployments don't unexpectedly
re-forward every entry on upgrade.

Tests cover:
  - re-pair when TXT appears after a had_report=False forward
  - had_report=True entries stay skipped permanently
  - legacy entries (missing field) treated as fully forwarded
  - state.status() returns None for unknown sha
  - re-marking had_report=False then True promotes to fully-done

36 watcher tests pass (was 31, +5 new).

CHANGELOG.md

@@ -6,38 +6,40 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ---
 
-## [1.5.1] - 2026-05-10
+## [5-11-26] — v1.5.0
 
-### Added
-- **SFM Forward tab in the Settings dialog.**  v1.5.0 shipped the `event_forwarder.py` module + INI keys but missed the GUI; operators had to edit `config.ini` by hand to enable forwarding.  The settings dialog now exposes:
-  - **Forward events to SFM** checkbox
-  - **SFM Server URL** entry with a **Test** button (mirrors the Connection tab — GETs `/health` and shows the result)
-  - **Forward Interval / Quiescence / Missing-Report Grace / HTTP Timeout** spinboxes
-  - **State File** entry with a Browse... button (defaults to `<log dir>/sfm_forwarded.json` when blank)
-- Save-time guard: enabling SFM Forward without filling in the URL shows a validation error rather than silently saving a non-functional config.
+First release of the SFM event forwarder.
 
-## [1.5.0] - 2026-05-09
-
-### Added
-- **SFM event forwarder.**  When `SFM_FORWARD_ENABLED=true` and `SFM_URL` is set, every Blastware event binary is forwarded to an SFM server's `/db/import/blastware_file` endpoint as a multipart POST.  The corresponding `<binary>.TXT` ASCII report (which Blastware's ACH writes alongside each event) is paired by filename and shipped in the same request, letting the SFM server index the full per-channel stats (PPV, ZC Freq, Time of Peak, Peak Acceleration / Displacement, Peak Vector Sum + time, sensor self-check Pass/Fail, monitor-log timestamps) without depending on the still-undecoded Blastware waveform body codec.
+### Added — SFM event forwarder
+- **Forward Blastware event binaries (+ paired BW ACH ASCII reports) to an SFM server.**  When `SFM_FORWARD_ENABLED=true` and `SFM_URL` is set, every event binary in the BW ACH watch folder is POSTed as multipart to `/db/import/blastware_file` along with its `<stem>_<ext>_ASCII.TXT` partner report (BW ACH convention; manual-export `<binary>.TXT` is also supported as a fallback).  SFM parses the report and indexes the full per-channel stats (PPV, ZC Freq, Time of Peak, Peak Acceleration / Displacement, Peak Vector Sum + time, sensor self-check Pass/Fail, monitor-log timestamps) into a searchable database — no codec decoding required.
 - **Idempotent forwarding.**  Forwarded files are tracked by sha256 in a JSON state file (default `<log dir>/sfm_forwarded.json`, override via `SFM_STATE_FILE`).  Re-scans don't re-POST and the state survives restarts / auto-updates.
-- **Quiescence + grace-period guards.**  Files modified within `SFM_QUIESCENCE_SECONDS` (default 5s) are skipped to avoid forwarding mid-write.  If a binary's `.TXT` partner hasn't appeared after `SFM_MISSING_REPORT_GRACE_SECONDS` (default 60s), the binary is forwarded alone rather than blocking forever.
-- New `event_forwarder.py` module + 17 unit tests in `test_event_forwarder.py` covering filename matching, state idempotency, scan logic, multipart encoding, and a fake-server end-to-end POST.
+- **Re-pair on late-arriving TXT.**  When the watcher forwards a binary alone (`_ASCII.TXT` partner didn't appear within `SFM_MISSING_REPORT_GRACE_SECONDS`), the state file records `had_report: false`.  On subsequent scans, the watcher re-checks whether the TXT has since arrived.  If yes, the event is re-forwarded with the TXT attached — the SFM server's upsert path refreshes the DB row with the report's device-authoritative peak / project values.  Without this, slow-disk or AV-interrupted TXT writes would permanently leave that event with broken-codec peaks in the SFM database.  Legacy state-file entries (without the `had_report` field) default to `had_report: true` so an upgrade doesn't unexpectedly re-forward existing entries.
+- **Quiescence + grace-period guards.**  Files modified within `SFM_QUIESCENCE_SECONDS` (default 5s) are skipped to avoid forwarding mid-write.  If a binary's report partner hasn't appeared after `SFM_MISSING_REPORT_GRACE_SECONDS` (default 60s), the binary is forwarded alone rather than blocking forever.
+- **Per-pass rate cap.**  `SFM_MAX_FORWARDS_PER_PASS` (default 500) drips first-deploy backfill instead of hammering the SFM server in one burst.  At 60-second `SFM_FORWARD_INTERVAL_SECONDS` cadence that's ~30K events/hour throughput.  Set to `0` for unlimited.  Scan walks oldest-first so backfill advances chronologically and successive scans reliably progress.
+- **`event_forwarder.py --seed-state` CLI mode.**  Walks the watch folder once, sha256s every in-window event, and marks them all as already-forwarded *without* POSTing anything.  Recommended pre-deploy workflow on machines with a large historical archive — flip `SFM_FORWARD_ENABLED=true` after seeding and only events that appear from then on get forwarded.
+- **SFM Forward tab in the Settings dialog** with: Forward checkbox, SFM URL + Test button (GETs `/health`), Forward Interval / Quiescence / Missing-Report Grace / HTTP Timeout / Max Events Per Pass spinboxes, State File entry with Browse...  Save-time guard: enabling forwarding without a URL shows a validation error.
+- **Histogram-aware log clarity.**  Histogram events (extensions ending in `H`) don't get auto-exported reports from BW; the log distinguishes that case (`(histogram, no report expected)`) from a waveform with unexpectedly missing report (`no report ⚠`).
+- **README "First-time deployment" section** documenting the seed-state workflow + the rate cap as belt-and-suspenders for safe rollout on machines with hundreds of thousands of historical events.
+- 31 new unit tests in `test_event_forwarder.py` covering filename matching, state idempotency, scan logic (quiescence / grace period / max age / already-forwarded / TXT pairing), multipart byte shape, rate cap (oldest-first, cap=0 unlimited, cap=N enforcement), seed-state mode (in-window seeding / max-age skip / end-to-end skip-after-seed / idempotent re-runs), histogram classification, and an end-to-end POST against a stdlib fake server.
 
 ### Configuration
 
 New `[agent]` keys (all default-off — existing 1.4.x deployments don't change behaviour on auto-update):
-- `SFM_FORWARD_ENABLED` (default `false`)
-- `SFM_URL` (e.g. `http://10.0.0.44:8200`)
-- `SFM_FORWARD_INTERVAL_SECONDS` (default `60`)
-- `SFM_QUIESCENCE_SECONDS` (default `5`)
-- `SFM_MISSING_REPORT_GRACE_SECONDS` (default `60`)
-- `SFM_HTTP_TIMEOUT` (default `60`)
-- `SFM_STATE_FILE` (default: `<log dir>/sfm_forwarded.json`)
+
+| Key | Default | Notes |
+|---|---|---|
+| `SFM_FORWARD_ENABLED` | `false` | Master toggle for the forwarder |
+| `SFM_URL` | empty | e.g. `http://10.0.0.44:8200` |
+| `SFM_FORWARD_INTERVAL_SECONDS` | `60` | Scan-and-forward cadence |
+| `SFM_QUIESCENCE_SECONDS` | `5` | Skip files modified in the last N seconds |
+| `SFM_MISSING_REPORT_GRACE_SECONDS` | `60` | Forward without TXT after this delay |
+| `SFM_HTTP_TIMEOUT` | `60` | Per-request HTTP timeout |
+| `SFM_STATE_FILE` | `<log dir>/sfm_forwarded.json` | Override location of the forwarded-sha256 state file |
+| `SFM_MAX_FORWARDS_PER_PASS` | `500` | Per-scan cap (`0` = unlimited) |
 
 ### Compatibility
 
-- Requires SFM server v0.16+ (the `/db/import/blastware_file` endpoint that accepts paired `.TXT` reports — released alongside this watcher version on the seismo-relay side).
+- Requires SFM server v0.16+ (the `/db/import/blastware_file` endpoint that accepts paired `_ASCII.TXT` reports + the BW-report label normalisation — released alongside this watcher version on the seismo-relay side).
 
 ## [1.4.4] - 2026-03-17
 

README.md

@@ -1,4 +1,4 @@
-# Series 3 Watcher v1.5.1
+# Series 3 Watcher v1.5.0
 
 Monitors Instantel **Series 3 (Minimate)** call-in activity on a Blastware server. Runs as a **system tray app** that starts automatically on login, reports heartbeats to terra-view, and self-updates from Gitea.
 
@@ -88,7 +88,7 @@ All settings live in `config.ini`. The Setup Wizard covers every field, but here
 | `UPDATE_SOURCE` | `gitea` (default) or `url` — where to check for updates |
 | `UPDATE_URL` | Base URL of the update server when `UPDATE_SOURCE = url` (e.g. terra-view URL). The watcher fetches `/api/updates/series3-watcher/version.txt` and `/api/updates/series3-watcher/series3-watcher.exe` from this base. |
 
-### SFM Event Forwarder (v1.5.1+)
+### SFM Event Forwarder (v1.5.0+)
 
 Forwards each Blastware event binary (and its paired `<binary>.TXT` ASCII report when present) to an SFM server's `/db/import/blastware_file` endpoint, where the report is parsed and the rich per-channel stats (PPV, ZC Freq, Time of Peak, Peak Acceleration / Displacement, sensor self-check) land in a searchable database. **Default-off** — existing deployments keep their old behaviour after auto-updating until the operator opts in.
 
@@ -101,9 +101,27 @@ Forwards each Blastware event binary (and its paired `<binary>.TXT` ASCII report
 | `SFM_MISSING_REPORT_GRACE_SECONDS` | If a `.TXT` partner hasn't appeared after this many seconds, forward the binary alone (default `60`) |
 | `SFM_HTTP_TIMEOUT` | Per-request HTTP timeout in seconds (default `60`) |
 | `SFM_STATE_FILE` | Path to the JSON state file tracking sha256 of forwarded events. Leave blank to default to `<log dir>/sfm_forwarded.json` |
+| `SFM_MAX_FORWARDS_PER_PASS` | Max events forwarded per scan tick (default `500`, `0` = unlimited). Drip-feeds backfill so a folder with thousands of qualifying events doesn't hammer the SFM server in one giant burst. |
 
 Forwarded files are tracked by sha256 in the state file, so re-scans / restarts / auto-updates never re-POST the same content. A failed POST stays in the pending pool and is retried on the next interval.
 
+#### First-time deployment on a folder with a large historical archive
+
+If you're enabling SFM forwarding on a Blastware ACH machine that's been accumulating events for years (tens or hundreds of thousands of files in the watch folder), you almost certainly **don't** want the watcher to forward all of them on first run. Two options:
+
+1. **Skip the historical backfill (recommended).** Run the seed-state CLI once before flipping `SFM_FORWARD_ENABLED=true`. It walks the folder, sha256s every existing in-window event, and marks them all as already-forwarded — without POSTing anything. The watcher then only forwards events that appear *after* the seed run.
+
+   ```
+   python event_forwarder.py --seed-state ^
+       --watch "C:\Blastware 10\Event\autocall home" ^
+       --state "C:\Users\<you>\AppData\Local\Series3Watcher\agent_logs\sfm_forwarded.json" ^
+       --max-age-days 365
+   ```
+
+2. **Throttle the backfill.** Leave `SFM_MAX_FORWARDS_PER_PASS` at its 500 default and let the watcher drip-feed. With a 60-second `SFM_FORWARD_INTERVAL_SECONDS` that's ~30K events/hour throughput. Backfill of 30K events takes about an hour, 100K takes ~3.5 hours. The cap fires per scan, so heartbeat and forwarding share the watcher's main loop without saturating it.
+
+Combine both for a fully controlled rollout: seed-state to skip the deep archive, then leave the cap on as a steady-state safety net.
+
 ---
 
 ## Tray Icon
@@ -135,9 +153,21 @@ To view connected watchers: **Settings → Developer → Watcher Manager**.
 
 ---
 
+## Roadmap (Future)
+
+Deferred work — parked but worth tracking.  Pairs with seismo-relay's
+[Roadmap (Future)](https://gitea.serversdown.net/serversdown/seismo-relay#roadmap-future)
+where the corresponding server-side work lives.
+
+- [ ] **File archive manager.**  Move BW autocall-home events older than ~90 days into `<watch_folder>_archive/<year>/<month>/` subfolders so the active watch directory doesn't accumulate hundreds of thousands of entries (filesystem dir lookups slow at 100K+, BW UI hangs opening the folder, watcher's own scandir gets expensive).  Plan drafted in the codec-RE branch's plan-mode session, including a critical pre-coding test (does Blastware UI walk subfolders or only see the flat watch folder?) that determines the archive layout (in-place subfolders vs sibling archive).  Default-off, dry-run mode, opt-in per machine.
+- [ ] **MLG forwarding.**  Currently the watcher's `is_event_binary()` filter explicitly excludes `.MLG` per-unit monitor log files — only event binaries (`.AB0` / `.PG0H` / etc.) and their paired `_ASCII.TXT` reports get forwarded.  Adding an `POST /db/import/mlg_file` SFM endpoint + a parallel `.MLG` scan path on the watcher would populate `monitor_log` rows for non-ACH-routed units (coverage queries, "was this unit monitoring on date X" lookups).  MLG files are append-only / mutable so the watcher needs a different dedup strategy than the per-event sha256 state file — better to forward whole file every scan and let the server dedup by `(serial, start_time)` on insert.
+- [ ] **Pre-deploy seed-state UX in the Settings dialog.**  Currently `event_forwarder.py --seed-state` is a CLI-only operation.  A "Skip backfill" button next to the SFM Forward checkbox would let operators opt-out of re-forwarding the historical archive without dropping to a command line.
+
+---
+
 ## Versioning
 
-Follows **Semantic Versioning**. Current release: **v1.5.1**.
+Follows **Semantic Versioning**. Current release: **v1.5.0**.
 See `CHANGELOG.md` for full history.
 
 ---

config-template.ini

@@ -63,3 +63,12 @@ SFM_HTTP_TIMEOUT = 60
 # to default to <log dir>/sfm_forwarded.json.
 SFM_STATE_FILE =
 
+# Per-pass cap — forward at most N events per scan tick.  0 = unlimited.
+# Default 500 throttles first-deploy backfill on machines with large
+# historical archives (tens or hundreds of thousands of events) so
+# the SFM server isn't hammered with one giant burst.  At 60s scan
+# interval × 500 events/pass that's 30K events/hour throughput.
+# See README "First-time deployment" for the recommended
+# `--seed-state` workflow that skips the historical backfill entirely.
+SFM_MAX_FORWARDS_PER_PASS = 500
+

event_forwarder.py

@@ -96,9 +96,61 @@ def is_event_binary(path: str) -> bool:
     return True
 
 
+def ach_report_name(binary_name: str) -> str:
+    """BW ACH report-naming convention.
+
+    Blastware's official Auto Call Home server writes per-event ASCII
+    reports as ``<stem>_<ext>_ASCII.TXT`` — the ``.`` between stem and
+    ext is replaced with ``_`` and ``_ASCII.TXT`` is appended.
+
+    Examples:
+        ``M529LK44.AB0``  →  ``M529LK44_AB0_ASCII.TXT``
+        ``N844L20G.630H`` →  ``N844L20G_630H_ASCII.TXT``
+        ``H907L1R7.PG0H`` →  ``H907L1R7_PG0H_ASCII.TXT``
+
+    For a filename without a dot (defensive — shouldn't happen for real
+    BW events) we still append ``_ASCII.TXT``.
+    """
+    stem, dot, ext = binary_name.rpartition(".")
+    if not dot:
+        return binary_name + "_ASCII.TXT"
+    return stem + "_" + ext + "_ASCII.TXT"
+
+
+def legacy_report_name(binary_name: str) -> str:
+    """Manual-export convention: ``<binary>.TXT`` (e.g. when an operator
+    saves an event report to text directly from BW's UI rather than
+    letting ACH auto-export it).  Kept as a fallback so the codec-agent
+    test fixtures (``decode-re/5-8-26/event-c/M529LK44.AB0.TXT``) still
+    pair correctly."""
+    return binary_name + ".TXT"
+
+
 def report_path_for(binary_path: str) -> str:
-    """Return the conventional `<binary>.TXT` partner path."""
-    return binary_path + ".TXT"
+    """Legacy entry point — returns the manual-export path.  Prefer
+    :func:`ach_report_name` for new BW deployments.  Retained for
+    backward compatibility with any caller still on the old convention."""
+    return legacy_report_name(binary_path)
+
+
+def is_histogram_event(filename: str) -> bool:
+    """True if the filename's extension marks the file as a Full Histogram
+    event (BW filename scheme: 4-char extensions of the form ``AB0T`` where
+    ``T = H``).  Old-firmware events use 3-char extensions where waveform-vs-
+    histogram is not encoded in the name; we can't tell those apart and
+    return False (the conservative answer — we don't want to suppress
+    "no report" warnings on potentially-waveform old-firmware events).
+
+    Used purely for log clarity — when a forward goes through without a
+    paired TXT, the log distinguishes "histogram, no report expected"
+    (acceptable: BW may not have written one even though it normally
+    does for ACH-routed histograms) from "no report ⚠" on a waveform
+    (more suspicious: BW almost always writes the TXT for waveform events).
+    Forwarding logic itself doesn't depend on this check.
+    """
+    name = os.path.basename(filename)
+    ext = os.path.splitext(name)[1].lstrip(".").upper()
+    return len(ext) == 4 and ext.endswith("H")
 
 
 # ── State file ────────────────────────────────────────────────────────────────
@@ -166,11 +218,53 @@ class ForwardState:
     def is_forwarded(self, sha256: str) -> bool:
         return sha256 in self._data["forwarded"]
 
-    def mark_forwarded(self, sha256: str, filename: str, size: int) -> None:
+    def status(self, sha256: str) -> Optional[bool]:
+        """Return forwarding status for *sha256*.
+
+        Returns:
+            None  — never forwarded.  Eligible for a fresh forward.
+            True  — forwarded successfully with its paired report
+                    (or in a legacy entry that pre-dates the
+                    had_report field — assumed complete for safety).
+                    NOT a candidate for re-forward.
+            False — forwarded WITHOUT its paired ``_ASCII.TXT``
+                    (BW's TXT-write lagged past the grace period).
+                    Eligible for re-forward IF the TXT now exists,
+                    so the SFM server's upsert path can refresh the
+                    DB row with the report's authoritative values.
+
+        Legacy state-file entries without a ``had_report`` key default
+        to ``True`` so an upgrade doesn't unexpectedly re-forward
+        every entry the operator has accumulated.
+        """
+        entry = self._data["forwarded"].get(sha256)
+        if entry is None:
+            return None
+        return bool(entry.get("had_report", True))
+
+    def mark_forwarded(
+        self,
+        sha256: str,
+        filename: str,
+        size: int,
+        had_report: bool = True,
+    ) -> None:
+        """Record a successful forward.
+
+        Set ``had_report=False`` when the forward shipped the binary
+        without its paired ASCII report.  Such entries are re-checked
+        on subsequent scans and re-forwarded once the TXT appears, so
+        SFM's upsert refreshes the DB row with the device-authoritative
+        peak/project values.
+
+        Idempotent: re-marking an existing sha256 with ``had_report=True``
+        is the explicit promotion path used when a re-pair succeeds.
+        """
         self._data["forwarded"][sha256] = {
             "filename":     filename,
             "size":         size,
             "forwarded_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
+            "had_report":   had_report,
         }
         self._save()
 
@@ -209,6 +303,7 @@ def find_pending_events(
     max_age_days: int,
     quiescence_seconds: float = DEFAULT_QUIESCENCE_SECONDS,
     missing_report_grace_seconds: float = DEFAULT_MISSING_REPORT_GRACE_SECONDS,
+    max_per_pass: int = 0,
 ) -> List[Tuple[str, Optional[str]]]:
     """
     Walk `watch_dir` and return the list of (binary_path, txt_path_or_None)
@@ -226,6 +321,11 @@ def find_pending_events(
         missing_report_grace_seconds, we forward without the TXT.
         Younger binaries with a missing TXT are deferred — let BW
         finish writing the report.
+      - When `max_per_pass > 0`, return at most that many pairs.
+        Older files (lower mtime) are forwarded first so backfill
+        proceeds chronologically.  Use this to drip-feed a folder
+        with thousands of qualifying events instead of hammering
+        the SFM server with one giant burst.
     """
     if not os.path.isdir(watch_dir):
         log.warning("forward scan: watch dir not found: %s", watch_dir)
@@ -248,6 +348,20 @@ def find_pending_events(
     # Cache existence of TXT partners so we don't stat() each twice.
     names = {e.name for e in entries if e.is_file()}
 
+    # Sort by mtime ASCENDING so chronological backfill happens oldest-first.
+    # When max_per_pass clamps the list, we always advance — we don't get
+    # stuck re-considering the same N newest files every scan.
+    def _mtime(entry: os.DirEntry) -> float:
+        try:
+            return entry.stat().st_mtime
+        except OSError:
+            return 0.0
+
+    entries = sorted(
+        (e for e in entries if e.is_file()),
+        key=_mtime,
+    )
+
     for e in entries:
         if not e.is_file():
             continue
@@ -268,21 +382,45 @@ def find_pending_events(
             continue
 
         # Idempotency: skip if we already forwarded this content
+        # successfully.  Three cases via state.status(digest):
+        #   True  — forwarded WITH report → permanently done, skip.
+        #   False — forwarded WITHOUT report → re-pair candidate.
+        #           Forward again only if a paired TXT is now present
+        #           so SFM's upsert refreshes the DB row.
+        #   None  — never forwarded → normal first-forward path.
         try:
             digest = sha256_of_file(e.path)
         except OSError as exc:
             log.warning("forward scan: sha256 failed for %s: %s", e.path, exc)
             continue
-        if state.is_forwarded(digest):
+        fwd_status = state.status(digest)
+        if fwd_status is True:
             skipped_already_forwarded += 1
             continue
 
-        # TXT pairing
-        txt_name = e.name + ".TXT"
-        # Case-insensitive match on the .TXT suffix
-        if txt_name not in names:
-            txt_name_lc = txt_name.lower()
-            txt_name = next((n for n in names if n.lower() == txt_name_lc), None)
+        # TXT pairing — try BW ACH convention first
+        # (<stem>_<ext>_ASCII.TXT) and fall back to the manual-export
+        # convention (<binary>.TXT).  Both checked case-insensitively
+        # against the cached directory listing.  ACH wins when both
+        # exist — that's the format BW's official ACH server writes.
+        candidates = [ach_report_name(e.name), legacy_report_name(e.name)]
+
+        # Case-insensitive name lookup against the cached set.
+        names_lc_to_actual = None
+        txt_name: Optional[str] = None
+        for cand in candidates:
+            if cand in names:
+                txt_name = cand
+                break
+            # Build lower-case index lazily — most folders have very few
+            # TXT files relative to binaries, so the linear scan only
+            # fires when neither exact-case candidate matches.
+            if names_lc_to_actual is None:
+                names_lc_to_actual = {n.lower(): n for n in names}
+            actual = names_lc_to_actual.get(cand.lower())
+            if actual:
+                txt_name = actual
+                break
 
         txt_path: Optional[str] = None
         if txt_name:
@@ -291,9 +429,19 @@ def find_pending_events(
                 txt_path = candidate
             # else: TXT is mid-write; treat as not-yet-paired and defer.
 
-        if txt_path is None:
-            # No TXT (or not yet quiescent).  Wait for the grace
-            # period before forwarding alone.
+        if fwd_status is False:
+            # Previously forwarded WITHOUT report.  We're here looking
+            # for a re-pair opportunity.  If the TXT is now present
+            # and quiescent, include in pending for re-forward (the
+            # SFM server's upsert will refresh the DB row with the
+            # report's authoritative values).  Otherwise skip — no
+            # point re-forwarding the same binary alone again.
+            if txt_path is None:
+                skipped_already_forwarded += 1
+                continue
+        elif txt_path is None:
+            # First-time forward and TXT not yet present.  Wait for the
+            # grace period before forwarding alone.
             if (now_ts - mtime) < missing_report_grace_seconds:
                 skipped_inflight += 1
                 continue
@@ -302,9 +450,13 @@ def find_pending_events(
         # Stash size + digest on the tuple-replacement for use during forward;
         # callers can re-derive but caching avoids a second sha256.
 
+        # Per-pass cap: once we have enough pending, stop scanning.
+        if max_per_pass and len(pending) >= max_per_pass:
+            break
+
     log.debug(
-        "forward scan: %d pending  skipped_inflight=%d  already_forwarded=%d",
-        len(pending), skipped_inflight, skipped_already_forwarded,
+        "forward scan: %d pending  skipped_inflight=%d  already_forwarded=%d  cap=%d",
+        len(pending), skipped_inflight, skipped_already_forwarded, max_per_pass,
     )
     return pending
 
@@ -442,6 +594,7 @@ def forward_pending(
     quiescence_seconds: float = DEFAULT_QUIESCENCE_SECONDS,
     missing_report_grace_seconds: float = DEFAULT_MISSING_REPORT_GRACE_SECONDS,
     timeout: float = DEFAULT_HTTP_TIMEOUT,
+    max_per_pass: int = 0,
     logger: Optional[Any] = None,
 ) -> Dict[str, int]:
     """
@@ -467,6 +620,7 @@ def forward_pending(
         max_age_days=max_age_days,
         quiescence_seconds=quiescence_seconds,
         missing_report_grace_seconds=missing_report_grace_seconds,
+        max_per_pass=max_per_pass,
     )
 
     counts = {"scanned": len(pending), "forwarded": 0, "errors": 0, "with_report": 0}
@@ -480,19 +634,45 @@ def forward_pending(
             try:
                 digest = sha256_of_file(binary_path)
                 size   = os.path.getsize(binary_path)
-                state.mark_forwarded(digest, os.path.basename(binary_path), size)
+                # Record whether this forward shipped a paired TXT.
+                # Forwards without a TXT are flagged had_report=False so
+                # subsequent scans re-check whether the TXT has since
+                # appeared and trigger a re-forward (the SFM server's
+                # upsert path refreshes the DB row with the report's
+                # authoritative values).
+                state.mark_forwarded(
+                    digest,
+                    os.path.basename(binary_path),
+                    size,
+                    had_report=(txt_path is not None),
+                )
             except OSError as exc:
                 _log(f"[forward] post-success state save failed for "
                      f"{os.path.basename(binary_path)}: {exc}")
             counts["forwarded"] += 1
             if txt_path:
                 counts["with_report"] += 1
+
+            # Differentiate three cases in the log so "no report" is only
+            # noisy when something's actually unexpected:
+            #   - waveform + TXT          → "+ <txt> attached"
+            #   - waveform without TXT    → "no report ⚠"  (BW maybe didn't auto-export)
+            #   - histogram (any flavour) → "(histogram, no report expected)"
+            if txt_path:
+                report_token = "+ {} attached".format(os.path.basename(txt_path))
+            elif is_histogram_event(binary_path):
+                report_token = "(histogram, no report expected)"
+            else:
+                report_token = "no report ⚠"
+
             _log(
-                f"[forward] OK  {os.path.basename(binary_path)}  "
-                f"({result.get('filesize', 0)}B, "
-                f"{'with' if txt_path else 'no'} report, "
-                f"inserted={result.get('inserted', 0)}, "
-                f"skipped={result.get('skipped', 0)})"
+                "[forward] OK  {}  ({}B, {}, inserted={}, skipped={})".format(
+                    os.path.basename(binary_path),
+                    result.get("filesize", 0),
+                    report_token,
+                    result.get("inserted", 0),
+                    result.get("skipped", 0),
+                )
             )
         else:
             counts["errors"] += 1
@@ -502,3 +682,152 @@ def forward_pending(
             )
 
     return counts
+
+
+# ── Seed-state mode (skip historical backfill on first deploy) ────────────────
+
+
+def seed_state_from_folder(
+    watch_dir: str,
+    state: ForwardState,
+    *,
+    max_age_days: int = 365,
+    logger: Optional[Any] = None,
+) -> Dict[str, int]:
+    """Walk `watch_dir` and mark every existing event binary as already
+    forwarded — without POSTing anything.
+
+    This is the right tool for a first deploy on a machine that already
+    has tens or hundreds of thousands of historical events in the BW
+    ACH folder.  Run it ONCE before enabling SFM_FORWARD_ENABLED:
+
+        python event_forwarder.py --seed-state \
+            --watch "C:\\Blastware 10\\Event\\autocall home" \
+            --state "C:\\...\\sfm_forwarded.json" \
+            [--max-age-days 365]
+
+    The watcher then starts forwarding only events that appear AFTER
+    the seed run.  Files older than `max_age_days` are still skipped
+    by the regular scan loop — we don't bother seeding them because
+    they wouldn't be forwarded anyway.
+
+    Returns a counts dict:
+        {"scanned": int, "seeded": int, "already_known": int, "skipped_too_old": int}
+    """
+    def _log(msg: str) -> None:
+        if logger:
+            logger(msg)
+        else:
+            log.info(msg)
+
+    counts = {"scanned": 0, "seeded": 0, "already_known": 0, "skipped_too_old": 0}
+
+    if not os.path.isdir(watch_dir):
+        _log(f"[seed] watch dir not found: {watch_dir}")
+        return counts
+
+    now_ts = time.time()
+    max_age_seconds = max(1, int(max_age_days)) * 86400.0
+
+    try:
+        with os.scandir(watch_dir) as it:
+            entries = [e for e in it if e.is_file()]
+    except OSError as exc:
+        _log(f"[seed] scandir failed on {watch_dir}: {exc}")
+        return counts
+
+    for e in entries:
+        if not is_event_binary(e.path):
+            continue
+        counts["scanned"] += 1
+        try:
+            mtime = e.stat().st_mtime
+            size  = e.stat().st_size
+        except OSError:
+            continue
+        if (now_ts - mtime) > max_age_seconds:
+            counts["skipped_too_old"] += 1
+            continue
+        try:
+            digest = sha256_of_file(e.path)
+        except OSError as exc:
+            _log(f"[seed] sha256 failed for {e.path}: {exc}")
+            continue
+        if state.is_forwarded(digest):
+            counts["already_known"] += 1
+            continue
+        state.mark_forwarded(digest, e.name, size)
+        counts["seeded"] += 1
+        if counts["seeded"] % 1000 == 0:
+            _log(f"[seed] progress: {counts['seeded']} seeded so far...")
+
+    _log(
+        f"[seed] done.  scanned={counts['scanned']}  seeded={counts['seeded']}  "
+        f"already_known={counts['already_known']}  "
+        f"skipped_too_old={counts['skipped_too_old']}"
+    )
+    return counts
+
+
+# ── CLI entry point ─────────────────────────────────────────────────────────
+
+
+def _main() -> int:
+    """Command-line interface for one-shot operations.
+
+    Currently supports a single mode:
+
+        python event_forwarder.py --seed-state \
+            --watch "<path/to/BW autocall folder>" \
+            --state "<path/to/sfm_forwarded.json>" \
+            [--max-age-days 365]
+
+    which marks every existing in-window event binary as already
+    forwarded (without POSTing) so the watcher only forwards events
+    appearing AFTER the seed.
+    """
+    import argparse
+    parser = argparse.ArgumentParser(
+        description="Series 3 Watcher — SFM event forwarder utilities",
+    )
+    parser.add_argument(
+        "--seed-state", action="store_true",
+        help="Mark every event binary in --watch as already-forwarded "
+             "(without POSTing).  Use this BEFORE enabling SFM_FORWARD "
+             "on a machine with a large historical archive.",
+    )
+    parser.add_argument(
+        "--watch", required=True,
+        help="Path to the Blastware ACH folder.",
+    )
+    parser.add_argument(
+        "--state", required=True,
+        help="Path to the JSON state file.  Will be created if missing.",
+    )
+    parser.add_argument(
+        "--max-age-days", type=int, default=365,
+        help="Only seed files newer than this many days (default 365).",
+    )
+    args = parser.parse_args()
+
+    if not args.seed_state:
+        parser.error("specify --seed-state (no other modes supported yet)")
+
+    print(f"[seed] watch_dir = {args.watch}")
+    print(f"[seed] state     = {args.state}")
+    print(f"[seed] max_age   = {args.max_age_days} days")
+
+    state = ForwardState(args.state)
+    print(f"[seed] state currently has {state.count()} entries")
+    seed_state_from_folder(
+        args.watch, state,
+        max_age_days=args.max_age_days,
+        logger=lambda m: print(m),
+    )
+    print(f"[seed] state now has {state.count()} entries")
+    return 0
+
+
+if __name__ == "__main__":
+    import sys
+    sys.exit(_main())

installer.iss

@@ -3,7 +3,7 @@
 
 [Setup]
 AppName=Series 3 Watcher
-AppVersion=1.5.1
+AppVersion=1.5.0
 AppPublisher=Terra-Mechanics Inc.
 DefaultDirName={pf}\Series3Watcher
 DefaultGroupName=Series 3 Watcher

series3_tray.py

@@ -1,5 +1,5 @@
 """
-Series 3 Watcher — System Tray Launcher  v1.5.1
+Series 3 Watcher — System Tray Launcher  v1.5.0
 Requires: pystray, Pillow, tkinter (stdlib)
 
 Run with:  pythonw series3_tray.py   (no console window)

series3_watcher.py

@@ -104,6 +104,12 @@ def load_config(path: str) -> Dict[str, Any]:
         # State file for forwarded-sha256 idempotency tracking.
         # Defaults next to the log file for easy operator access.
         "SFM_STATE_FILE":              get_str("SFM_STATE_FILE", ""),
+        # Per-pass cap — forward at most N events per scan tick.
+        # 0 = unlimited.  Default 500 as a safety against accidentally
+        # backfilling tens of thousands of events in one burst on
+        # first deploy in a folder that's been accumulating for years.
+        # See README "First-time deployment" section.
+        "SFM_MAX_FORWARDS_PER_PASS":   get_int("SFM_MAX_FORWARDS_PER_PASS", 500),
     }
 
 
@@ -241,7 +247,7 @@ def scan_latest(
 
 
 # --- API heartbeat / SFM telemetry helpers ---
-VERSION = "1.5.1"
+VERSION = "1.5.0"
 
 
 def _read_log_tail(log_file: str, n: int = 25) -> Optional[list]:
@@ -522,6 +528,7 @@ def run_watcher(state: Dict[str, Any], stop_event: threading.Event) -> None:
                                 cfg.get("SFM_MISSING_REPORT_GRACE_SECONDS", 60)
                             ),
                             timeout=int(cfg.get("SFM_HTTP_TIMEOUT", 60)),
+                            max_per_pass=int(cfg.get("SFM_MAX_FORWARDS_PER_PASS", 500)),
                             logger=lambda m: log_message(LOG_FILE, ENABLE_LOGGING, m),
                         )
                         last_forward_ts = now_ts

settings_dialog.py

@@ -1,5 +1,5 @@
 """
-Series 3 Watcher — Settings Dialog  v1.5.1
+Series 3 Watcher — Settings Dialog  v1.5.0
 
 Provides a Tkinter settings dialog that doubles as a first-run wizard.
 
@@ -52,6 +52,7 @@ DEFAULTS = {
     "SFM_MISSING_REPORT_GRACE_SECONDS": "60",
     "SFM_HTTP_TIMEOUT":                 "60",
     "SFM_STATE_FILE":                   "",
+    "SFM_MAX_FORWARDS_PER_PASS":        "500",
 }
 
 
@@ -257,6 +258,7 @@ class SettingsDialog:
         self.var_sfm_missing_report_grace    = tk.StringVar(value=v["SFM_MISSING_REPORT_GRACE_SECONDS"])
         self.var_sfm_http_timeout            = tk.StringVar(value=v["SFM_HTTP_TIMEOUT"])
         self.var_sfm_state_file              = tk.StringVar(value=v["SFM_STATE_FILE"])
+        self.var_sfm_max_per_pass            = tk.StringVar(value=v["SFM_MAX_FORWARDS_PER_PASS"])
 
     # --- UI construction ---
 
@@ -522,15 +524,16 @@ class SettingsDialog:
         self._sfm_test_status.grid(row=0, column=2, padx=(6, 0))
 
         _add_label_spinbox(f, 2, "Forward Interval (sec)",     self.var_sfm_forward_interval,     5, 3600)
-        _add_label_spinbox(f, 3, "Quiescence (sec)",           self.var_sfm_quiescence,           1,   60)
-        _add_label_spinbox(f, 4, "Missing-Report Grace (sec)", self.var_sfm_missing_report_grace, 0,  600)
-        _add_label_spinbox(f, 5, "HTTP Timeout (sec)",         self.var_sfm_http_timeout,         5,  600)
+        _add_label_spinbox(f, 3, "Max Events Per Pass",        self.var_sfm_max_per_pass,         0, 100000)
+        _add_label_spinbox(f, 4, "Quiescence (sec)",           self.var_sfm_quiescence,           1,   60)
+        _add_label_spinbox(f, 5, "Missing-Report Grace (sec)", self.var_sfm_missing_report_grace, 0,  600)
+        _add_label_spinbox(f, 6, "HTTP Timeout (sec)",         self.var_sfm_http_timeout,         5,  600)
 
         tk.Label(f, text="State File", anchor="w").grid(
-            row=6, column=0, sticky="w", padx=(8, 4), pady=4
+            row=7, column=0, sticky="w", padx=(8, 4), pady=4
         )
         state_frame = tk.Frame(f)
-        state_frame.grid(row=6, column=1, sticky="ew", padx=(0, 8), pady=4)
+        state_frame.grid(row=7, column=1, sticky="ew", padx=(0, 8), pady=4)
         state_frame.columnconfigure(0, weight=1)
 
         state_entry = ttk.Entry(state_frame, textvariable=self.var_sfm_state_file, width=32)
@@ -560,7 +563,7 @@ class SettingsDialog:
             "to default to <log dir>/sfm_forwarded.json."
         )
         tk.Label(f, text=hint_text, justify="left", fg="#555555", wraplength=380).grid(
-            row=7, column=0, columnspan=2, sticky="w", padx=(8, 8), pady=(8, 4)
+            row=8, column=0, columnspan=2, sticky="w", padx=(8, 8), pady=(8, 4)
         )
 
     def _test_sfm_connection(self):
@@ -637,6 +640,7 @@ class SettingsDialog:
             (self.var_sfm_quiescence,              "SFM Quiescence",                1,   60,   5),
             (self.var_sfm_missing_report_grace,    "SFM Missing-Report Grace",      0,  600,  60),
             (self.var_sfm_http_timeout,            "SFM HTTP Timeout",              5,  600,  60),
+            (self.var_sfm_max_per_pass,            "SFM Max Events Per Pass",       0, 100000, 500),
         ]
         int_values = {}
         for var, name, mn, mx, dflt in checks:
@@ -693,6 +697,7 @@ class SettingsDialog:
             "SFM_MISSING_REPORT_GRACE_SECONDS": str(int_values["SFM Missing-Report Grace"]),
             "SFM_HTTP_TIMEOUT":                 str(int_values["SFM HTTP Timeout"]),
             "SFM_STATE_FILE":                   self.var_sfm_state_file.get().strip(),
+            "SFM_MAX_FORWARDS_PER_PASS":        str(int_values["SFM Max Events Per Pass"]),
         }
 
         try:

test_event_forwarder.py

@@ -50,6 +50,33 @@ class TestIsEventBinary(unittest.TestCase):
                      "something.h5", "noise.json"]:
             self.assertFalse(ef.is_event_binary(name), name)
 
+    def test_ach_report_name(self):
+        """BW ACH convention: <stem>.<ext> → <stem>_<ext>_ASCII.TXT"""
+        cases = [
+            ("M529LK44.AB0",    "M529LK44_AB0_ASCII.TXT"),
+            ("N844L20G.630H",   "N844L20G_630H_ASCII.TXT"),
+            ("I145L64P.GD0W",   "I145L64P_GD0W_ASCII.TXT"),
+            ("H907L1R7.PG0H",   "H907L1R7_PG0H_ASCII.TXT"),
+        ]
+        for binary, expected in cases:
+            self.assertEqual(ef.ach_report_name(binary), expected, binary)
+
+    def test_legacy_report_name(self):
+        """Manual-export convention: <binary>.TXT"""
+        self.assertEqual(ef.legacy_report_name("M529LK44.AB0"),
+                         "M529LK44.AB0.TXT")
+
+    def test_is_histogram_event(self):
+        # 4-char extension ending in H = histogram
+        for name in ["H907L1R7.PG0H", "S353L4H0.8S0H", "P036L318.C80H"]:
+            self.assertTrue(ef.is_histogram_event(name), name)
+        # 4-char extension ending in W = waveform
+        for name in ["S353L4H0.3M0W", "M529LKVQ.6S0W", "P036L318.C80W"]:
+            self.assertFalse(ef.is_histogram_event(name), name)
+        # 3-char old-firmware extensions can't be classified — return False
+        for name in ["M529LK44.AB0", "M529LIY6.N00", "M529LJ8V.490"]:
+            self.assertFalse(ef.is_histogram_event(name), name)
+
     def test_rejects_non_matching_filenames(self):
         for name in ["", "no_extension",
                      "TooShort.AB0",            # stem must be 8 chars
@@ -128,6 +155,86 @@ class TestFindPendingEvents(unittest.TestCase):
             self.assertEqual(os.path.basename(pending[0][0]), "M529LK44.AB0")
             self.assertEqual(os.path.basename(pending[0][1]), "M529LK44.AB0.TXT")
 
+    def test_pairs_with_ach_underscore_ascii_naming(self):
+        """BW ACH writes M529LK44.AB0 + M529LK44_AB0_ASCII.TXT.  The
+        watcher must pair these even though the .TXT filename doesn't
+        carry a literal copy of the binary's name."""
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_p = Path(tmp)
+            self._make(tmp_p, "N844L20G.630H", age_seconds=120, content=b"binary")
+            self._make(tmp_p, "N844L20G_630H_ASCII.TXT", age_seconds=100, content=b"report")
+
+            state = ef.ForwardState(str(tmp_p / "fwd.json"))
+            pending = ef.find_pending_events(
+                str(tmp_p), state,
+                max_age_days=30,
+                quiescence_seconds=5,
+                missing_report_grace_seconds=60,
+            )
+            self.assertEqual(len(pending), 1)
+            self.assertEqual(os.path.basename(pending[0][0]), "N844L20G.630H")
+            self.assertEqual(os.path.basename(pending[0][1]),
+                             "N844L20G_630H_ASCII.TXT")
+
+    def test_pairs_with_ach_underscore_ascii_naming_for_waveform(self):
+        """Same as above but for new-firmware waveform events
+        (extension ends in W)."""
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_p = Path(tmp)
+            self._make(tmp_p, "I145L64P.GD0W", age_seconds=120, content=b"binary")
+            self._make(tmp_p, "I145L64P_GD0W_ASCII.TXT", age_seconds=100, content=b"report")
+
+            state = ef.ForwardState(str(tmp_p / "fwd.json"))
+            pending = ef.find_pending_events(
+                str(tmp_p), state,
+                max_age_days=30,
+                quiescence_seconds=5,
+                missing_report_grace_seconds=60,
+            )
+            self.assertEqual(len(pending), 1)
+            self.assertEqual(os.path.basename(pending[0][1]),
+                             "I145L64P_GD0W_ASCII.TXT")
+
+    def test_pairing_prefers_ach_naming_when_both_exist(self):
+        """If a folder has BOTH conventions (operator manually exported
+        AND ACH also auto-exported), ACH wins because that's the
+        canonical name in modern BW deployments."""
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_p = Path(tmp)
+            self._make(tmp_p, "M529LK44.AB0", age_seconds=120, content=b"binary")
+            # Both partner files present
+            self._make(tmp_p, "M529LK44.AB0.TXT",      age_seconds=100, content=b"manual")
+            self._make(tmp_p, "M529LK44_AB0_ASCII.TXT", age_seconds=100, content=b"ach")
+
+            state = ef.ForwardState(str(tmp_p / "fwd.json"))
+            pending = ef.find_pending_events(
+                str(tmp_p), state,
+                max_age_days=30,
+                quiescence_seconds=5,
+                missing_report_grace_seconds=60,
+            )
+            self.assertEqual(len(pending), 1)
+            self.assertEqual(os.path.basename(pending[0][1]),
+                             "M529LK44_AB0_ASCII.TXT")
+
+    def test_pairing_falls_back_to_dot_txt_when_ach_absent(self):
+        """If only the manual-export filename exists, the legacy
+        convention still works (preserves codec-agent test fixtures)."""
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_p = Path(tmp)
+            self._make(tmp_p, "M529LK44.AB0",     age_seconds=120, content=b"binary")
+            self._make(tmp_p, "M529LK44.AB0.TXT", age_seconds=100, content=b"manual")
+
+            state = ef.ForwardState(str(tmp_p / "fwd.json"))
+            pending = ef.find_pending_events(
+                str(tmp_p), state,
+                max_age_days=30,
+                quiescence_seconds=5,
+                missing_report_grace_seconds=60,
+            )
+            self.assertEqual(len(pending), 1)
+            self.assertEqual(os.path.basename(pending[0][1]), "M529LK44.AB0.TXT")
+
     def test_skips_if_already_forwarded(self):
         with tempfile.TemporaryDirectory() as tmp:
             tmp_p = Path(tmp)
@@ -178,6 +285,95 @@ class TestFindPendingEvents(unittest.TestCase):
             self.assertEqual(os.path.basename(bin_path), "M529LK44.AB0")
             self.assertIsNone(txt_path)
 
+    def test_re_pair_after_late_arriving_txt(self):
+        """If we forwarded the binary alone (TXT was late) and the TXT
+        later appears, the binary becomes eligible for re-forward."""
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_p = Path(tmp)
+            bin_p = self._make(tmp_p, "M529LK44.AB0",
+                               age_seconds=200, content=b"binary")
+            # Mark as already-forwarded WITHOUT a paired report (the
+            # state we'd be in after a TXT-too-late forward).
+            state = ef.ForwardState(str(tmp_p / "fwd.json"))
+            digest = ef.sha256_of_file(str(bin_p))
+            state.mark_forwarded(digest, "M529LK44.AB0", len(b"binary"),
+                                 had_report=False)
+
+            # First scan: TXT not present yet → still skipped.
+            pending = ef.find_pending_events(
+                str(tmp_p), state, max_age_days=30,
+                quiescence_seconds=5, missing_report_grace_seconds=60,
+            )
+            self.assertEqual(pending, [],
+                             "no TXT present → no re-pair attempt")
+
+            # Now BW finally writes the TXT.
+            self._make(tmp_p, "M529LK44.AB0.TXT",
+                       age_seconds=100, content=b"report")
+            pending = ef.find_pending_events(
+                str(tmp_p), state, max_age_days=30,
+                quiescence_seconds=5, missing_report_grace_seconds=60,
+            )
+            self.assertEqual(len(pending), 1,
+                             "TXT now present → re-pair attempt expected")
+            self.assertEqual(os.path.basename(pending[0][0]), "M529LK44.AB0")
+            self.assertEqual(os.path.basename(pending[0][1]), "M529LK44.AB0.TXT")
+
+    def test_re_pair_not_attempted_when_already_had_report(self):
+        """Successful WITH-report forwards stay permanently skipped.
+        Adding more files later does NOT trigger a re-forward."""
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_p = Path(tmp)
+            bin_p = self._make(tmp_p, "M529LK44.AB0", age_seconds=200, content=b"x")
+            self._make(tmp_p, "M529LK44.AB0.TXT", age_seconds=100, content=b"r")
+            state = ef.ForwardState(str(tmp_p / "fwd.json"))
+            state.mark_forwarded(ef.sha256_of_file(str(bin_p)),
+                                 "M529LK44.AB0", 1, had_report=True)
+            pending = ef.find_pending_events(
+                str(tmp_p), state, max_age_days=30,
+                quiescence_seconds=5, missing_report_grace_seconds=60,
+            )
+            self.assertEqual(pending, [],
+                             "had_report=True forwards stay skipped")
+
+    def test_legacy_state_entries_default_to_had_report_true(self):
+        """Backward compat: state-file entries from before the
+        had_report field existed are treated as fully forwarded so
+        an upgrade doesn't re-forward every entry."""
+        import json
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_p = Path(tmp)
+            path = str(tmp_p / "fwd.json")
+            with open(path, "w") as f:
+                json.dump({
+                    "version": 1,
+                    "forwarded": {
+                        "abc123": {
+                            "filename":     "M529LK01.AB0",
+                            "size":         123,
+                            "forwarded_at": "2025-01-01T00:00:00Z",
+                            # No had_report field — legacy entry
+                        }
+                    }
+                }, f)
+            state = ef.ForwardState(path)
+            self.assertIs(state.status("abc123"), True,
+                          "legacy entry must default to 'fully forwarded'")
+
+    def test_state_status_returns_none_for_unknown_sha(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            state = ef.ForwardState(str(Path(tmp) / "fwd.json"))
+            self.assertIs(state.status("never-seen"), None)
+
+    def test_state_mark_with_had_report_false(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            state = ef.ForwardState(str(Path(tmp) / "fwd.json"))
+            state.mark_forwarded("xyz", "f.AB0", 100, had_report=False)
+            self.assertIs(state.status("xyz"), False)
+            # Subsequent re-mark with had_report=True promotes to done.
+            state.mark_forwarded("xyz", "f.AB0", 100, had_report=True)
+            self.assertIs(state.status("xyz"), True)
+
     def test_defers_when_txt_missing_and_within_grace(self):
         with tempfile.TemporaryDirectory() as tmp:
             tmp_p = Path(tmp)
@@ -223,6 +419,154 @@ class TestFindPendingEvents(unittest.TestCase):
             )
             self.assertEqual(len(pending), 0)
 
+    def test_max_per_pass_caps_returned_count(self):
+        """When max_per_pass is set, return at most that many pairs."""
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_p = Path(tmp)
+            # Create 5 distinct event binaries with paired .TXTs
+            for i, name in enumerate(
+                ["M529LK01.AB0", "M529LK02.AB0", "M529LK03.AB0",
+                 "M529LK04.AB0", "M529LK05.AB0"],
+            ):
+                self._make(tmp_p, name, age_seconds=120 + i,
+                           content=("bin-" + str(i)).encode())
+                self._make(tmp_p, name + ".TXT", age_seconds=110 + i,
+                           content=b"report")
+
+            state = ef.ForwardState(str(tmp_p / "fwd.json"))
+            pending = ef.find_pending_events(
+                str(tmp_p), state,
+                max_age_days=30,
+                quiescence_seconds=5,
+                missing_report_grace_seconds=60,
+                max_per_pass=2,
+            )
+            self.assertEqual(len(pending), 2)
+
+    def test_max_per_pass_zero_means_unlimited(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_p = Path(tmp)
+            for i in range(4):
+                self._make(tmp_p, "M529LK0{}.AB0".format(i),
+                           age_seconds=120 + i,
+                           content=("bin-" + str(i)).encode())
+            state = ef.ForwardState(str(tmp_p / "fwd.json"))
+            pending = ef.find_pending_events(
+                str(tmp_p), state,
+                max_age_days=30,
+                quiescence_seconds=5,
+                missing_report_grace_seconds=60,
+                max_per_pass=0,
+            )
+            self.assertEqual(len(pending), 4)
+
+    def test_max_per_pass_returns_oldest_first(self):
+        """Backfill should advance chronologically — oldest qualifying
+        files first.  This way successive scans always make progress
+        instead of getting stuck re-considering the same N newest files."""
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_p = Path(tmp)
+            # ages: 200s (oldest), 150s, 100s, 50s (skipped — within grace)
+            ages = [200, 150, 100, 50]
+            for i, age in enumerate(ages):
+                self._make(tmp_p, "M529LK0{}.AB0".format(i),
+                           age_seconds=age, content=("c" + str(i)).encode())
+                self._make(tmp_p, "M529LK0{}.AB0.TXT".format(i),
+                           age_seconds=age - 10, content=b"r")
+
+            state = ef.ForwardState(str(tmp_p / "fwd.json"))
+            pending = ef.find_pending_events(
+                str(tmp_p), state,
+                max_age_days=30, quiescence_seconds=5,
+                missing_report_grace_seconds=60, max_per_pass=2,
+            )
+            # Oldest two should be M529LK00 (200s) and M529LK01 (150s)
+            names = [os.path.basename(p[0]) for p in pending]
+            self.assertEqual(names, ["M529LK00.AB0", "M529LK01.AB0"])
+
+
+# ── Seed-state mode ──────────────────────────────────────────────────────────
+
+
+class TestSeedStateFromFolder(unittest.TestCase):
+
+    def _make(self, dir_path: Path, name: str, age_seconds: float = 100,
+              content: bytes = b"x") -> Path:
+        p = dir_path / name
+        p.write_bytes(content)
+        target = time.time() - age_seconds
+        os.utime(p, (target, target))
+        return p
+
+    def test_seeds_every_in_window_event_without_posting(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_p = Path(tmp)
+            for i in range(3):
+                self._make(tmp_p, "M529LK0{}.AB0".format(i),
+                           age_seconds=120 + i, content=("e" + str(i)).encode())
+            # Plus a non-event file we should ignore
+            self._make(tmp_p, "BE11529.MLG", age_seconds=120, content=b"mlg")
+
+            state = ef.ForwardState(str(tmp_p / "seed.json"))
+            counts = ef.seed_state_from_folder(
+                str(tmp_p), state, max_age_days=30,
+            )
+            self.assertEqual(counts["scanned"], 3)
+            self.assertEqual(counts["seeded"], 3)
+            self.assertEqual(counts["already_known"], 0)
+            self.assertEqual(state.count(), 3)
+
+    def test_seed_skips_files_beyond_max_age_days(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_p = Path(tmp)
+            self._make(tmp_p, "M529LK01.AB0", age_seconds=120, content=b"new")
+            self._make(tmp_p, "M529LK02.AB0", age_seconds=10 * 86400,
+                       content=b"in-window")    # 10d < 30d cutoff
+            self._make(tmp_p, "M529LK03.AB0", age_seconds=400 * 86400,
+                       content=b"way-old")      # 400d > 30d cutoff
+
+            state = ef.ForwardState(str(tmp_p / "seed.json"))
+            counts = ef.seed_state_from_folder(
+                str(tmp_p), state, max_age_days=30,
+            )
+            self.assertEqual(counts["seeded"], 2)
+            self.assertEqual(counts["skipped_too_old"], 1)
+
+    def test_seeded_files_are_then_skipped_by_normal_scan(self):
+        """End-to-end: seed once, then a normal scan should produce
+        zero pending events for the seeded files."""
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_p = Path(tmp)
+            self._make(tmp_p, "M529LK01.AB0", age_seconds=120, content=b"x")
+            self._make(tmp_p, "M529LK01.AB0.TXT", age_seconds=110, content=b"r")
+            self._make(tmp_p, "M529LK02.AB0", age_seconds=120, content=b"y")
+            self._make(tmp_p, "M529LK02.AB0.TXT", age_seconds=110, content=b"r")
+
+            state = ef.ForwardState(str(tmp_p / "seed.json"))
+            ef.seed_state_from_folder(str(tmp_p), state, max_age_days=30)
+
+            pending = ef.find_pending_events(
+                str(tmp_p), state,
+                max_age_days=30, quiescence_seconds=5,
+                missing_report_grace_seconds=60,
+            )
+            self.assertEqual(len(pending), 0,
+                             "seed should have marked everything already-forwarded")
+
+    def test_seed_is_idempotent(self):
+        """Re-running seed twice doesn't duplicate entries or POST anything."""
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_p = Path(tmp)
+            self._make(tmp_p, "M529LK01.AB0", age_seconds=120, content=b"x")
+
+            state = ef.ForwardState(str(tmp_p / "seed.json"))
+            counts1 = ef.seed_state_from_folder(str(tmp_p), state, max_age_days=30)
+            counts2 = ef.seed_state_from_folder(str(tmp_p), state, max_age_days=30)
+            self.assertEqual(counts1["seeded"], 1)
+            self.assertEqual(counts2["seeded"], 0)
+            self.assertEqual(counts2["already_known"], 1)
+            self.assertEqual(state.count(), 1)
+
 
 # ── Multipart encoder ────────────────────────────────────────────────────────