[codex] improve bounded native cognition extraction

README.md

@@ -218,6 +218,18 @@ python3 scripts/import_native_history.py
 `DNA_MEMORY_PROPOSAL {JSON}` 或通过有界信号提取的短结论才进入候选队列，
 并且仍需经过 daily 维护的类型、敏感信息、容量和去重检查。
 
+规则升级时先在独立临时数据库回测最近 7 天，再做有界生产重提取。导入器只输出
+聚合计数；自动结果仍进入 `memory_proposal`，不会绕过人工检查和 daily 安全结晶：
+
+```bash
+python3 -m scripts.import_native_history --backtest-days 7 \
+  --backtest-db /tmp/dna-memory-backtest.db
+python3 -m scripts.import_native_history --reextract-days 7
+```
+
+只有人工抽查误采集率不高于 10% 且敏感泄露为 0 时才运行生产重提取。完整操作与
+回滚边界见 [`docs/mcp-and-client-adapters.md`](docs/mcp-and-client-adapters.md)。
+
 这意味着“扫描到了会话”不等于“已经形成长期认知”。衡量系统价值时应同时看：
 
 1. 自动捕获覆盖率。

README_EN.md

@@ -178,6 +178,20 @@ pointers. Explicit `DNA_MEMORY_PROPOSAL {JSON}` markers and bounded signal
 extraction create review candidates; daily maintenance still applies type,
 sensitivity, capacity, and deduplication gates.
 
+After changing native extraction rules, backtest the last seven days in a
+separate empty database before bounded production re-extraction. The importer
+emits aggregate metrics only, and automatic results remain reviewed
+`memory_proposal` events rather than direct long-term memories:
+
+```bash
+python3 -m scripts.import_native_history --backtest-days 7 \
+  --backtest-db /tmp/dna-memory-backtest.db
+python3 -m scripts.import_native_history --reextract-days 7
+```
+
+Proceed only when manual sampling finds at most 10% false positives and zero
+sensitive leakage. Daily maintenance remains the crystallization gate.
+
 Capture is not the same as durable learning. Evaluate both capture coverage and
 the number of verified memories that are later recalled and marked useful.
 

docs/mcp-and-client-adapters.md

@@ -175,6 +175,45 @@ python3 scripts/import_native_history.py
 来源指针由 `max_candidate_events` 限制，默认 10,000 条待处理普通事件。达到上限
 时拒绝新增普通指针，但不阻塞客户端主任务。不要用扩大上限代替维护。
 
