FilesystemMailService._seen grows unboundedly and the poll loop rescans every historical inbox entry each 0.5s tick

[huangzesen]

Summary

FilesystemMailService._seen (src/lingtai_kernel/services/mail.py:116) accumulates one entry per message ever received and is never pruned, even after the email intrinsic archives or deletes the message and its inbox directory is gone. In addition, the Phase 1 poll loop re-iterates the entire mailbox/inbox/ directory every 500 ms, so a long-lived agent with high mail volume pays a slowly growing memory cost plus an O(inbox size) filesystem scan on every tick. This issue proposes pruning _seen against the set of directories actually present in the inbox on each completed scan, plus an optional mtime gate to skip scans when the inbox has not changed.

Narrative

LingTai agents are designed to be long-lived. An agent process starts its FilesystemMailService, calls listen(), and from then on a daemon thread wakes every half second to look for new mail. To avoid re-delivering messages, the service keeps an in-memory dedupe set:

# src/lingtai_kernel/services/mail.py:116
self._seen: set[str] = set()

listen() seeds it with a snapshot of existing inbox entries (services/mail.py:223-227), Phase 1 adds every newly dispatched message id (services/mail.py:248), and Phase 2 pre-marks ids claimed from pseudo-agent outboxes (services/mail.py:335). The only removals in the whole file are the two rollback paths in _poll_pseudo_outbox (services/mail.py:352 and services/mail.py:365), which fire only when a claim fails mid-flight. On the happy path, nothing is ever removed.

Meanwhile the message lifecycle on disk does not match. The email intrinsic treats mailbox/inbox/ as the durable unread/read store, and when the agent archives a message, EmailManager._archive (src/lingtai_kernel/intrinsics/email/manager.py:716-736) physically moves the directory out of the inbox with shutil.move(str(src), str(dst)) into mailbox/archive/. Deletion removes it entirely. So over weeks of operation, an agent that dutifully archives its mail ends up with a small inbox on disk — but _seen still holds the id of every message it has ever received. Each entry is a 20-character string (_new_mailbox_id produces <YYYYMMDDTHHMMSS>-<4 hex>, src/lingtai_kernel/intrinsics/email/primitives.py:22-26), so this is not a fast leak, but it is unbounded, and "unbounded growth in a daemon designed to run forever" is exactly the kind of defect that surfaces as mysterious RSS creep months later.

The second half of the problem hits agents that do not archive aggressively. Phase 1 of the poll loop (services/mail.py:234-248) does this every tick:

if self._inbox_dir.is_dir():
    for entry in self._inbox_dir.iterdir():
        if not entry.is_dir():
            continue
        if entry.name in self._seen:
            continue
        ...
self._poll_stop.wait(0.5)

Every historical inbox entry — including messages read long ago that were never archived — is re-yielded by iterdir() and re-checked against _seen twice per second, forever. With a few thousand accumulated inbox entries this is a few thousand readdir results plus is_dir() stats per tick, per agent, multiplied across every agent on the machine. It is pure waste: the poll loop only cares about new directories.

The current design got here honestly. The dedupe set is the simplest correct mechanism for "notify once per new directory," and the inbox has to keep its directories in place because the email intrinsic (_list_inbox, read/unread tracking, archive) reads them as the durable mailbox. The alternative used on the pseudo-outbox path — atomically renaming claimed messages into sent/ so the polled directory stays small — is not available here, because moving a dispatched message out of inbox/ would break the user-visible mailbox semantics.

Better looks like: after every complete Phase 1 scan, intersect _seen with the set of directory names actually observed. A pruned id cannot cause a duplicate dispatch, because Phase 1 only dispatches directories that exist — if the directory is gone (archived/deleted), there is nothing to re-deliver; and mailbox ids are timestamp-prefixed, so an old id essentially never reappears as a new message. This makes _seen size track the live inbox instead of all history. Optionally, a directory-mtime gate can skip the iterdir() entirely on the (vast majority of) ticks where the inbox has not changed, cutting steady-state scan cost to a single stat.

Current behavior

