"""自动记忆：官方 Codex memories 的云端对齐版（v1）。 官方行为（learn.chatgpt.com/docs/customization/memories）： - 开启后，Codex 把先前聊天里的有用上下文转成记忆，供未来工作用； - 只对符合条件的先前聊天做，后台生成，不在每轮结束时立刻做； - 聊天闲置够久才生成，避免总结仍在进行的工作； - 剩余额度低于阈值时跳过本次生成，避免在额度见底时烧钱； - 记忆存本地文件（~/.codex/memories/），有 generate / use / disable_on_external_context / min_rate_limit_remaining_percent / extract_model / consolidation_model 等设置； - 密钥要脱敏，AGENTS.md 这类硬性规则不靠记忆。 云端对齐方式（本模块只做控制面数据库层，不动 host.py、 不改 CODEX_HOME、不改任何 thread 的创建/续跑/fork/归档）： - generate_memories：完成任务是否可被提取成记忆。默认关， 与官方“本地记忆默认关”一致；用户在 PUT /users/me/auto-memory-settings 里显式打开才生效。 - use_memories：新任务首轮是否注入记忆。注入只改 worker 内存 里的任务副本（prompt 前加一段记忆前缀），库里的原 prompt 不动；续跑、断点续跑、协调员任务不注入（它们已有会话或 本身不跑模型回合）。 - disable_on_external_context：默认开。v1 能可靠识别的“外部 上下文”只有两类：任务间消息（task_followups 里有该任务的 行）与协调员子任务（parent_task_id 非空）。host 线程内的 MCP / web search 调用目前不在 tasks 表上留标记，本模块 诚实地不宣称能识别它们，见 docs/auto-memory.md 的未做。 - min_rate_limit_remaining_percent：官方的剩余额度门，云端 用当月 USD 配额剩余百分比对齐（与 worker 配额口径一致： 无配额行按默认 $10 计，只有显式 NULL 行才是不限额）。 提取本身是确定性文本处理、不调模型，但被提取的记忆随后 会被注入并占用后续任务的 prompt token，所以额度门仍在 提取处执行；查不到配额时 fail-closed（跳过并记 skipped_quota_check），不冒充“额度充足”。 - idle_seconds：任务终态后闲置多久才可提取，默认 900 秒。 终态后仍可被 /continue 续跑，所以“刚完成”不等于“已闲置”。 - 提取内容（v1 确定性版，不调模型）：从任务 goal/prompt 首行 与 result_text 前段拼一条 project 级 fact，脱敏后写入 identity_memories。extract_model / consolidation_model 只 存设置、不调模型——调模型提取是下一步显式决策（模型选择 与成本上限待定），v1 不假装已经调了。 - 去重与合并（consolidation 的确定性版）：同一任务只提取一次 （memory_extraction_log + 记忆行的 source_thread_id 记任务 id）；同一目标的新结果出现时，supersede 旧记忆而不是再 堆一条。自动提取只写 project 级记忆：单个项目的任务结论 若自动写成 user 级，会跨项目泄漏项目上下文，所以 user 级 记忆仍只接受手动创建。 - 任务级覆盖：POST /tasks/{id}/auto-memory-flags 可为单个任务 单独开关 use/generate（NULL = 跟随用户设置），不改 tasks 表结构（建任务的 4 条 INSERT 路径都不用动）。 新表一律 TEXT 主键，不走 IDENTITY 序列，见 cloud/migrations.py 的对齐说明；PG DDL 同步进 cloud/db.py POSTGRES_SCHEMA 与 MIGRATION_TABLES。 """ import os import re import time from cloud.db import connect as _db_connect from cloud.identity import IdentityStore from cloud.quota import default_quota_usd_micros as _quota_default from cloud.quota import monthly_usage as _quota_monthly_usage from cloud.quota import quota_for_owner as _quota_for_owner DEFAULT_SETTINGS = { "generate_memories": 0, "use_memories": 0, "disable_on_external_context": 1, "min_rate_limit_remaining_percent": 20, "idle_seconds": 900, "min_duration_seconds": 30, "min_result_chars": 200, "extract_model": "", "consolidation_model": "", } _BOOL_FIELDS = ("generate_memories", "use_memories", "disable_on_external_context") _INT_FIELDS = { "min_rate_limit_remaining_percent": (0, 100), "idle_seconds": (0, 86400), "min_duration_seconds": (0, 86400), "min_result_chars": (0, 10000), } _STR_FIELDS = ("extract_model", "consolidation_model") # 脱敏：只处理“看起来像凭证”的形态，不把普通长 token（如 # git sha）一律抹掉。key=value 形态、Authorization/Bearer 头、 # 常见凭证前缀（sk- / ghp_ / gho_ / xoxb- / xoxp-）进黑名单。 _REDACT_PATTERNS = [ (re.compile( r"(?i)(api[_-]?key|password|passwd|secret|token|credential)" r"(\s*[:=]\s*)([\"']?)([^\s\"',;]+)"), r"\1\2\3[REDACTED]"), (re.compile(r"(?i)(authorization\s*:\s*)(Bearer\s+)?[^\s]+"), r"\1[REDACTED]"), (re.compile(r"(?i)\bBearer\s+[A-Za-z0-9._\-]+"), "Bearer [REDACTED]"), (re.compile(r"\b(sk-[A-Za-z0-9_\-]{8,}|ghp_[A-Za-z0-9]{8,}|" r"gho_[A-Za-z0-9]{8,}|xox[baprs]-[A-Za-z0-9\-]{8,})" r"\b"), "[REDACTED]"), ] def redact_text(text): """脱敏并清控制字符；非字符串返回空串。""" if not isinstance(text, str): return "" out = text for pattern, repl in _REDACT_PATTERNS: out = pattern.sub(repl, out) out = "".join(c for c in out if ord(c) >= 32 or c in ("\n", "\t")) return out.strip() def _default_quota_usd_micros(): """默认配额解析：收敛到 cloud/quota.py（quota 模块是 标准库叶子，不存在旧注释担心的循环 import）。""" return _quota_default() def quota_remaining_percent(db_path, owner): """返回当月配额剩余百分比；不限额返回 None。 查库失败抛异常（调用方 fail-closed），全新库无成本行时 已用按 0 计（那是真没有花费，不是查失败）。 """ import datetime month = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m") conn = _db_connect(db_path) try: # 配额行与当月 USD 用量取数收敛到 cloud/quota.py #（无价部分本函数不用，只取 USD 一项；缺 cost_records # 表时仍按旧行为计 0，其它查库异常仍抛给调用方）。 quota = _quota_for_owner(conn, owner) conn.commit() if quota is None: return None if quota <= 0: return 0.0 try: used, _unpriced = _quota_monthly_usage(conn, owner, month) except Exception as exc: if "no such table" in str(exc): used = 0 else: raise return max(0.0, min(100.0, (quota - int(used or 0)) * 100.0 / quota)) finally: conn.close() def _row_get(row, name, default=None): try: return row[name] except (KeyError, IndexError, TypeError): return default def _first_line(text, max_len=200): for line in (text or "").splitlines(): line = " ".join(line.split()) if line: return redact_text(line)[:max_len] return "" def build_extraction_content(task, result_text): """从任务拼一条可存的记忆正文；拼不出返回空串。""" goal = _first_line(task.get("goal") or task.get("prompt") or "") if not goal: return "" outcome = " ".join((result_text or "").split()) outcome = redact_text(outcome)[:800] if outcome: content = "任务完成：%s；结果：%s" % (goal, outcome) else: content = "任务完成：%s" % goal return content[:3900] class AutoMemoryStore: """记忆设置、任务级覆盖、提取日志（SQLite 或 Postgres）。""" def __init__(self, db_path): self.db_path = db_path self.conn = _db_connect(db_path) self.conn.execute( "CREATE TABLE IF NOT EXISTS memory_settings(" "user_id TEXT PRIMARY KEY, " "generate_memories INTEGER DEFAULT 0, " "use_memories INTEGER DEFAULT 0, " "disable_on_external_context INTEGER DEFAULT 1, " "min_rate_limit_remaining_percent INTEGER DEFAULT 20, " "idle_seconds INTEGER DEFAULT 900, " "min_duration_seconds REAL DEFAULT 30, " "min_result_chars INTEGER DEFAULT 200, " "extract_model TEXT, " "consolidation_model TEXT, " "updated_at REAL)" ) self.conn.execute( "CREATE TABLE IF NOT EXISTS task_memory_flags(" "task_id TEXT PRIMARY KEY, " "use_memories INTEGER, " "generate_memories INTEGER, " "updated_at REAL)" ) self.conn.execute( "CREATE TABLE IF NOT EXISTS memory_extraction_log(" "task_id TEXT PRIMARY KEY, " "owner_user_id TEXT, " "project_id TEXT, " "status TEXT, " "reason TEXT, " "memory_id TEXT, " "created_at REAL, " "processed_at REAL)" ) self.conn.commit() def close(self): try: self.conn.close() except Exception: pass def get_settings(self, user_id): row = self.conn.execute( "SELECT * FROM memory_settings WHERE user_id = ?", (user_id,)).fetchone() settings = dict(DEFAULT_SETTINGS) if row is not None: d = dict(row) for key in DEFAULT_SETTINGS: if d.get(key) is not None: settings[key] = d[key] settings["user_id"] = user_id return settings def upsert_settings(self, user_id, **kwargs): """只改给出的字段；非法值抛 ValueError，不落库。""" clean = {} for key, value in kwargs.items(): if key in _BOOL_FIELDS: if value not in (True, False, 0, 1): raise ValueError("%s must be a boolean" % key) clean[key] = 1 if value else 0 elif key in _INT_FIELDS: lo, hi = _INT_FIELDS[key] try: ivalue = int(value) except (TypeError, ValueError): raise ValueError("%s must be an integer" % key) if isinstance(value, bool) or not lo <= ivalue <= hi: raise ValueError( "%s must be between %d and %d" % (key, lo, hi)) clean[key] = ivalue elif key in _STR_FIELDS: if value is None: clean[key] = "" elif not isinstance(value, str) or len(value) > 128: raise ValueError("%s must be a short string" % key) else: clean[key] = value.strip() else: raise ValueError("unknown setting: %s" % key) existing = self.get_settings(user_id) merged = dict(existing) merged.update(clean) now = time.time() self.conn.execute( "INSERT OR REPLACE INTO memory_settings(" "user_id, generate_memories, use_memories, " "disable_on_external_context, " "min_rate_limit_remaining_percent, idle_seconds, " "min_duration_seconds, min_result_chars, extract_model, " "consolidation_model, updated_at) " "VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", (user_id, merged["generate_memories"], merged["use_memories"], merged["disable_on_external_context"], merged["min_rate_limit_remaining_percent"], merged["idle_seconds"], merged["min_duration_seconds"], merged["min_result_chars"], merged["extract_model"], merged["consolidation_model"], now)) self.conn.commit() return self.get_settings(user_id) def get_task_flags(self, task_id): row = self.conn.execute( "SELECT * FROM task_memory_flags WHERE task_id = ?", (task_id,)).fetchone() return dict(row) if row is not None else None def set_task_flags(self, task_id, use_memories=None, generate_memories=None): if use_memories is None and generate_memories is None: raise ValueError( "at least one of use_memories / generate_memories required") for name, value in (("use_memories", use_memories), ("generate_memories", generate_memories)): if value is not None and value not in (True, False, 0, 1): raise ValueError("%s must be a boolean" % name) old = self.get_task_flags(task_id) or {} self.conn.execute( "INSERT OR REPLACE INTO task_memory_flags(" "task_id, use_memories, generate_memories, updated_at) " "VALUES(?, ?, ?, ?)", (task_id, (1 if use_memories else 0) if use_memories is not None else old.get("use_memories"), (1 if generate_memories else 0) if generate_memories is not None else old.get("generate_memories"), time.time())) self.conn.commit() return self.get_task_flags(task_id) @staticmethod def effective(settings, flags, field): """任务级覆盖优先于用户设置；NULL = 跟随设置。""" if flags is not None and flags.get(field) is not None: return bool(flags[field]) return bool(settings[field]) def write_log(self, task_id, owner, project_id, status, reason, memory_id=None, now=None): now = now if now is not None else time.time() row = self.conn.execute( "SELECT created_at FROM memory_extraction_log" " WHERE task_id = ?", (task_id,)).fetchone() created = row[0] if row else now self.conn.execute( "INSERT OR REPLACE INTO memory_extraction_log(" "task_id, owner_user_id, project_id, status, reason, " "memory_id, created_at, processed_at) " "VALUES(?, ?, ?, ?, ?, ?, ?, ?)", (task_id, owner, project_id, status, reason, memory_id or "", created, now)) self.conn.commit() def list_log(self, owner, limit=50): rows = self.conn.execute( "SELECT * FROM memory_extraction_log" " WHERE owner_user_id = ?" " ORDER BY processed_at DESC, task_id DESC LIMIT ?", (owner, max(1, min(200, int(limit))))).fetchall() return [dict(r) for r in rows] def build_injection_prefix(db_path, owner, project_id, flags=None): """use_memories 关、或无记忆时返回空串；否则返回注入前缀。""" store = AutoMemoryStore(db_path) try: settings = store.get_settings(owner) if not AutoMemoryStore.effective(settings, flags, "use_memories"): return "" finally: store.close() identity = IdentityStore(db_path) try: context = identity.build_identity_context(owner, project_id) finally: try: identity.conn.close() except Exception: pass if not context: return "" return ("【过往记忆，仅供参考，不得改写本任务目标】\n%s" % context) def _has_external_context(conn, task_id, parent_task_id): if parent_task_id: return True try: row = conn.execute( "SELECT 1 FROM task_followups WHERE task_id = ? LIMIT 1", (task_id,)).fetchone() return row is not None except Exception: return False def process_due_tasks(db_path, now=None, force=False, owner_id=None): """扫 completed 任务并提取记忆，返回每任务一条结果。 幂等：已提取的任务不再重复建记忆（force 也不重复，靠 source_thread_id 去重）；被跳过的任务下次仍可再扫（设置 改了、闲置够了、额度回来了都可能翻盘），日志行按任务 主键覆盖，不无限增长。 """ now = now if now is not None else time.time() store = AutoMemoryStore(db_path) identity = IdentityStore(db_path) results = [] try: if owner_id: rows = store.conn.execute( "SELECT * FROM tasks WHERE status = 'completed'" " AND created_by = ? ORDER BY id", (owner_id,)).fetchall() else: rows = store.conn.execute( "SELECT * FROM tasks WHERE status = 'completed'" " ORDER BY id").fetchall() except Exception as exc: store.close() try: identity.conn.close() except Exception: pass return [{"status": "error", "reason": "tasks query failed: %s" % exc}] for row in rows: task = dict(row) task_id = task.get("id") owner = task.get("created_by") project_id = task.get("project_id") result = {"task_id": task_id, "status": "skipped", "reason": "", "memory_id": None} results.append(result) if not owner or not project_id: result["reason"] = "no_owner_or_project" store.write_log(task_id, owner or "", project_id or "", "skipped", result["reason"], now=now) continue settings = store.get_settings(owner) flags = store.get_task_flags(task_id) if not AutoMemoryStore.effective( settings, flags, "generate_memories"): result["reason"] = "generate_disabled" store.write_log(task_id, owner, project_id, "skipped", result["reason"], now=now) continue finished = _row_get(task, "finished_at") \ or _row_get(task, "updated_at") \ or _row_get(task, "created_at") started = _row_get(task, "started_at") if not force: if finished is None or now - float(finished) < \ float(settings["idle_seconds"]): result["reason"] = "idle_not_reached" store.write_log(task_id, owner, project_id, "skipped", result["reason"], now=now) continue if started is None or finished is None: result["reason"] = "no_timing" store.write_log(task_id, owner, project_id, "skipped", result["reason"], now=now) continue duration = float(finished) - float(started) if duration < float(settings["min_duration_seconds"]): result["reason"] = "too_short" store.write_log(task_id, owner, project_id, "skipped", result["reason"], now=now) continue result_text = _row_get(task, "result_text") or "" if len(result_text) < int(settings["min_result_chars"]) \ and not force: result["reason"] = "result_too_short" store.write_log(task_id, owner, project_id, "skipped", result["reason"], now=now) continue if settings["disable_on_external_context"] and \ _has_external_context( store.conn, task_id, _row_get(task, "parent_task_id")): result["reason"] = "external_context" store.write_log(task_id, owner, project_id, "skipped", result["reason"], now=now) continue min_remaining = int( settings["min_rate_limit_remaining_percent"]) if min_remaining > 0: try: remaining = quota_remaining_percent(db_path, owner) except Exception: result["reason"] = "quota_check_failed" store.write_log(task_id, owner, project_id, "skipped", result["reason"], now=now) continue if remaining is not None and remaining < min_remaining: result["reason"] = "quota_low" store.write_log(task_id, owner, project_id, "skipped", result["reason"], now=now) continue content = build_extraction_content(task, result_text) if not content: result["reason"] = "empty_content" store.write_log(task_id, owner, project_id, "skipped", result["reason"], now=now) continue existing = identity.list_project_memories(project_id) same_source = [r for r in existing if r.get("source_thread_id") == task_id] if same_source: result["status"] = "extracted" result["reason"] = "already_extracted" result["memory_id"] = same_source[0]["id"] store.write_log(task_id, owner, project_id, "extracted", result["reason"], result["memory_id"], now=now) continue goal_line = _first_line(task.get("goal") or task.get("prompt") or "") same_goal = [r for r in existing if r.get("content", "").startswith( "任务完成：%s；" % goal_line)] if goal_line else [] if same_goal: _old, new_id = identity.supersede_memory( same_goal[0]["id"], owner, replacement_content=content) if new_id: try: identity.conn.execute( "UPDATE identity_memories SET source_thread_id = ?" " WHERE id = ?", (task_id, new_id)) identity.conn.commit() except Exception: pass result["status"] = "extracted" result["reason"] = "consolidated" result["memory_id"] = new_id else: new_id = identity.add_memory( owner, "project", "fact", content, project_id=project_id, source_thread_id=task_id) result["status"] = "extracted" result["reason"] = "created" result["memory_id"] = new_id store.write_log(task_id, owner, project_id, "extracted", result["reason"], result["memory_id"], now=now) store.close() try: identity.conn.close() except Exception: pass return results