A 方案落地：观察式复现单 ＋ 采样器（第一组对照已拿到，且修正上一轮假设——事件速率不是充分因，嫌疑收窄到大文件写入）

mem_watch.py

@@ -0,0 +1,65 @@
+# -*- coding: utf-8 -*-
+"""
+临时采样器 · daemon 内存 ＋ 事件计数（会话活跃期观察）
+用途：复现「会话活跃 ⇒ 事件洪峰 ⇒ daemon 内存暴涨」是否成立（A 方案）
+只读 daemon.log，不碰任何进程；输出追加到 tmp/mem-watch-<日期>.csv
+"""
+import re
+import time
+import os
+import datetime
+
+LOG = r"E:/ProgramData/.workbuddy/logs/daemon.log"
+OUT = r"E:/ProgramData/AIProject/contentm_agent/tmp/mem-watch-20261009.csv"
+DURATION = 300      # 总时长（秒）
+INTERVAL = 20       # 采样间隔（秒）
+
+MEM_RE = re.compile(r"\[DaemonMemWatch\] pid=(\d+) heapUsed=(\d+)MB heapTotal=(\d+)MB rss=(\d+)MB")
+EV_RE = re.compile(r'Conversation event push summary",\{"received":(\d+),"immediateSent":(\d+)')
+
+
+def tail_text(path, nbytes=500000):
+    size = os.path.getsize(path)
+    with open(path, "rb") as f:
+        if size > nbytes:
+            f.seek(size - nbytes)
+        return f.read().decode("utf-8", "ignore")
+
+
+def main():
+    d = os.path.dirname(OUT)
+    if d and not os.path.exists(d):
+        os.makedirs(d, exist_ok=True)
+    with open(OUT, "a", encoding="utf-8") as fo:
+        fo.write("# sampling start %s\n" % datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
+        fo.flush()
+        end = time.time() + DURATION
+        while time.time() < end:
+            lines = tail_text(LOG).splitlines()
+            mem = ev = None
+            for ln in reversed(lines):
+                if mem is None and "DaemonMemWatch" in ln:
+                    m = MEM_RE.search(ln)
+                    if m:
+                        mem = m.groups()
+                if ev is None and "Conversation event push summary" in ln:
+                    m = EV_RE.search(ln)
+                    if m:
+                        ev = m.groups()
+                if mem and ev:
+                    break
+            now = datetime.datetime.now().strftime("%H:%M:%S")
+            if mem and ev:
+                pid, hu, ht, rss = mem
+                rec, sent = ev
+                fo.write("%s,pid=%s,heap=%sMB,heapTotal=%sMB,rss=%sMB,received=%s,immediateSent=%s\n"
+                         % (now, pid, hu, ht, rss, rec, sent))
+            else:
+                fo.write("%s,(miss)\n" % now)
+            fo.flush()
+            time.sleep(INTERVAL)
+        fo.write("# sampling end %s\n" % datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
+
+
+if __name__ == "__main__":
+    main()