• FilesystemMailService.__init__ creates self._seen: set[str] = set() (src/lingtai_kernel/services/mail.py:116).
• listen() seeds _seen from existing inbox entries (services/mail.py:223-227), then starts _poll_loop which runs Phase 1 (own inbox scan, services/mail.py:234-248) and Phase 2 (pseudo-outbox claims, services/mail.py:252-253) every 0.5 s (services/mail.py:256).
• Additions to _seen: Phase 1 after dispatch (services/mail.py:248), Phase 2 pre-mark before placing the claimed copy (services/mail.py:335).
• Removals from _seen: only the Phase 2 rollback paths (services/mail.py:352, services/mail.py:365). There is no pruning anywhere; stop() (services/mail.py:464-469) does not clear it either.
• The email intrinsic moves archived entries out of inbox/ via shutil.move (src/lingtai_kernel/intrinsics/email/manager.py:716-736), so _seen retains ids whose directories no longer exist under the polled path.
• Phase 1 iterates every inbox entry on every tick regardless of whether anything changed.

Detailed implementation plan

All changes are in src/lingtai_kernel/services/mail.py, class FilesystemMailService.

1. Prune _seen after each complete Phase 1 scan. Restructure the Phase 1 block inside _poll_loop to collect the names observed during iteration and intersect at the end:

   # Phase 1 — own inbox.
   if self._inbox_dir.is_dir():
       existing: set[str] = set()
       for entry in self._inbox_dir.iterdir():
           if not entry.is_dir():
               continue
           existing.add(entry.name)
           if entry.name in self._seen:
               continue
           msg_file = entry / "message.json"
           if msg_file.is_file():
               try:
                   payload = json.loads(msg_file.read_text(encoding="utf-8"))
                   on_message(payload)
               except (json.JSONDecodeError, OSError):
                   pass
               self._seen.add(entry.name)
       # Prune ids whose inbox dirs are gone (archived/deleted).
       # Safe: a missing dir can never be re-dispatched by this loop,
       # and mailbox ids are timestamp-prefixed so they don't recur.
       self._seen &= existing

   Correctness notes for the implementer:

   • Prune only on a complete scan. The self._seen &= existing line must stay inside the existing try: so that an OSError raised mid-iterdir() skips pruning (a partial existing set would wrongly evict ids for directories that still exist, causing duplicate dispatch on the next tick). The snippet above already has this property because the exception jumps to the except OSError handler at services/mail.py:254 before the intersection runs.
   • No race with Phase 2. _poll_loop is single-threaded: Phase 2's pre-mark at services/mail.py:335 is followed in the same iteration by creating inbox/<uuid>/ (services/mail.py:339-343) or by an explicit discard on rollback, so by the time the next tick's Phase 1 prune runs, every pre-marked id either has a directory on disk or has already been discarded. Do not move the prune anywhere that could run between Phase 2's pre-mark and its inbox write.
   • Directories present but without message.json yet (in-flight send() with attachments, services/mail.py:184-197) are included in existing but not in _seen — unchanged behavior, and they are not pruned into oblivion.

2. (Optional, second commit) Skip unchanged-inbox scans with an mtime gate. Add a field self._inbox_mtime_ns: int = -1 in __init__, and at the top of Phase 1:

   try:
       mtime_ns = self._inbox_dir.stat().st_mtime_ns
   except OSError:
       mtime_ns = -1
   if mtime_ns != -1 and mtime_ns == self._inbox_mtime_ns and not force_scan:
       skip Phase 1 this tick
   else:
       run the full scan + prune, then self._inbox_mtime_ns = mtime_ns

   POSIX guarantees the parent directory's mtime changes when entries are created, renamed in/out, or removed, which covers send() delivery, Phase 2 self-placement, archive, and delete. To be robust against coarse mtime granularity on some filesystems and against non-local mounts, force a full scan (force_scan) at least once every N ticks (e.g. every 20 ticks = 10 s) and whenever the observed mtime_ns is within ~2 s of the current time. Keep this commit separate so it can be reverted independently if a filesystem misbehaves; step 1 alone fixes the unbounded growth.

3. Back-compat / migration. None needed. _seen is private, in-memory, per-process state; no on-disk schema changes, no config changes. The public MailService ABC is untouched.