+### 多行认知提取的安全回测
+
+升级提取规则后，先把最近 7 天来源写入一个全新的临时数据库。回测只输出文件数、
+错误类别、候选数和记忆类型分布，不修改生产 checkpoint，也不运行 daily：
+
+```bash
+python3 -m scripts.import_native_history \
+  --backtest-days 7 \
+  --backtest-db /tmp/dna-memory-backtest.db
+```
+
+回测数据库必须不存在或为空，且不能是 profile 指向的生产数据库。`errors` 表示解析
+失败；`proposals` 表示识别出的候选数；`proposal_types` 用于检查八类记忆是否异常
+偏斜。需要人工抽查时，只在本机查看临时库，不要把摘要复制到 issue、日志或公开报告：
+
+```bash
+sqlite3 /tmp/dna-memory-backtest.db \
+  "SELECT client, memory_type, excerpt FROM candidate_events WHERE event_type='memory_proposal' ORDER BY event_id;"
+```
+
+把新增候选逐条标为真阳性、误采集或模糊。只有明确误采集比例不高于 10%，且凭证、
+完整工具输出和大段 transcript 泄露为 0 时，才允许生产重提取：
+
+```bash
+python3 -m scripts.import_native_history --reextract-days 7
+python3 -m scripts.import_native_history --reextract-days 7
+```
+
+`--reextract-days` 只让指定时间范围内的文件绕过相同哈希 checkpoint 的快速返回；
+它不删除 checkpoint、事件或原生会话。稳定 event ID 保证第二次运行不会重复入队。
+生产候选仍只是 `memory_proposal`，必须再次人工查看后才可运行：
+
+```bash
+python3 dna.py memory maintain daily --json
+```
+
+如果质量门槛未通过，停止在临时库或 pending proposal 阶段，先把误采集样本脱敏为
+回归测试；不要运行 daily，也不要扩大扫描时间范围。
+
 ## 维护与价值
 
 ```bash

scripts/import_native_history.py

@@ -102,29 +102,67 @@ def prune_obsolete_sources(paths_by_client, queue):
     return {"checkpoints": len(obsolete), "events": events}
 
 
-def import_paths(paths_by_client, queue, max_files=40, max_bytes=64 * 1024, min_age_seconds=120):
-    total = {"files": 0, "processed": 0, "enqueued": 0, "proposals": 0}
+def import_paths(
+    paths_by_client,
+    queue,
+    max_files=40,
+    max_bytes=64 * 1024,
+    min_age_seconds=120,
+    reextract_days=None,
+    now=None,
+):
+    reference_time = time.time() if now is None else float(now)
+    if reextract_days is not None and int(reextract_days) <= 0:
+        raise ValueError("reextract_days must be positive")
+    cutoff = (
+        reference_time - int(reextract_days) * 86400
+        if reextract_days is not None else None
+    )
+    total = {
+        "files": 0,
+        "processed": 0,
+        "enqueued": 0,
+        "proposals": 0,
+        "errors": 0,
+        "proposal_types": {},
+        "error_types": {},
+    }
     for client, paths in source_files(paths_by_client).items():
         paths = sorted(paths, key=lambda p: p.stat().st_mtime, reverse=True)
         pending = []
         for path in paths:
             stat = path.stat()
-            if time.time() - stat.st_mtime < int(min_age_seconds):
+            if reference_time - stat.st_mtime < int(min_age_seconds):
                 continue
             checkpoint = queue.get_checkpoint(
                 "native-auto:{}:{}".format(client, path.expanduser().resolve())
             )
-            if not checkpoint or checkpoint[0] != stat.st_ino or checkpoint[1] != stat.st_size:
-                pending.append(path)
-        for path in pending[:int(max_files)]:
+            force = cutoff is not None and stat.st_mtime >= cutoff
+            if force or not checkpoint or checkpoint[0] != stat.st_ino or checkpoint[1] != stat.st_size:
+                pending.append((path, force))
+        for path, force in pending[:int(max_files)]:
+            total["files"] += 1
             try:
-                result = import_native_file(path, queue, client=client, max_bytes=max_bytes)
-            except (OSError, ValueError):
+                result = import_native_file(
+                    path,
+                    queue,
+                    client=client,
+                    max_bytes=max_bytes,
+                    force=force,
+                )
+            except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as error:
+                total["errors"] += 1
+                suffix = path.suffix.lower().lstrip(".") or "unknown"
+                key = "{}:{}:{}".format(client, suffix, type(error).__name__)
+                total["error_types"][key] = total["error_types"].get(key, 0) + 1
                 continue
-            total["files"] += 1
             total["processed"] += int(bool(result["processed"]))
             total["enqueued"] += result["enqueued"]
             total["proposals"] += result["proposals"]
+            for memory_type, count in result.get("proposal_types", {}).items():
+                total["proposal_types"][memory_type] = (
+                    total["proposal_types"].get(memory_type, 0) + count
+                )
     return total
 
 
@@ -144,17 +182,86 @@ def configured_paths(config):
     }
 
 
+def run_backtest(
+    paths_by_client,
+    database_path,
+    days=7,
+    max_files=40,
+    max_bytes=64 * 1024,
+    min_age_seconds=120,
+    now=None,
+):
+    days = int(days)
+    if days <= 0:
+        raise ValueError("days must be positive")
+    database_path = Path(database_path).expanduser().resolve()
+    if database_path.exists() and database_path.stat().st_size > 0:
+        raise ValueError("backtest database must be new or empty")
+    reference_time = time.time() if now is None else float(now)
+    cutoff = reference_time - days * 86400
+    recent = {
+        client: [
+            path for path in paths
+            if cutoff <= path.stat().st_mtime
+            and reference_time - path.stat().st_mtime >= int(min_age_seconds)
+        ]
+        for client, paths in source_files(paths_by_client).items()
+    }
+    queue = CandidateEventQueue(database_path)
+    try:
+        result = import_paths(
+            recent,
+            queue,
+            max_files=max_files,
+            max_bytes=max_bytes,
+            min_age_seconds=min_age_seconds,
+            now=reference_time,
+        )
+    finally:
+        queue.connection.close()
+    return {"mode": "backtest", **result}
+
+
 def main(argv=None):
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--max-files", type=int, default=int(os.getenv("DNA_MEMORY_MAX_FILES", "40")))
     parser.add_argument("--max-bytes", type=int, default=64 * 1024)
     parser.add_argument("--prune-obsolete", action="store_true")
+    mode = parser.add_mutually_exclusive_group()
+    mode.add_argument("--reextract-days", type=int)
+    mode.add_argument("--backtest-days", type=int)
+    parser.add_argument("--backtest-db", type=Path)
     args = parser.parse_args(argv)
+    if args.reextract_days is not None and args.reextract_days <= 0:
+        parser.error("--reextract-days must be positive")
+    if args.backtest_days is not None and args.backtest_days <= 0:
+        parser.error("--backtest-days must be positive")
+    if bool(args.backtest_days) != bool(args.backtest_db):
+        parser.error("--backtest-days and --backtest-db must be used together")
     config = load_config()
+    paths = configured_paths(config)
+    if args.backtest_days is not None:
+        backtest_path = args.backtest_db.expanduser().resolve()
+        if backtest_path == Path(config.database_path).expanduser().resolve():
+            parser.error("--backtest-db must not be the production database")
+        result = run_backtest(
+            paths,
+            backtest_path,
+            days=args.backtest_days,
+            max_files=args.max_files,
+            max_bytes=args.max_bytes,
+        )
+        print(json.dumps(result, ensure_ascii=False))
+        return 0
     queue = CandidateEventQueue(config.database_path, config.max_candidate_events)
     try:
-        paths = configured_paths(config)
-        result = import_paths(paths, queue, args.max_files, args.max_bytes)
+        result = import_paths(
+            paths,
+            queue,
+            args.max_files,
+            args.max_bytes,
+            reextract_days=args.reextract_days,
+        )
         if args.prune_obsolete:
             result["pruned"] = prune_obsolete_sources(paths, queue)
     finally: