Consolidate duplicated storage helpers across mcp_servers: atomic writes onto lingtai_kernel._fsutil, shared watermark store, message listing, read-state, and conversation preview

[huangzesen]

Summary

The curated messaging MCP servers under src/lingtai/mcp_servers/ hand-roll the same small persistence helpers over and over: a tmp-file + os.replace atomic JSON write appears in at least eight places with four different durability/permission behaviors, two byte-for-byte identical WatermarkStore.save() implementations exist, and _list_messages / _read_ids + _mark_read / _build_conversation_preview (with its embedded _rel_time) are triplicated across telegram/feishu/wechat with subtle drift. The kernel already ships a hardened, audited helper (lingtai_kernel._fsutil.atomic_write_json) that this layer is allowed to import — src/lingtai/mcp_servers/daemon_common/server.py:18 already does — so the duplication is pure debt. This issue proposes shared _storage.py, _watermark.py, and _preview.py modules in src/lingtai/mcp_servers/ (following the existing _skill.py / _identity.py precedent) and migrating the ad-hoc writers onto _fsutil.

Narrative

Every channel addon (Telegram, Feishu, WeChat, WhatsApp, IMAP, Cloud Mail) persists small JSON state files under the agent workdir: contacts, read-message IDs, poll watermarks, per-message message.json records. Because these servers run as independent subprocesses that can be killed at any moment (agent sleep, molt, host restart), every one of these writes wants to be crash-atomic: write a sibling temp file, then os.replace it over the target. That pattern was solved once in the kernel during the issue #510 fsutil migration — src/lingtai_kernel/_fsutil.py has atomic_write_text / atomic_write_json with a carefully documented design (uuid4 sibling temp names to avoid same-process collisions, exclusive-create open, umask-preserving permissions, opt-in fsync, serialize-before-touching-the-file). But the mcp_servers layer largely predates or ignored that work, and each addon re-solved the problem locally.

The result is that today the "same" write behaves differently per channel. The LICC fallback writer in src/lingtai/mcp_servers/_licc_compat.py:101-108 fsyncs before the rename ("Write + fsync + atomic replace so the host poller never sees a half-written file"); none of the other seven writers fsync. Everything based on tempfile.mkstemp (both watermark stores, _identity.py, imap/telegram/feishu _save_contacts, telegram/feishu _mark_read, wechat _atomic_write) silently creates state files with mode 0o600 — the exact permissions side effect the _fsutil docstring calls out at src/lingtai_kernel/_fsutil.py:66-71. WhatsApp took yet another path: src/lingtai/mcp_servers/whatsapp/manager.py:125-129 writes via a fixed-name temp file with no cleanup-on-error:

def _save_contacts(self, alias: str, contacts: dict[str, Any]) -> None:
    path = self._contacts_path(alias)
    tmp = path.with_suffix(".json.tmp")
    tmp.write_text(json.dumps(contacts, ensure_ascii=False, indent=2), encoding="utf-8")
    tmp.replace(path)

A failed write leaves contacts.json.tmp littered forever, and two concurrent writers share one temp path — precisely the collision the _fsutil audit flagged and fixed with uuid4 sibling names.

The watermark duplication is the most stark case. src/lingtai/mcp_servers/imap/_watermark.py:39-59 and src/lingtai/mcp_servers/cloud_mail/_watermark.py:45-63 contain a character-for-character identical save() (mkstemp, os.write, os.close, os.replace, the same nested try/except cleanup); the two WatermarkStore classes differ only in the docstring, a load() type check, and Cloud Mail's convenience properties (last_email_id, seeded). Someone hardening one (say, adding fsync so a power loss cannot roll back the watermark and re-deliver old mail) will forget the other.

The higher-level helpers drifted the same way. Telegram's _list_messages (src/lingtai/mcp_servers/telegram/manager.py:704-720) and Feishu's (src/lingtai/mcp_servers/feishu/manager.py:751-767) are identical except for the sort key (m.get("date", "") vs m.get("date") or m.get("sent_at") or ""); WhatsApp's _iter_messages (whatsapp/manager.py:148-161) does the same job with a glob, a stored_at sort key, and a bare except Exception. _read_ids/_mark_read are copy-pasted between telegram/manager.py:871-898 and feishu/manager.py:877-904, while WeChat keeps read IDs in memory and persists via its own _atomic_write (wechat/manager.py:938-942, 1017-1029). And _build_conversation_preview — a ~100-line function that formats the last N messages of a chat into a guidance-prefixed markdown body with relative timestamps, reply quoting, and a 10,000-char truncation policy — exists three times (telegram/manager.py:738-869, feishu/manager.py:769-875, wechat/manager.py:782-863), each embedding its own _rel_time. The three _rel_times already disagree: telegram/feishu parse only "%Y-%m-%dT%H:%M:%SZ" via strptime, while wechat uses the lenient datetime.fromisoformat plus a naive-tz fallback. A message whose date carries an offset renders as a relative time in WeChat previews but falls through to the raw string in Telegram/Feishu previews. The 10,000-char tail-truncation block is textually identical in all three — a bug fix there means three edits or two missed copies.