4. Documentation. Update the Phase 1 description in the listen() docstring (services/mail.py:216-222) and the services/ANATOMY.md entry for mail.py to mention that _seen is pruned to the live inbox contents each pass.

Risks and alternatives

• Risk: duplicate dispatch after a wrongful prune. The only way pruning re-dispatches a message is if an id is evicted while its directory still exists under inbox/. The plan prevents this by (a) building existing in the same iterdir() pass used for dispatch, (b) pruning only when that pass completes without error, and (c) relying on the single poll thread so Phase 2 pre-marks cannot interleave with the prune. Reviewers should treat any relocation of the &= line as a correctness question, not a style one.
• Risk: mailbox id reuse. _new_mailbox_id (intrinsics/email/primitives.py:22-26) is <UTC timestamp>-<4 hex>; a pruned id could in principle recur only via a same-second collision producing a new message with an old id — in which case dispatching it is the correct behavior anyway.
• Alternative: move dispatched dirs out of inbox/ (mirroring the pseudo-outbox outbox/ → sent/ rename at services/mail.py:324-361). Rejected: inbox/ is the durable store the email intrinsic reads — _list_inbox (intrinsics/email/primitives.py:72-78), read/unread digests, and _archive all assume messages stay in inbox/ until the agent archives or deletes them. Moving on dispatch would change user-visible mailbox semantics and require migrating every consumer.
• Alternative: cap _seen with an LRU / max size. Rejected: evicting an id whose directory still sits in the inbox re-dispatches an old message — a correctness bug traded for a memory bound. Pruning against on-disk reality has no such failure mode.
• Alternative: OS file watchers (inotify / FSEvents / kqueue). Would eliminate polling cost entirely but adds a platform-specific dependency and a very different failure surface; the 0.5 s polling design is deliberate and simple. The mtime gate in step 2 captures most of the benefit with a single stat per tick.
• The scan-cost half remains O(live inbox) for agents that never archive. That is inherent to inbox-as-durable-store; the mtime gate reduces it to O(1) per unchanged tick, and encouraging archive hygiene is out of scope here.

Testing plan

Add to tests/test_services_mail.py (or tests/test_filesystem_mail.py, whichever hosts the poll-loop tests):

1. test_seen_pruned_after_inbox_dir_removed — start listen(), deliver a message via send() from a sibling service, wait for dispatch, assert the id is in svc._seen; then move the inbox entry to a sibling archive/ directory (simulating EmailManager._archive), wait ≥ 2 poll ticks, and assert the id is no longer in svc._seen.
2. test_prune_does_not_redispatch — same setup as above with a counting on_message; after the archive-move and several ticks, assert the handler was called exactly once (the pruned id must not cause a duplicate delivery, and the now-missing dir must not be re-read).
3. test_seen_retains_live_inbox_entries — deliver two messages, archive only one; after several ticks assert _seen contains exactly the surviving entry's id and the handler fired once per message.
4. test_listen_snapshot_then_prune — pre-populate the inbox before listen() (exercising the snapshot at services/mail.py:223-227), remove one pre-existing dir after listening starts, and assert it is pruned while the other snapshot ids remain and were never dispatched.
5. test_pseudo_claim_id_survives_prune — set up a pseudo-agent outbox message addressed to the service, let Phase 2 claim it, and assert across several subsequent ticks that the claimed id stays in _seen (its inbox copy exists) and on_message fired exactly once — guarding the Phase 2 pre-mark interaction described in the plan.
6. test_partial_scan_does_not_prune — monkeypatch Path.iterdir (or the inbox path object) to yield one entry then raise OSError; assert _seen is unchanged after the tick, verifying the prune-only-on-complete-scan invariant.
7. If step 2 (mtime gate) lands: test_mtime_gate_detects_new_delivery — deliver a message while the gate is active and assert it is dispatched within the forced-rescan window; and test_mtime_gate_skips_unchanged_inbox — assert via an instrumented iterdir wrapper that unchanged ticks skip the directory iteration.

Generated by a Claude Fable 5 deep review of the lingtai-kernel codebase.