feat: improve cross-session memory quality

CHANGELOG.md

@@ -12,12 +12,15 @@ All notable changes to this project will be documented in this file.
 - 跨客户端共享 Skill 清单、诊断和安全同步
 - 通用 `dna-memory-loop` Skill
 - CI 测试、编译和公开敏感信息扫描
+- 跨会话精确去重：同类型且规范化摘要相同的 active 记忆只保留一条，并合并来源与客户端
+- daily 维护的 `deduplicated` 指标，区分新认知结晶与重复候选
 
 ### Changed
 - README、快速上手和客户端文档改为当前跨端用法
 - 本机 profile、运行配置和记忆数据全部迁出 Git 仓库
 - macOS 自动化标识和示例路径改为通用命名
 - 旧 Claude 同步脚本不再包含固定用户或项目路径
+- 自动提取过滤操作交接、计划性叙述和无主体发布话术，减少过程话术结晶
 
 ### Removed
 - 私人 Skill、个人部署计划、真实 vault 名称和运行时记忆样例

README.md

@@ -194,6 +194,11 @@ python3 dna.py memory maintain monthly --json
 不能解释为待审记忆数量。`clients` 中的 `recall_hits` 和 `recall_share` 用于
 观察 Codex、Claude、Hermes 的主动召回采用情况；占比不代表召回质量。
 
