diff --git a/README.md b/README.md index a7963c4..31a6891 100644 --- a/README.md +++ b/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. 自动捕获覆盖率。 diff --git a/README_EN.md b/README_EN.md index c56244e..2fe7de2 100644 --- a/README_EN.md +++ b/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. diff --git a/docs/mcp-and-client-adapters.md b/docs/mcp-and-client-adapters.md index 3d7418f..f9f297b 100644 --- a/docs/mcp-and-client-adapters.md +++ b/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 diff --git a/scripts/import_native_history.py b/scripts/import_native_history.py index bb7e165..eb9edc5 100644 --- a/scripts/import_native_history.py +++ b/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: diff --git a/scripts/native_auto_extract.py b/scripts/native_auto_extract.py index da3981a..46c338f 100644 --- a/scripts/native_auto_extract.py +++ b/scripts/native_auto_extract.py @@ -8,23 +8,57 @@ from scripts.bounded_proposals import DEFAULT_MAX_PROPOSALS from scripts.candidate_events import CandidateEventQueue +from scripts.policy import inspect_content DEFAULT_MAX_BYTES = 64 * 1024 DEFAULT_MAX_MESSAGES = 12 +DEFAULT_MAX_UNITS = 12 _USER_RULES = ( ("preference", re.compile(r"(?:记住|以后.{0,80}(?:不要|必须|优先|默认)|默认(?:使用|采用|不要)|不要再|必须始终|优先使用|我偏好|remember|by default|always|never)")), ("decision", re.compile(r"(?:我(?:们)?决定|决定采用|最终决定|确定使用|采用.{0,60}方案|选择.{0,60}而不是|we decided|final decision|adopt.{0,60}approach)")), ("workflow", re.compile(r"(?:(?:以后|必须)先.{0,80}再|流程是|步骤是|first.{0,80}then)")), ) -_ASSISTANT_RULES = ( - ("fact", re.compile(r"(?:已验证[：:]?|验证结果[：:]|测试通过[：:]?|已修复[：:]?|根因[：:]|结论[：:]|verified[：:]?|tests? pass(?:ed)?[：:]?|root cause[：:])", re.IGNORECASE)), - ("workflow", re.compile(r"(?:流程是|步骤是|应当先|应该先|the workflow|steps are)")), +_FACT_EVIDENCE = re.compile( + r"(?:已验证[：:]?|验证结果[：:]|测试证明|verified[：:]?|confirmed[：:]?|tests? pass(?:ed)?[：:]?)", + re.IGNORECASE, +) +_PROJECT_STATE = re.compile( + r"(?:已完成|已部署|已发布|已合并|已修复|仍未完成|尚未完成|" + r"completed|published|merged|fixed|(?:has been|is|was) deployed)", + re.IGNORECASE, +) +_OPEN_LOOP = re.compile( + r"(?:仍需|尚未|仍未|未完成|阻塞(?:于)?|下一步(?:是|需|要)?|pending|blocked)", + re.IGNORECASE, +) +_INSIGHT = re.compile( + r"(?:可迁移规律|通用规律|规律[：:]|核心约束[：:]|模式是|" + r"(?:因为|由于).{2,160}(?:导致|所以|因此)|会导致|意味着)", + re.IGNORECASE, +) +_WORKFLOW = re.compile( + r"(?:流程是|步骤是|应当先|应该先|先.{2,100}再|the workflow|steps are|first.{2,100}then)", + re.IGNORECASE, +) +_ERROR_CAUSE = re.compile(r"(?:根因[：:]|失败条件[：:]|root cause[：:]?|fails? when)", re.IGNORECASE) +_ERROR_REMEDY = re.compile( + r"(?:修复(?:后|[：:])|解决[：:]|规避[：:]|改为|复测|验证后|fixed by|avoid(?:ed)? by)", + re.IGNORECASE, +) +_CONCRETE_SUBJECT = re.compile( + r"(?:`[^`]+`|[/~][\w./-]+|[A-Za-z][A-Za-z0-9_.-]{2,}|" + r"项目|系统|功能|文件|脚本|分支|流程|命令|接口|数据库|会话|候选|回填|发布|" + r"测试|客户端|规则|字段|解析|记忆|版本|提交)" +) +_TRANSIENT = re.compile( + r"(?:现在|今天|明天|今晚|待会|下午|上午|本周|下周|if I am|later today)", + re.IGNORECASE, ) -_TRANSIENT = re.compile(r"(?:现在|今天|明天|今晚|待会|下午|上午|if I am|later today)", re.IGNORECASE) _MEMORY_ECHO = re.compile( r"(?:共找到\s*\d+\s*条记忆|\bID:\s*\d+|内容:\s*|" - r"^(?:web )?search results? for query:)", + r"^(?:web )?search results? for query:|" + r"^(?:MEMORY\.md|rollout_summaries/|skills/).*\|note=\[)", re.IGNORECASE, ) _VAGUE_ASSISTANT = re.compile( @@ -32,7 +66,11 @@ r"(?:对话|任务|问题|内容)(?:串)?[。.]?|I(?:'ve| have) published and verified the live post[.]?)$", re.IGNORECASE, ) -_META_MEMORY = re.compile(r"(?:人工审查|自动提取|误采集|记忆检索|候选(?:里|中|项)|DNA_MEMORY_PROPOSAL)", re.IGNORECASE) +_META_MEMORY = re.compile( + r"(?:人工审查|自动提取|误采集|记忆检索|候选(?:里|中|项)|DNA_MEMORY_PROPOSAL|" + r"memory_(?:remember|recall|feedback|close_session))", + re.IGNORECASE, +) _LOW_SIGNAL_STATUS = re.compile( r"(?:publication report|same-day content|最新主题和详情|" r"(?:全量|现有|单元)?测试(?:通过|\s*`?\d+ passed)|" @@ -47,6 +85,46 @@ r"^(?:你说得对[，,]?我应该|已创建文件\s*\[)", re.IGNORECASE, ) +_CHECKLIST_STATUS = re.compile(r"^(?:✅|☑|✔)") +_LOG_OUTPUT = re.compile(r"^(?:Traceback|\$\s|(?:DEBUG|INFO|WARN|ERROR)\b)", re.IGNORECASE) +_FUTURE_NARRATION = re.compile( + r"(?:^|[，,：:]\s*)(?:我(?:会|将)?先|我会优先|I(?:'ll| will) (?:first|start by))", + re.IGNORECASE, +) +_UNVERIFIED_SPECULATION = re.compile(r"^(?:最可能|大概率|可能是|推测)", re.IGNORECASE) +_IN_PROGRESS_NARRATION = re.compile( + r"(?:当前)?正在(?:做|进行|检查|核验|处理)", + re.IGNORECASE, +) +_COMMAND_PREFACE = re.compile( + r"^(?:但)?你可以直接执行下面", + re.IGNORECASE, +) +_GENERIC_COMPLETION = re.compile( + r"^(?:Markdown\s+格式转换|(?:所有|全部).{0,30}(?:文档|文件)|" + r"(?:不过)?其他.{0,20}功能|好的[，,]?\s*子agent).{0,50}" + r"(?:已完成|已合并|已验证)", + re.IGNORECASE, +) +_RESOURCE_STATUS = re.compile( + r"^Goal\s+已完成.{0,80}(?:tokens?|耗时)", + re.IGNORECASE, +) +_UNVERIFIED_PROMOTIONAL = re.compile( + r"(?:为什么\s*\d+\s*倍|提升\s*\d+\s*倍|减少\s*\d+\s*%)", + re.IGNORECASE, +) + +_IMPORTANCE = { + "preference": 0.8, + "decision": 0.8, + "error_lesson": 0.8, + "insight": 0.75, + "project_state": 0.75, + "workflow": 0.75, + "fact": 0.7, + "open_loop": 0.7, +} def _text(value): @@ -67,19 +145,26 @@ def _record_message(record): if not isinstance(record, dict): return None role = record.get("role") + role = role if isinstance(role, str) else None + record_type = record.get("type") + record_type = record_type if isinstance(record_type, str) else None body = record - if role not in {"user", "assistant"} and record.get("type") in {"user", "assistant"}: - role = record["type"] - if role not in {"user", "assistant"} and record.get("type") == "response_item": - body = record.get("payload") or {} + if role not in {"user", "assistant"} and record_type in {"user", "assistant"}: + role = record_type + if role not in {"user", "assistant"} and record_type == "response_item": + body = record.get("payload") + if not isinstance(body, dict): + return None if body.get("type") != "message": return None role = body.get("role") + role = role if isinstance(role, str) else None if role not in {"user", "assistant"}: return None content = body.get("content") if content is None: - content = (body.get("message") or {}).get("content") + message = body.get("message") + content = message.get("content") if isinstance(message, dict) else None text = _text(content).strip() return { "role": role, @@ -145,10 +230,49 @@ def read_bounded_messages(path, max_bytes=DEFAULT_MAX_BYTES, max_messages=DEFAUL return messages[-max_messages:] -def _sentence(text): - line = next((line.strip() for line in text.splitlines() if line.strip()), "") - parts = re.split(r"(?<=[。！？!?])|(?<=\.)\s+", line) - return next((part.strip() for part in parts if part.strip()), line) +def _candidate_units(text, max_units=DEFAULT_MAX_UNITS): + units = [] + for raw_line in str(text or "").splitlines(): + line = re.sub(r"\s+", " ", raw_line).strip() + line = re.sub(r"^(?:[-*+]\s+|\d+[.)、]\s*)", "", line).strip() + if not line: + continue + for part in re.split(r"(?<=[。！？!?])|(?<=\.)\s+", line): + unit = part.strip() + if unit: + units.append(unit[:800].strip()) + if len(units) >= int(max_units): + return units + return units + + +def _has_concrete_subject(text): + return len(text) >= 12 and not _VAGUE_REFERENCE.search(text) and bool( + _CONCRETE_SUBJECT.search(text) + ) + + +def _classify_candidate(role, text): + if role == "user": + for memory_type, pattern in _USER_RULES: + if pattern.search(text): + return memory_type + return None + if role != "assistant" or not _has_concrete_subject(text): + return None + if _ERROR_CAUSE.search(text) and _ERROR_REMEDY.search(text): + return "error_lesson" + if _OPEN_LOOP.search(text): + return "open_loop" + if _PROJECT_STATE.search(text): + return "project_state" + if _FACT_EVIDENCE.search(text): + return "fact" + if _INSIGHT.search(text): + return "insight" + if _WORKFLOW.search(text): + return "workflow" + return None def extract_automatic_proposals(messages, max_proposals=DEFAULT_MAX_PROPOSALS): @@ -156,57 +280,63 @@ def extract_automatic_proposals(messages, max_proposals=DEFAULT_MAX_PROPOSALS): proposals = [] seen = set() for message in messages: - text = _sentence(message.get("content", ""))[:800].strip() - if not text: - continue - if (_MEMORY_ECHO.search(text) or _META_MEMORY.search(text) - or _LOW_SIGNAL_STATUS.search(text) or _VAGUE_REFERENCE.search(text) - or _CONVERSATIONAL_STATUS.search(text)): - continue - if message.get("role") == "user" and _TRANSIENT.search(text): - continue - if message.get("role") == "assistant" and _VAGUE_ASSISTANT.match(text): - continue - rules = _USER_RULES if message.get("role") == "user" else _ASSISTANT_RULES - for memory_type, pattern in rules: - if not pattern.search(text): + for text in _candidate_units(message.get("content", "")): + if (_MEMORY_ECHO.search(text) or _META_MEMORY.search(text) + or _LOW_SIGNAL_STATUS.search(text) or _VAGUE_REFERENCE.search(text) + or _CONVERSATIONAL_STATUS.search(text) + or _CHECKLIST_STATUS.search(text) or _TRANSIENT.search(text) + or _LOG_OUTPUT.search(text) or _FUTURE_NARRATION.search(text) + or _UNVERIFIED_SPECULATION.search(text) + 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 not inspect_content(text).allowed): continue - key = re.sub(r"\s+", " ", text).lower() - if key in seen: - break - seen.add(key) - proposals.append({ - "type": memory_type, - "summary": text, - "confidence": "high" if message.get("role") == "user" else "medium", - "importance": 0.8 if memory_type in {"preference", "decision"} else 0.7, - }) - break - if len(proposals) >= max_proposals: - break + if message.get("role") == "assistant" and _VAGUE_ASSISTANT.match(text): + continue + memory_type = _classify_candidate(message.get("role"), text) + if memory_type: + key = re.sub(r"\s+", " ", text).lower() + if key in seen: + continue + seen.add(key) + proposals.append({ + "type": memory_type, + "summary": text, + "confidence": "high" if message.get("role") == "user" else "medium", + "importance": _IMPORTANCE[memory_type], + }) + if len(proposals) >= max_proposals: + return proposals[:max_proposals] return proposals[:max_proposals] def _session_id(path, messages): - try: - raw, size = _read_tail_bytes(path, DEFAULT_MAX_BYTES) - data = json.loads(raw.decode("utf-8")) if Path(path).suffix == ".json" and size <= len(raw) else {} - if isinstance(data, dict): - return data.get("session_id") or data.get("sessionId") or data.get("cliSessionId") - except (OSError, UnicodeError, json.JSONDecodeError): - pass + if Path(path).suffix == ".json": + try: + raw, size = _read_tail_bytes(path, DEFAULT_MAX_BYTES) + data = json.loads(raw.decode("utf-8")) if size <= len(raw) else {} + if isinstance(data, dict): + value = data.get("session_id") or data.get("sessionId") or data.get("cliSessionId") + if isinstance(value, (str, int)) and str(value).strip(): + return str(value) + except (OSError, UnicodeError, json.JSONDecodeError): + pass if Path(path).suffix == ".jsonl": try: raw, _ = _read_tail_bytes(path, DEFAULT_MAX_BYTES) for line in raw.decode("utf-8", errors="ignore").splitlines(): - record = json.loads(line) + try: + record = json.loads(line) + except json.JSONDecodeError: + continue payload = record.get("payload") if isinstance(record, dict) else None value = record.get("session_id") or record.get("sessionId") or record.get("cliSessionId") if not value and isinstance(payload, dict): value = payload.get("id") or payload.get("session_id") - if value: + if isinstance(value, (str, int)) and str(value).strip(): return str(value) - except (OSError, json.JSONDecodeError): + except OSError: pass return Path(path).stem @@ -218,7 +348,7 @@ def _project_path(path, messages): return None -def import_native_file(path, queue, client, max_bytes=DEFAULT_MAX_BYTES): +def import_native_file(path, queue, client, max_bytes=DEFAULT_MAX_BYTES, force=False): path = Path(path).expanduser().resolve() stat = path.stat() source_ref = str(path) @@ -226,14 +356,29 @@ def import_native_file(path, queue, client, max_bytes=DEFAULT_MAX_BYTES): tail, _ = _read_tail_bytes(path, max_bytes) digest = hashlib.sha256(tail).hexdigest() checkpoint = queue.get_checkpoint(checkpoint_ref) - if checkpoint and checkpoint[2] == digest: - return {"processed": False, "enqueued": 0, "proposals": 0} + if not force and checkpoint and checkpoint[2] == digest: + return { + "processed": False, + "enqueued": 0, + "proposals": 0, + "proposal_types": {}, + } messages = read_bounded_messages(path, max_bytes=max_bytes) proposals = extract_automatic_proposals(messages) + proposal_types = {} + for proposal in proposals: + memory_type = proposal["type"] + proposal_types[memory_type] = proposal_types.get(memory_type, 0) + 1 session_id = _session_id(path, messages) project_path = _project_path(path, messages) enqueued = 0 - for index, proposal in enumerate(proposals): + existing_proposals = queue.connection.execute( + "SELECT COUNT(*) FROM candidate_events " + "WHERE client=? AND session_id=? AND event_type='memory_proposal'", + (client, session_id), + ).fetchone()[0] + remaining = max(0, DEFAULT_MAX_PROPOSALS - existing_proposals) + for index, proposal in enumerate(proposals[:remaining]): normalized = re.sub(r"\s+", " ", proposal["summary"]).strip().lower() event_hash = hashlib.sha256( (client + "|" + str(session_id) + "|" + normalized).encode() @@ -268,4 +413,9 @@ def import_native_file(path, queue, client, max_bytes=DEFAULT_MAX_BYTES): }): enqueued += 1 queue.update_checkpoint(checkpoint_ref, stat.st_ino, stat.st_size, digest) - return {"processed": True, "enqueued": enqueued, "proposals": len(proposals)} + return { + "processed": True, + "enqueued": enqueued, + "proposals": len(proposals), + "proposal_types": proposal_types, + } diff --git a/tests/test_native_auto_extract.py b/tests/test_native_auto_extract.py index 2e11ee3..978ea5f 100644 --- a/tests/test_native_auto_extract.py +++ b/tests/test_native_auto_extract.py @@ -1,8 +1,11 @@ import json from pathlib import Path +import pytest + from scripts.candidate_events import CandidateEventQueue from scripts.native_auto_extract import ( + _candidate_units, extract_automatic_proposals, import_native_file, read_bounded_messages, @@ -65,7 +68,13 @@ def test_native_import_is_idempotent_and_caps_three_proposals(tmp_path): second = import_native_file(source, queue, client="hermes") assert first["enqueued"] == 3 - assert second == {"processed": False, "enqueued": 0, "proposals": 0} + assert first["proposal_types"] == {"preference": 3} + assert second == { + "processed": False, + "enqueued": 0, + "proposals": 0, + "proposal_types": {}, + } assert queue.connection.execute( "SELECT COUNT(*) FROM candidate_events WHERE event_type='memory_proposal'" ).fetchone()[0] == 3 @@ -103,10 +112,10 @@ def test_rejects_multiline_logs_release_status_and_old_memory_self_report(): def test_sentence_does_not_split_inside_dot_path(): proposals = extract_automatic_proposals([{ "role": "assistant", - "content": "结论：根目录在 `~/.codex/sessions`，并且来源文件可读取。 后续句子不应进入摘要。", + "content": "已验证：根目录在 `~/.codex/sessions`，并且来源文件可读取。 后续句子不应进入摘要。", }]) - assert proposals[0]["summary"] == "结论：根目录在 `~/.codex/sessions`，并且来源文件可读取。" + assert proposals[0]["summary"] == "已验证：根目录在 `~/.codex/sessions`，并且来源文件可读取。" def test_same_session_summary_deduplicates_across_mirrored_files(tmp_path): @@ -124,6 +133,48 @@ def test_same_session_summary_deduplicates_across_mirrored_files(tmp_path): assert import_native_file(second, queue, client="claude-desktop")["enqueued"] == 0 +def test_jsonl_import_uses_stable_fallback_session_id(tmp_path): + source = tmp_path / "rollout-stable-session.jsonl" + source.write_text(json.dumps({ + "role": "user", + "content": "最终决定采用稳定 JSONL 会话标识。", + }, ensure_ascii=False) + "\n") + queue = CandidateEventQueue(tmp_path / "memory.db") + + import_native_file(source, queue, client="codex") + + assert queue.connection.execute( + "SELECT session_id FROM candidate_events WHERE event_type='memory_proposal'" + ).fetchone()[0] == "rollout-stable-session" + + +def test_session_proposal_cap_applies_across_multiple_files(tmp_path): + queue = CandidateEventQueue(tmp_path / "memory.db") + first = tmp_path / "first.json" + second = tmp_path / "second.json" + first.write_text(json.dumps({ + "session_id": "shared-session", + "messages": [ + {"role": "user", "content": "以后默认优先使用有界提取方案 A。"}, + {"role": "user", "content": "最终决定采用候选规则 A。"}, + ], + }, ensure_ascii=False)) + second.write_text(json.dumps({ + "session_id": "shared-session", + "messages": [ + {"role": "user", "content": "以后默认优先使用有界提取方案 B。"}, + {"role": "user", "content": "最终决定采用候选规则 B。"}, + ], + }, ensure_ascii=False)) + + assert import_native_file(first, queue, client="hermes")["enqueued"] == 2 + assert import_native_file(second, queue, client="hermes")["enqueued"] == 1 + assert queue.connection.execute( + "SELECT COUNT(*) FROM candidate_events " + "WHERE event_type='memory_proposal' AND session_id='shared-session'" + ).fetchone()[0] == 3 + + def test_large_hermes_json_reads_complete_messages_from_bounded_tail(tmp_path, monkeypatch): source = tmp_path / "session_large.json" source.write_text(json.dumps({ @@ -144,3 +195,134 @@ def test_large_hermes_json_reads_complete_messages_from_bounded_tail(tmp_path, m assert messages[-1]["role"] == "assistant" assert "有界尾部" in messages[-1]["content"] assert sum(len(message["content"]) for message in messages) < 4096 + + +def test_large_hermes_tail_ignores_nested_non_string_role_and_reads_later_message(tmp_path): + source = tmp_path / "session_nested_role.json" + source.write_text(json.dumps({ + "session_id": "nested-role-session", + "system_prompt": "x" * 10000, + "context": { + "role": {"kind": "worker"}, + "content": "nested tool context is not a message", + }, + "messages": [{ + "role": "assistant", + "content": "已验证：Hermes 合法消息仍可读取。", + }], + }, ensure_ascii=False)) + + messages = read_bounded_messages(source, max_bytes=4096) + + assert messages[-1]["role"] == "assistant" + assert messages[-1]["content"] == "已验证：Hermes 合法消息仍可读取。" + + +def test_extracts_high_signal_later_markdown_bullet(): + proposals = extract_automatic_proposals([{ + "role": "assistant", + "content": ( + "任务处理完成。\n" + "- 根因：Hermes 嵌套对象的 role 是字典；修复后 100 个文件解析无异常。" + ), + }]) + + assert len(proposals) == 1 + assert proposals[0]["summary"].startswith("根因：Hermes 嵌套对象") + + +def test_extracts_numbered_decision_after_low_signal_first_line(): + proposals = extract_automatic_proposals([{ + "role": "user", + "content": ( + "好的。\n" + "1. 最终决定采用有界多行提取方案，而不是逐会话调用外部模型。" + ), + }]) + + assert len(proposals) == 1 + assert proposals[0]["type"] == "decision" + assert proposals[0]["summary"].startswith("最终决定采用") + + +def test_later_candidate_preserves_paths_versions_decimals_and_commit_hashes(): + proposals = extract_automatic_proposals([{ + "role": "assistant", + "content": ( + "任务处理完成。\n" + "- 结论：`~/.codex/sessions` 使用 v2.1.0，阈值 3.14，提交 ce28c4a 已验证。" + ), + }]) + + summary = proposals[0]["summary"] + assert "~/.codex/sessions" in summary + assert "v2.1.0" in summary + assert "3.14" in summary + assert "ce28c4a" in summary + + +def test_candidate_units_are_capped_at_twelve_per_message(): + units = _candidate_units("\n".join( + "- 候选单元 {}。".format(index) for index in range(13) + )) + + assert len(units) == 12 + assert units[-1] == "候选单元 11。" + + +@pytest.mark.parametrize(("role", "content", "expected_type"), [ + ("user", "以后默认优先使用 CodeGraph 定位代码，不要先全库搜索。", "preference"), + ("assistant", "已验证：scripts/native_auto_extract.py 的尾部读取上限是 64KB。", "fact"), + ("assistant", "可迁移规律：当会话结论位于后续项目符号时，只检查第一句会导致候选漏提取。", "insight"), + ("user", "最终决定采用有界多行提取方案，而不是逐会话调用外部模型。", "decision"), + ("assistant", "DNA Memory 自动认知分支已完成 Hermes 解析修复，提交为 6a32e99。", "project_state"), + ("assistant", "DNA Memory 的七天生产回填仍需完成，当前阻塞于质量抽查。", "open_loop"), + ("user", "流程是先运行临时回测，再执行生产回填。", "workflow"), + ("assistant", "根因：Hermes 的 role 字段可能是字典；修复：先验证字段类型，复测 100 个文件无异常。", "error_lesson"), +]) +def test_extracts_all_eight_memory_types(role, content, expected_type): + proposals = extract_automatic_proposals([{"role": role, "content": content}]) + + assert len(proposals) == 1 + assert proposals[0]["type"] == expected_type + + +@pytest.mark.parametrize("role,content", [ + ("assistant", "已完成。"), + ("assistant", "该方案已完成部署。"), + ("assistant", "Traceback: root cause found in line 42."), + ("assistant", "根因：role 字段可能是字典。"), + ("assistant", "仍需处理，当前阻塞。"), + ("assistant", "已验证：DNA Memory password=hunter2 不应保存。"), + ("assistant", "DNA_MEMORY_PROPOSAL {\"type\":\"fact\",\"summary\":\"已验证某功能\"}"), + ("user", "Web search results for query: always prefer this result"), +]) +def test_type_specific_negative_cases_are_rejected(role, content): + assert extract_automatic_proposals([{"role": role, "content": content}]) == [] + + +@pytest.mark.parametrize("content", [ + "调用 memory_remember 写入 type=project_state，summary=跨客户端烟雾测试已完成。", + "Markdown 格式转换已完成。", + "所有汇总文档已合并为一个完整文件。", + "不过其他核心功能都已验证成功。", + "好的，子agent已完成OCR识别。", + "Goal 已完成，用量 500000 tokens，耗时约 40 分钟。", + "核心定位：Forward Deployed Engineer", + "我会先读取相关 skill，再检查全局规则。", + "最可能是外部安全数据库尚未同步。", + "若下周一仍失败，再提交反馈。", + "为什么10倍：先写测试再写代码，质量提升5倍，返工减少80%。", + "文档已经写入，当前正在做最后核验：检查链接与状态是否一致。", + "按需求选择本地工具，我会先读取 skill 说明，再只改全局入口。", + "但你可以直接执行下面这一段，会先创建目录，再追加规则。", + "本周成果已经归档；阻塞集中在两项尚未落地的接入决策。", + "本周 Codex Skill 研究仍有两项接入决策尚未完成。", + "若下周一 Kimi WebBridge 仍未恢复，再提交反馈。", + "Codex 文档已完成初稿，当前正在做最后核验。", + "rollout_summaries/2026-01-01-example.md:10-12|note=[verified image blocker]", +]) +def test_rejects_real_backtest_false_positive_shapes(content): + assert extract_automatic_proposals([ + {"role": "assistant", "content": content}, + ]) == [] diff --git a/tests/test_native_history.py b/tests/test_native_history.py index 557d768..18215bf 100644 --- a/tests/test_native_history.py +++ b/tests/test_native_history.py @@ -1,7 +1,16 @@ import json +import os +import time +import pytest + +import scripts.import_native_history as native_history from scripts.candidate_events import CandidateEventQueue -from scripts.import_native_history import SourceSpec, import_paths, prune_obsolete_sources +from scripts.import_native_history import ( + SourceSpec, + import_paths, + prune_obsolete_sources, +) def test_history_import_scans_supported_sources_with_file_budget(tmp_path): @@ -33,7 +42,15 @@ def test_history_import_scans_supported_sources_with_file_budget(tmp_path): {"codex": [codex], "hermes": [hermes]}, queue, max_files=1, min_age_seconds=0, ) - assert second == {"files": 0, "processed": 0, "enqueued": 0, "proposals": 0} + assert second == { + "files": 0, + "processed": 0, + "enqueued": 0, + "proposals": 0, + "errors": 0, + "proposal_types": {}, + "error_types": {}, + } def test_history_import_uses_explicit_client_source_specs(tmp_path): @@ -111,3 +128,156 @@ def test_prune_obsolete_removes_only_auto_events_for_invalid_sources(tmp_path): assert queue.connection.execute( "SELECT COUNT(*) FROM candidate_events WHERE event_id='manual-pointer'" ).fetchone()[0] == 1 + + +def test_history_import_counts_bad_file_and_continues(tmp_path, monkeypatch): + root = tmp_path / "hermes" + root.mkdir() + bad = root / "a_bad.jsonl" + good = root / "b_good.jsonl" + bad.write_text("invalid native record\n") + good.write_text(json.dumps({ + "role": "user", + "content": "以后必须先验证再发布。", + }, ensure_ascii=False) + "\n") + queue = CandidateEventQueue(tmp_path / "memory.db") + real_import = native_history.import_native_file + + def import_or_fail(path, *args, **kwargs): + if path.name == bad.name: + raise ValueError("invalid native record") + return real_import(path, *args, **kwargs) + + monkeypatch.setattr(native_history, "import_native_file", import_or_fail) + + result = import_paths( + {"hermes": SourceSpec((root,), ("*.jsonl",))}, + queue, + max_files=10, + min_age_seconds=0, + ) + + assert result["files"] == 2 + assert result["processed"] == 1 + assert result["proposals"] == 1 + assert result["errors"] == 1 + assert result["error_types"] == {"hermes:jsonl:ValueError": 1} + serialized = json.dumps(result, ensure_ascii=False) + assert str(bad) not in serialized + assert "invalid native record" not in serialized + + +def test_reextract_days_forces_only_recent_files_and_remains_idempotent(tmp_path): + root = tmp_path / "native" + root.mkdir() + recent = root / "recent.jsonl" + old = root / "old.jsonl" + recent.write_text(json.dumps({ + "role": "user", + "content": "最终决定采用最近七天有界回填方案。", + }, ensure_ascii=False) + "\n") + old.write_text(json.dumps({ + "role": "user", + "content": "最终决定采用旧历史全量回填方案。", + }, ensure_ascii=False) + "\n") + now = time.time() - 300 + os.utime(recent, (now - 86400, now - 86400)) + os.utime(old, (now - 10 * 86400, now - 10 * 86400)) + queue = CandidateEventQueue(tmp_path / "memory.db") + paths = {"codex": SourceSpec((root,), ("*.jsonl",))} + initial = import_paths(paths, queue, max_files=10, min_age_seconds=0, now=now) + assert initial["files"] == 2 + row_count = queue.connection.execute( + "SELECT COUNT(*) FROM candidate_events" + ).fetchone()[0] + + first = import_paths( + paths, + queue, + max_files=10, + min_age_seconds=0, + reextract_days=7, + now=now, + ) + second = import_paths( + paths, + queue, + max_files=10, + min_age_seconds=0, + reextract_days=7, + now=now, + ) + + assert first["files"] == 1 + assert first["processed"] == 1 + assert first["proposals"] == 1 + assert first["enqueued"] == 0 + assert second["files"] == 1 + assert second["enqueued"] == 0 + assert queue.connection.execute( + "SELECT COUNT(*) FROM candidate_events" + ).fetchone()[0] == row_count + + +def test_backtest_uses_separate_database_and_returns_aggregate_metrics(tmp_path): + root = tmp_path / "native" + root.mkdir() + source = root / "recent.jsonl" + source.write_text(json.dumps({ + "role": "assistant", + "content": "已验证：scripts/native_auto_extract.py 可写入隔离回测队列。", + }, ensure_ascii=False) + "\n") + now = time.time() - 300 + os.utime(source, (now - 3600, now - 3600)) + production_path = tmp_path / "production.db" + production = CandidateEventQueue(production_path) + production.update_checkpoint("production-checkpoint", 1, 2, "unchanged") + production.connection.close() + production_bytes = production_path.read_bytes() + backtest_path = tmp_path / "backtest.db" + + result = native_history.run_backtest( + {"codex": SourceSpec((root,), ("*.jsonl",))}, + backtest_path, + days=7, + max_files=10, + min_age_seconds=0, + now=now, + ) + + assert result == { + "mode": "backtest", + "files": 1, + "processed": 1, + "enqueued": 1, + "proposals": 1, + "errors": 0, + "proposal_types": {"fact": 1}, + "error_types": {}, + } + assert production_path.read_bytes() == production_bytes + backtest = CandidateEventQueue(backtest_path) + try: + assert backtest.connection.execute( + "SELECT COUNT(*) FROM candidate_events WHERE event_type='memory_proposal'" + ).fetchone()[0] == 1 + finally: + backtest.connection.close() + serialized = json.dumps(result, ensure_ascii=False) + assert str(source) not in serialized + assert "可写入隔离回测队列" not in serialized + + +def test_backtest_rejects_existing_nonempty_database(tmp_path): + database = tmp_path / "existing.db" + database.write_bytes(b"not a fresh sqlite database") + + with pytest.raises(ValueError, match="new or empty"): + native_history.run_backtest({}, database, days=7) + + +def test_backtest_cli_requires_days_and_database_together(): + with pytest.raises(SystemExit) as error: + native_history.main(["--backtest-days", "7"]) + + assert error.value.code == 2