How did it get here? Each channel addon was built by copying the previous one (the repo's own ANATOMY.md shows the family resemblance), and the copies were made before lingtai_kernel._fsutil existed. The repo already recognizes the fix pattern: cross-channel concerns were previously factored into src/lingtai/mcp_servers/_skill.py (bundled manuals) and src/lingtai/mcp_servers/_identity.py (identity envelope + write). Storage is simply the next concern in line. Better looks like: one atomic-write implementation (the kernel's), one watermark store, one message-folder reader, one read-state helper, one preview builder parameterized by the few genuinely channel-specific bits (chat matching, sender naming, reply-target resolution) — so the next durability or truncation fix lands everywhere at once, and the next channel addon starts ~200 lines smaller.

Current behavior

Hand-rolled atomic writes (tmp + os.replace), each slightly different:

• src/lingtai/mcp_servers/imap/_watermark.py:39-59 — WatermarkStore.save(): mkstemp + replace, no fsync, mode 0o600.
• src/lingtai/mcp_servers/cloud_mail/_watermark.py:45-63 — WatermarkStore.save(): byte-identical to the IMAP one; class adds last_email_id / seeded / set_last_email_id (lines 65-80).
• src/lingtai/mcp_servers/_identity.py:47-67 — write_identity_file(): mkstemp + os.fdopen + replace, trailing newline, no fsync.
• src/lingtai/mcp_servers/_licc_compat.py:95-109 — _fallback_push_inbox_event: the only writer that fsyncs; uses a predictable <event_id>.tmp sibling name.
• src/lingtai/mcp_servers/imap/manager.py:425-440 — _save_contacts: mkstemp + replace.
• src/lingtai/mcp_servers/telegram/manager.py:880-898 (_mark_read) and 1009-1025 (_save_contacts): two copies of the mkstemp block in one file.
• src/lingtai/mcp_servers/feishu/manager.py:886-904 (_mark_read) and 915-931 (_save_contacts): same two copies again.
• src/lingtai/mcp_servers/wechat/manager.py:1017-1029 — private _atomic_write used by _save_state/_save_contacts/_save_read/_save_seen (lines 922-942, 1005-1015).
• src/lingtai/mcp_servers/whatsapp/manager.py:125-129 — _save_contacts: fixed-name temp, no error cleanup, temp-path collision under concurrency.
• Additional copies of the same mkstemp block live in src/lingtai/mcp_servers/telegram/account.py:1074-1083, src/lingtai/mcp_servers/feishu/account.py:540-549, and src/lingtai/mcp_servers/wechat/login.py:85-93 (session/credential writers).

Triplicated higher-level helpers:

• _list_messages: telegram/manager.py:704-720, feishu/manager.py:751-767; WhatsApp variant _iter_messages at whatsapp/manager.py:148-161; WeChat variant _load_messages_from at wechat/manager.py:873-888.
• _read_ids / _mark_read: telegram/manager.py:871-898, feishu/manager.py:877-904; WeChat in-memory + _save_read at wechat/manager.py:898-902, 938-942.
• _load_contacts / _save_contacts: telegram (1000-1025), feishu (906-931), wechat (892-897, 932-936), whatsapp (121-129), imap (416-440).
• _build_conversation_preview + embedded _rel_time + identical 10,000-char truncation: telegram/manager.py:738-869, feishu/manager.py:769-875, wechat/manager.py:782-863.

Kernel helper already available to this layer: src/lingtai/mcp_servers/daemon_common/server.py:18 imports lingtai_kernel._fsutil.atomic_write_json and uses it at line 125.

Detailed implementation plan

Ship as three small PRs (atomic writes; watermark; listing/read-state/preview) so review stays tractable and behavior changes are bisectable.

Step 1 — migrate all hand-rolled atomic writers to lingtai_kernel._fsutil.
No new code needed for the primitive; atomic_write_json(path, obj, ensure_ascii=..., indent=..., fsync=...) and atomic_write_text already cover every call site (they also mkdir(parents=True) the parent, so callers can drop their own mkdir lines).

1. src/lingtai/mcp_servers/_identity.py::write_identity_file → atomic_write_text(path, json.dumps(payload, indent=2, ensure_ascii=False) + "\n") (preserve the trailing newline; atomic_write_json does not emit one).
2. src/lingtai/mcp_servers/_licc_compat.py::_fallback_push_inbox_event → atomic_write_text(final, text, fsync=True). Keep the two-line comment about the host poller. Note this changes the temp filename from <event_id>.tmp to _fsutil's dotted uuid name — verify the host inbox poller's glob only matches *.json/EVENT_SUFFIX (it does; TMP_SUFFIX files were already skipped) so hidden temp names are equally invisible.
3. imap/manager.py::_save_contacts, telegram/manager.py::_save_contacts/_mark_read, feishu/manager.py::_save_contacts/_mark_read → atomic_write_json(target, contacts, indent=2) / atomic_write_json(target, sorted(ids), indent=None). Preserve each site's current ensure_ascii/indent choices exactly (telegram/feishu/imap contacts currently use json.dumps(..., indent=2) with default ensure_ascii=True; pass ensure_ascii=True there to keep bytes identical).
4. wechat/manager.py: delete _atomic_write and call atomic_write_text from _save_state/_save_contacts/_save_read/_save_seen (they already pre-serialize strings).
5. whatsapp/manager.py::_save_contacts → atomic_write_json(path, contacts, ensure_ascii=False, indent=2). This fixes the temp-file leak and the fixed-temp-name collision for free.
6. Same treatment for telegram/account.py:1074-1083, feishu/account.py:540-549, wechat/login.py:85-93 (credential/session writers — consider fsync=True for these since losing a session file forces re-login).
7. Remove now-unused tempfile imports.

Step 2 — shared watermark store.
Create src/lingtai/mcp_servers/_watermark.py:

from lingtai_kernel._fsutil import atomic_write_json, read_json

class JsonWatermarkStore:
    """Tiny JSON-on-disk store; corrupt/missing files read as {}."""
    def __init__(self, path: Path | str) -> None:
        self._path = Path(path)
    def load(self) -> dict:
        return read_json(self._path, default={}, expect=dict)
    def save(self, state: dict) -> None:
        atomic_write_json(self._path, state, ensure_ascii=False, indent=2)

• imap/_watermark.py becomes class WatermarkStore(JsonWatermarkStore) (or a re-export) keeping its module docstring about the per-folder UIDNEXT shape.
• cloud_mail/_watermark.py becomes a subclass adding only last_email_id, seeded, set_last_email_id.
• Note one intentional unification: IMAP's current load() does not type-check the top level while Cloud Mail's does; read_json(..., expect=dict) gives both the safer behavior.

Step 3 — shared message-folder listing.
Create src/lingtai/mcp_servers/_storage.py with:

def list_messages(account_dir: Path, folder: str, *,
                  sort_key=lambda m: m.get("date", ""),
                  newest_first: bool = True,
                  attach_dir: bool = True) -> list[dict]:
    """Load every <folder>/<msg_dir>/message.json; skip corrupt files."""

• Telegram/Feishu _list_messages become one-line delegations (Feishu passes its date or sent_at sort key). WeChat's _load_messages_from delegates with attach_dir=False. WhatsApp's _iter_messages delegates per folder with its stored_at key and keeps its _folder tagging.
• Also move the read-state pair here:

def load_read_ids(path: Path) -> set[str]: ...
def save_read_ids(path: Path, ids: set[str]) -> None:
    atomic_write_json(path, sorted(ids), indent=None)

Telegram/Feishu _read_ids/_mark_read delegate; WeChat's _save_read calls save_read_ids (this changes its on-disk order from arbitrary list(set) to sorted — readers only build a set, so it is safe and makes the file diff-stable).

Step 4 — shared conversation preview.
Create src/lingtai/mcp_servers/_preview.py:

def rel_time(date_str: str, *, now: datetime | None = None) -> str:
    """fromisoformat-based ('just now' / 'N min ago' / ... ) with raw-string fallback."""

def render_conversation_preview(*, channel: str, messages: list[dict],
    header_template: str, tail: str,
    sender_name: Callable[[dict], str],
    text_of: Callable[[dict], str],
    reply_line: Callable[[dict, dict[str, dict]], str | None] = lambda m, by_id: None,
    max_len: int = 10000) -> str:
    """Format sorted messages as '[rel] #id sender: text' lines, apply reply
    quoting via reply_line, prepend header+tail, tail-truncate to max_len."""

• Each manager keeps a thin _build_conversation_preview that (a) gathers and filters candidate messages for the chat/user (genuinely channel-specific: telegram parses compound IDs, feishu matches chat_id, wechat matches from_user_id/to_user_id), then (b) calls render_conversation_preview with channel-specific sender_name / text_of / reply_line closures. The 10,000-char truncation and line formatting live in exactly one place.
• Standardize on the wechat-style datetime.fromisoformat parser in rel_time (it accepts the %Y-%m-%dT%H:%M:%SZ strings telegram/feishu write on Python ≥3.11 — confirm the repo floor; if 3.10 must be supported, try strptime first then fromisoformat). This is a deliberate, tested unification: offset-bearing dates now render relatively in all channels.

Step 5 — docs. Update src/lingtai/mcp_servers/ANATOMY.md: add _storage.py, _watermark.py, _preview.py rows to the components table and list them in related_files, mirroring how _skill.py/_identity.py are documented.

Back-compat: no file format, path, or schema changes; all state files keep their names and JSON shapes (byte-identical except WeChat read.json ordering and any site where we deliberately keep ensure_ascii as-is). No migration code needed.

Risks and alternatives

• File-permission change (real, accepted): every mkstemp-based writer currently produces mode 0o600 files; _fsutil deliberately preserves the umask (typically 0o644) — see the rationale at src/lingtai_kernel/_fsutil.py:66-71. Contacts, read state, and watermarks are non-secret workdir state, so umask semantics are correct. Exception: the credential/session writers in telegram/account.py, feishu/account.py, and wechat/login.py may be relying on mkstemp's 0o600 for secrets under .secrets/. Audit those three before migrating; if they need 0o600, keep them on a chmod-after-write wrapper or leave them for a follow-up rather than silently loosening secret file modes.
• Temp-name change: leftover *.tmp siblings become .name.pid.uuid.tmp hidden files. Grep for any cleanup/glob code that assumes the old suffix (the LICC TMP_SUFFIX constant is the one known consumer).
• Preview drift is a feature and a risk: unifying rel_time changes output strings for edge-case dates in telegram/feishu previews. These bodies feed LLM notifications, not parsers, so drift is low-risk, but snapshot tests should pin the new behavior. Keep per-channel gather/filter logic out of the shared module so channel semantics (compound-ID parsing, callback_query text fallback, media placeholders) cannot be accidentally homogenized.
• Alternative — put everything in lingtai_kernel._fsutil: wrong layer; message listing, read state, and previews are channel-addon concepts, and _fsutil is intentionally a stdlib-only primitive module. Only the atomic-write primitive belongs in the kernel, and it is already there.
• Alternative — a common BaseChannelManager class: larger refactor with inheritance coupling across six managers that differ structurally (WeChat is single-account with in-memory state; others are per-account-dir). Free functions + tiny store classes match the existing _skill.py/_identity.py precedent and can be adopted incrementally.
• Alternative — do nothing: every durability hardening (fsync on watermarks, tmp cleanup) must be re-applied N times; whatsapp's known temp-leak/collision bug stays.

Testing plan

Add tests/test_mcp_storage_helpers.py:

• test_watermark_roundtrip — JsonWatermarkStore.save then load returns the same dict; parent dirs auto-created.
• test_watermark_corrupt_file_reads_empty — write garbage bytes to the path; load() returns {} (pins the rebootstrap contract from both existing docstrings).
• test_watermark_non_dict_top_level_reads_empty — a JSON list on disk returns {} (pins the unified expect=dict behavior for the former IMAP store).
• test_cloud_mail_watermark_convenience — set_last_email_id(42) → last_email_id == 42, seeded is True (guards the subclass after the base extraction).
• test_save_contacts_no_tmp_litter_on_error — monkeypatch json.dumps/serialization to raise; assert the target is untouched and no *.tmp* sibling remains (this is the WhatsApp regression fix).
• test_list_messages_skips_corrupt_and_sorts — build inbox dirs with one corrupt message.json; assert corrupt entry skipped, order matches sort key, _dir attached when requested.
• test_read_ids_roundtrip_and_sorted_on_disk — save_read_ids writes sorted JSON array; load_read_ids on missing/corrupt file returns empty set.

Add tests/test_mcp_preview_render.py:

• test_rel_time_buckets — fixed now; assert "just now", "N min ago", "N hr ago", "yesterday", absolute-date, and raw-string fallback for unparseable input.
• test_rel_time_accepts_offset_dates — an ISO string with +08:00 renders relatively (the unified behavior).
• test_render_preview_truncation — messages totalling >10,000 chars produce a body ≤10,000 starting with the header and containing the …\n continuation marker before the newest lines (pins the shared truncation block).
• test_render_preview_reply_quoting — a reply_line closure producing ↳ [...] lines appears indented beneath its parent.

Regression coverage via existing suites: tests/test_telegram_notification_read_state.py, tests/test_telegram_rich_formatting.py, tests/test_cloud_mail_addon.py, tests/test_wechat_inbound_replay.py (exercises _save_seen/_record_seen through the new writer), and tests/test_mcp_identity_discovery.py / tests/test_email_identity.py (identity file writes, including the trailing-newline contract). Add one assertion to the identity tests that the written file still ends with "\n" to pin the atomic_write_text migration.

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