+daily 维护结果中的 `crystallized` 表示新建长期认知，`deduplicated` 表示候选
+已命中同类型、同规范化摘要的 active 记忆。重复候选不会创建第二份 Markdown；
+其新的来源和客户端会合并到既有记忆的 provenance 字段中。系统只做精确文本去重，
+不使用模糊相似度自动合并不同结论。
+
 当新证据明确使旧结论失效时，调用 `memory_remember` 并显式传入旧 ID：
 
 ```json

README_EN.md

@@ -164,6 +164,13 @@ alias for `total_pending`; it is not the number of memories awaiting review.
 Per-client `recall_hits` and `recall_share` show active-recall adoption across
 Codex, Claude, and Hermes. Recall share is not a quality score.
 
+In daily maintenance results, `crystallized` counts newly created durable
+memories and `deduplicated` counts candidates that matched an active memory
+with the same type and normalized summary. Duplicate candidates do not create
+another Markdown page; their new sources and clients are merged into the
+existing provenance fields. Deduplication is exact after normalization, not a
+fuzzy similarity merge.
+
 When a new verified fact invalidates an older one, pass exact old memory IDs in
 `supersedes`. Old Markdown remains available as history, while default recall
 returns active conclusions only. DNA Memory never infers replacement from type

scripts/memory_operations.py

@@ -20,7 +20,7 @@ def daily(self, now=None):
         queue = CandidateEventQueue(self.config.database_path, self.config.max_candidate_events)
         result = {
             "crystallized": 0, "rejected": 0, "compacted": 0,
-            "expired": 0, "deleted": 0,
+            "deduplicated": 0, "expired": 0, "deleted": 0,
         }
         try:
             proposals = queue.connection.execute("""
@@ -52,12 +52,13 @@ def daily(self, now=None):
                         "clients": [client], "project_path": project_path,
                         "session_id": session_id,
                     })
+                    status = "deduplicated" if remembered.get("deduplicated") else "crystallized"
                     queue.connection.execute(
-                        "UPDATE candidate_events SET status='crystallized', processed_at=?, memory_id=?, error=NULL WHERE event_id=?",
-                        (now, remembered["memory_id"], event_id),
+                        "UPDATE candidate_events SET status=?, processed_at=?, memory_id=?, error=NULL WHERE event_id=?",
+                        (status, now, remembered["memory_id"], event_id),
                     )
                     queue.connection.commit()
-                    result["crystallized"] += 1
+                    result[status] += 1
             finally:
                 service.store.close()
 
@@ -85,7 +86,7 @@ def daily(self, now=None):
             result["expired"] = cursor.rowcount
             cursor = queue.connection.execute("""
                 DELETE FROM candidate_events
-                WHERE status IN ('crystallized', 'rejected', 'superseded', 'expired')
+                WHERE status IN ('crystallized', 'deduplicated', 'rejected', 'superseded', 'expired')
                   AND datetime(COALESCE(processed_at, created_at)) < datetime(?, '-7 days')
             """, (now,))
             result["deleted"] = cursor.rowcount

scripts/memory_service.py

@@ -8,6 +8,7 @@
 import json
 import re
 import shutil
+import unicodedata
 from pathlib import Path
 
 import yaml
@@ -40,10 +41,23 @@ def remember(self, proposal):
         source_hash = proposal.get("source_hash")
         if source_hash:
             row = self.store.connection.execute(
-                "SELECT memory_id FROM memory_index WHERE source_hash=?", (source_hash,)
+                "SELECT memory_id FROM memory_index "
+                "WHERE source_hash=? AND status='active' AND source_kind='markdown'",
+                (source_hash,),
             ).fetchone()
             if row:
-                return {"created": False, "memory_id": row[0], "superseded": []}
+                return {
+                    "created": False, "deduplicated": True,
+                    "memory_id": row[0], "superseded": [],
+                }
+        normalized_summary = self._normalize_summary(summary)
+        duplicate = self._find_active_duplicate(mem_type, normalized_summary)
+        if duplicate:
+            self._merge_duplicate_provenance(duplicate, proposal)
+            return {
+                "created": False, "deduplicated": True,
+                "memory_id": duplicate["memory_id"], "superseded": [],
+            }
         supersedes = self._normalize_supersedes(proposal.get("supersedes"))
         root = self.config.knowledge_root / self.config.managed_memory_dir
         targets = self._load_active_targets(supersedes, root)
@@ -80,6 +94,55 @@ def remember(self, proposal):
             "superseded": supersedes,
         }
 
+    @staticmethod
+    def _normalize_summary(summary):
+        """Normalize only exact textual duplicates; never use fuzzy similarity."""
+        normalized = unicodedata.normalize("NFKC", str(summary or ""))
+        return re.sub(r"\s+", " ", normalized).strip().casefold()
+
+    def _find_active_duplicate(self, mem_type, normalized_summary):
+        rows = self.store.connection.execute(
+            "SELECT memory_id, markdown_path, summary FROM memory_index "
+            "WHERE source_kind='markdown' AND status='active' AND type=?",
+            (mem_type,),
+        ).fetchall()
+        for row in rows:
+            if self._normalize_summary(row["summary"]) == normalized_summary:
+                return row
+        return None
+
+    @staticmethod
+    def _merge_unique(existing, incoming):
+        values = []
+        for collection in (existing, incoming):
+            if not isinstance(collection, list):
+                continue
+            for value in collection:
+                if value not in values:
+                    values.append(value)
+        return values
+
+    def _merge_duplicate_provenance(self, duplicate, proposal):
+        root = self.config.knowledge_root / self.config.managed_memory_dir
+        targets = self._load_active_targets([duplicate["memory_id"]], root)
+        target = targets[0]
+        meta = dict(target["meta"])
+        meta["source_refs"] = self._merge_unique(
+            meta.get("source_refs", []), proposal.get("source_refs", [])
+        )
+        meta["clients"] = self._merge_unique(
+            meta.get("clients", []), proposal.get("clients", [])
+        )
+        now = time.strftime("%Y-%m-%dT%H:%M:%S%z")
+        meta["updated"] = now
+        if proposal.get("project_path") and not meta.get("project_path"):
+            meta["project_path"] = proposal["project_path"]
+        if proposal.get("session_id") and not meta.get("session_id"):
+            meta["session_id"] = proposal["session_id"]
+        self._install_memory_files({
+            target["path"]: self._render_memory(meta, target["body"]),
+        })
+
     @staticmethod
     def _normalize_supersedes(value):
         if value is None:

scripts/native_auto_extract.py

@@ -100,6 +100,19 @@
     r"^(?:但)?你可以直接执行下面",
     re.IGNORECASE,
 )
+_ACTION_HANDOFF = re.compile(
+    r"^(?:请你先|告诉我(?:已|已经)|复制到剪切板|复制到剪贴板|"
+    r"再继续(?:插入|执行)|等你(?:确认|回复)|等待你)",
+    re.IGNORECASE,
+)
+_PLANNING_NARRATION = re.compile(
+    r"^(?:我会把|我将把|接下来我(?:将|会)|正在(?:把|将|整理|制作|进行))",
+    re.IGNORECASE,
+)
+_UNSUBSTANTIVE_COMPLETION = re.compile(
+    r"^(?:好了[，,]?\s*)?公开版本已完成发布[：:]?$",
+    re.IGNORECASE,
+)
 _GENERIC_COMPLETION = re.compile(
     r"^(?:Markdown\s+格式转换|(?:所有|全部).{0,30}(?:文档|文件)|"
     r"(?:不过)?其他.{0,20}功能|好的[，,]?\s*子agent).{0,50}"
@@ -290,6 +303,8 @@ def extract_automatic_proposals(messages, max_proposals=DEFAULT_MAX_PROPOSALS):
                     or _IN_PROGRESS_NARRATION.search(text) or _COMMAND_PREFACE.search(text)
                     or _GENERIC_COMPLETION.search(text) or _RESOURCE_STATUS.search(text)
                     or _UNVERIFIED_PROMOTIONAL.search(text)
+                    or _ACTION_HANDOFF.search(text) or _PLANNING_NARRATION.search(text)
+                    or _UNSUBSTANTIVE_COMPLETION.search(text)
                     or not inspect_content(text).allowed):
                 continue
             if message.get("role") == "assistant" and _VAGUE_ASSISTANT.match(text):

tests/test_memory_cli.py

@@ -90,7 +90,7 @@ def test_memory_maintain_daily_outputs_bounded_json(tmp_path, capsys):
     payload = json.loads(capsys.readouterr().out)
 
     assert payload == {
-        "compacted": 0, "crystallized": 0, "deleted": 0,
+        "compacted": 0, "crystallized": 0, "deduplicated": 0, "deleted": 0,
         "expired": 0, "rejected": 0,
     }
 

tests/test_memory_operations.py

@@ -38,6 +38,7 @@ def test_only_reviewed_safe_proposals_crystallize(tmp_path):
     result = MemoryOperations(config).daily(now="2026-07-11 12:00:00")
 
     assert result["crystallized"] == 1
+    assert result["deduplicated"] == 0
     assert result["rejected"] == 0
     rows = queue.connection.execute(
         "SELECT event_id, status, memory_id FROM candidate_events ORDER BY event_id"
@@ -90,6 +91,7 @@ def test_daily_rejects_invalid_then_crystallizes_valid_without_locking(tmp_path)
 
     assert result["rejected"] == 1
     assert result["crystallized"] == 1
+    assert result["deduplicated"] == 0
     rows = dict(queue.connection.execute(
         "SELECT event_id, status FROM candidate_events ORDER BY event_id"
     ).fetchall())
@@ -99,6 +101,30 @@ def test_daily_rejects_invalid_then_crystallizes_valid_without_locking(tmp_path)
     assert "同批次的有效记忆" in pages[0].read_text()
 
 
+def test_daily_marks_cross_session_duplicate_as_deduplicated(tmp_path):
+    config = _profile(tmp_path)
+    queue = CandidateEventQueue(config.database_path)
+    for event_id, client, source_ref in (
+        ("proposal-a", "codex", "codex://session/a"),
+        ("proposal-b", "hermes", "hermes://session/b"),
+    ):
+        queue.enqueue({
+            "event_id": event_id, "client": client,
+            "event_type": "memory_proposal", "memory_type": "fact",
+            "excerpt": "metaver.vip 当前返回 200",
+            "source_ref": source_ref, "source_hash": "source-" + event_id,
+        })
+
+    result = MemoryOperations(config).daily(now="2026-07-12 12:00:00")
+
+    assert result["crystallized"] == 1
+    assert result["deduplicated"] == 1
+    rows = dict(queue.connection.execute(
+        "SELECT event_id, status FROM candidate_events ORDER BY event_id"
+    ).fetchall())
+    assert rows == {"proposal-a": "crystallized", "proposal-b": "deduplicated"}
+
+
 def test_retention_expires_pointers_and_deletes_terminal_events(tmp_path):
     config = _profile(tmp_path)
     queue = CandidateEventQueue(config.database_path)

tests/test_memory_service.py

@@ -37,10 +37,52 @@ def test_source_hash_is_idempotent(tmp_path):
     first = svc.remember(proposal)
     second = svc.remember(proposal)
     assert second == {
-        "created": False, "memory_id": first["memory_id"], "superseded": [],
+        "created": False, "deduplicated": True,
+        "memory_id": first["memory_id"], "superseded": [],
     }
 
 
+def test_normalized_active_summary_deduplicates_and_merges_provenance(tmp_path):
+    svc = service(tmp_path)
+    first = svc.remember({
+        "type": "fact",
+        "summary": "飞书表格入口已可直接打开",
+        "source_refs": ["codex://session/one"],
+        "clients": ["codex"],
+    })
+
+    second = svc.remember({
+        "type": "fact",
+        "summary": "  飞书表格入口已可直接打开  ",
+        "source_refs": ["hermes://session/two"],
+        "clients": ["hermes"],
+    })
+
+    assert second == {
+        "created": False, "deduplicated": True,
+        "memory_id": first["memory_id"], "superseded": [],
+    }
+    record = svc.get(first["memory_id"])
+    assert record["source_refs"] == ["codex://session/one", "hermes://session/two"]
+    assert record["clients"] == ["codex", "hermes"]
+    assert len(list((tmp_path / "vault/00 System/Memory").glob("*.md"))) == 1
+
+
+def test_deduplication_is_scoped_to_type_and_active_status(tmp_path):
+    svc = service(tmp_path)
+    fact = svc.remember({"type": "fact", "summary": "same scoped conclusion"})
+    decision = svc.remember({"type": "decision", "summary": "same scoped conclusion"})
+
+    assert decision["created"] is True
+    svc.store.connection.execute(
+        "UPDATE memory_index SET status='superseded' WHERE memory_id=?",
+        (fact["memory_id"],),
+    )
+    svc.store.connection.commit()
+    replacement = svc.remember({"type": "fact", "summary": "same scoped conclusion"})
+    assert replacement["created"] is True
+
+
 def test_sensitive_proposal_is_rejected(tmp_path):
     svc = service(tmp_path)
     with pytest.raises(MemoryValidationError):
@@ -170,7 +212,8 @@ def test_source_hash_idempotency_does_not_repeat_supersede(tmp_path):
 
     assert first["superseded"] == [old["memory_id"]]
     assert second == {
-        "created": False, "memory_id": first["memory_id"], "superseded": [],
+        "created": False, "deduplicated": True,
+        "memory_id": first["memory_id"], "superseded": [],
     }
     assert old_path.read_text() == after_first
 

tests/test_native_auto_extract.py

@@ -321,6 +321,10 @@ def test_type_specific_negative_cases_are_rejected(role, content):
     "若下周一 Kimi WebBridge 仍未恢复，再提交反馈。",
     "Codex 文档已完成初稿，当前正在做最后核验。",
     "rollout_summaries/2026-01-01-example.md:10-12|note=[verified image blocker]",
+    "请你先把 Gemini Pro 会员的截图复制到剪切板，告诉我已复制，再继续插入。",
+    "我会把 01-06 做成一个完整合集，后面再统一发布。",
+    "接下来我将把这批素材整理成一个完整合集。",
+    "好了，公开版本已完成发布：",
 ])
 def test_rejects_real_backtest_false_positive_shapes(content):
     assert extract_automatic_proposals([