Merge pull request #108 from openqj/feat-auto-memory

云端身份记忆 + 自动提取注入：官方 memories 对齐（确定性 v1）

cloud/auto_memory.py

@@ -0,0 +1,550 @@
+"""自动记忆：官方 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
+
+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():
+    """与 worker/server 同款默认配额解析，避免循环 import。"""
+    try:
+        return max(0, int(os.environ.get(
+            "CLOUD_DEFAULT_QUOTA_USD_MICROS", "10000000")))
+    except (TypeError, ValueError):
+        return 10000000
+
+
+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:
+        conn.execute(
+            "CREATE TABLE IF NOT EXISTS user_quotas("
+            "user_id TEXT PRIMARY KEY, quota_usd_micros BIGINT, "
+            "updated_at DOUBLE PRECISION)")
+        conn.commit()
+        row = conn.execute(
+            "SELECT quota_usd_micros FROM user_quotas WHERE user_id = ?",
+            (owner,)).fetchone()
+        quota = _default_quota_usd_micros() if row is None else row[0]
+        if quota is None:
+            return None
+        if quota <= 0:
+            return 0.0
+        try:
+            used_row = conn.execute(
+                "SELECT COALESCE(SUM(c.usd_micros), 0)"
+                " FROM cost_records c JOIN tasks t"
+                " ON c.task_id = t.id"
+                " WHERE t.created_by = ? AND c.day LIKE ?",
+                (owner, month + "%")).fetchone()
+            used = used_row[0] if used_row else 0
+        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

cloud/db.py

@@ -946,6 +946,55 @@ POSTGRES_SCHEMA = (
     "preference TEXT NOT NULL, "
     "created_at DOUBLE PRECISION, "
     "UNIQUE(project_id, user_id, channel, target))",
+    # System-level identity (cloud/identity.py): TEXT primary keys,
+    # so no IDENTITY sequence to keep in sync in migrations.py.
+    # DDL must match cloud/identity.py CREATE TABLE statements.
+    "CREATE TABLE IF NOT EXISTS user_profiles("
+    "user_id TEXT PRIMARY KEY, "
+    "display_name TEXT, "
+    "timezone TEXT, "
+    "preferences TEXT, "
+    "updated_at DOUBLE PRECISION)",
+    "CREATE TABLE IF NOT EXISTS identity_memories("
+    "id TEXT PRIMARY KEY, "
+    "owner_user_id TEXT, "
+    "scope TEXT, "
+    "project_id TEXT, "
+    "kind TEXT, "
+    "content TEXT, "
+    "source_thread_id TEXT, "
+    "status TEXT DEFAULT 'active', "
+    "superseded_by TEXT, "
+    "created_at DOUBLE PRECISION, "
+    "updated_at DOUBLE PRECISION)",
+    # Auto memory (cloud/auto_memory.py): TEXT primary keys,
+    # so no IDENTITY sequence to keep in sync in migrations.py.
+    "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 DOUBLE PRECISION DEFAULT 30, "
+    "min_result_chars INTEGER DEFAULT 200, "
+    "extract_model TEXT, "
+    "consolidation_model TEXT, "
+    "updated_at DOUBLE PRECISION)",
+    "CREATE TABLE IF NOT EXISTS task_memory_flags("
+    "task_id TEXT PRIMARY KEY, "
+    "use_memories INTEGER, "
+    "generate_memories INTEGER, "
+    "updated_at DOUBLE PRECISION)",
+    "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 DOUBLE PRECISION, "
+    "processed_at DOUBLE PRECISION)",
 )
 
 

cloud/identity.py

@@ -0,0 +1,301 @@
+"""System-level identity and memory, cloud layer only.
+
+Design (deliberately NOT touching Codex threads):
+
+- Identity lives in the control-plane database, not in CODEX_HOME.
+  A Codex thread is never rewritten by this module: no thread_id is
+  created, resumed, forked or archived here, and no prompt is sent to
+  any worker. ``build_identity_context`` only *formats* text for a
+  caller that explicitly asks for it; whether that text ever reaches
+  a new turn is the caller's decision, recorded separately.
+- Two scopes, no cross-user leakage by construction:
+  - ``user``: owned by one user, visible only to that user, across
+    projects (project_id must be empty).
+  - ``project``: owned by a user but visible to project members;
+    the store does NOT know project membership, so the server must
+    check project ``read``/``create_task`` permission before calling
+    the project-scope methods. Project-scope ownership checks in
+    this module only guard the ``user`` scope; project membership
+    is enforced one layer up.
+- Supersede is a chain, not an overwrite: the old row becomes
+  ``superseded`` and points at its replacement (``superseded_by``),
+  so a correction stays auditable. ``delete`` is physical removal,
+  reserved for user-initiated forgetting, mirroring MemoryStore.
+- Profiles never hold secrets: preference keys containing password /
+  secret / token / api_key / credential are rejected, and values are
+  capped, so a profile cannot become a second credential store.
+
+Tables use TEXT primary keys on purpose: no IDENTITY sequence to
+keep in sync in cloud/migrations.py SEQUENCE_TABLES (see the
+test_p2_migrations alignment test). The same CREATE TABLE statements
+must appear verbatim in cloud/db.py POSTGRES_SCHEMA.
+"""
+
+import json
+import time
+import uuid
+
+from cloud.db import connect as _db_connect
+
+_SCOPES = ("user", "project")
+_KINDS = ("fact", "preference", "decision", "convention")
+_SECRET_KEY_WORDS = ("password", "secret", "token", "api_key", "apikey",
+                     "private_key", "credential")
+_MAX_CONTENT = 4000
+_MAX_PREFERENCES_BYTES = 8192
+
+
+def _clean_str(value, max_len, field):
+    if not isinstance(value, str):
+        raise ValueError("%s must be a string" % field)
+    value = value.strip()
+    if not value:
+        raise ValueError("%s required" % field)
+    if len(value) > max_len:
+        raise ValueError("%s too long" % field)
+    if any(ord(c) < 32 and c not in ("\n", "\t") for c in value):
+        raise ValueError("%s contains control characters" % field)
+    return value
+
+
+def _validate_preferences(preferences):
+    if preferences is None:
+        return None
+    if not isinstance(preferences, dict):
+        raise ValueError("preferences must be an object")
+    for key in preferences:
+        if not isinstance(key, str):
+            raise ValueError("preferences keys must be strings")
+        lowered = key.lower()
+        if any(word in lowered for word in _SECRET_KEY_WORDS):
+            raise ValueError(
+                "preferences must not hold secrets: %s" % key)
+    try:
+        encoded = json.dumps(preferences, ensure_ascii=False)
+    except (TypeError, ValueError):
+        raise ValueError("preferences must be JSON-serializable")
+    if len(encoded.encode("utf-8")) > _MAX_PREFERENCES_BYTES:
+        raise ValueError("preferences too large")
+    return preferences
+
+
+class IdentityStore:
+    """User profiles + scoped identity memories (SQLite or Postgres).
+
+    Instantiation creates the tables if missing (idempotent), the
+    same pattern as MemoryStore / KBStore.
+    """
+
+    def __init__(self, db_path):
+        self.db_path = db_path
+        self.conn = _db_connect(db_path)
+        self.conn.execute(
+            "CREATE TABLE IF NOT EXISTS user_profiles("
+            "user_id TEXT PRIMARY KEY, "
+            "display_name TEXT, "
+            "timezone TEXT, "
+            "preferences TEXT, "
+            "updated_at REAL)"
+        )
+        self.conn.execute(
+            "CREATE TABLE IF NOT EXISTS identity_memories("
+            "id TEXT PRIMARY KEY, "
+            "owner_user_id TEXT, "
+            "scope TEXT, "
+            "project_id TEXT, "
+            "kind TEXT, "
+            "content TEXT, "
+            "source_thread_id TEXT, "
+            "status TEXT DEFAULT 'active', "
+            "superseded_by TEXT, "
+            "created_at REAL, "
+            "updated_at REAL)"
+        )
+        self.conn.commit()
+
+    # -- profiles ------------------------------------------------------
+
+    def get_profile(self, user_id):
+        row = self.conn.execute(
+            "SELECT * FROM user_profiles WHERE user_id = ?",
+            (user_id,)).fetchone()
+        if row is None:
+            return {"user_id": user_id, "display_name": "",
+                    "timezone": "", "preferences": {},
+                    "updated_at": None}
+        d = dict(row)
+        try:
+            d["preferences"] = json.loads(d["preferences"]) \
+                if d["preferences"] else {}
+        except (TypeError, ValueError):
+            d["preferences"] = {}
+        return d
+
+    def upsert_profile(self, user_id, display_name=None, timezone=None,
+                       preferences=None):
+        """Partial update: only fields that are not None change."""
+        if display_name is not None:
+            display_name = _clean_str(display_name, 100, "display_name")
+        if timezone is not None:
+            timezone = _clean_str(timezone, 64, "timezone")
+        preferences = _validate_preferences(preferences)
+        now = time.time()
+        existing = self.conn.execute(
+            "SELECT * FROM user_profiles WHERE user_id = ?",
+            (user_id,)).fetchone()
+        if existing is None:
+            self.conn.execute(
+                "INSERT INTO user_profiles(user_id, display_name, "
+                "timezone, preferences, updated_at) VALUES(?, ?, ?, ?, ?)",
+                (user_id, display_name or "",
+                 timezone or "",
+                 json.dumps(preferences) if preferences is not None
+                 else "{}",
+                 now))
+        else:
+            row = dict(existing)
+            try:
+                old_prefs = json.loads(row["preferences"]) \
+                    if row["preferences"] else {}
+            except (TypeError, ValueError):
+                old_prefs = {}
+            self.conn.execute(
+                "UPDATE user_profiles SET display_name = ?, "
+                "timezone = ?, preferences = ?, updated_at = ? "
+                "WHERE user_id = ?",
+                (display_name if display_name is not None
+                 else (row["display_name"] or ""),
+                 timezone if timezone is not None
+                 else (row["timezone"] or ""),
+                 json.dumps(preferences) if preferences is not None
+                 else json.dumps(old_prefs),
+                 now, user_id))
+        self.conn.commit()
+        return self.get_profile(user_id)
+
+    # -- memories -------------------------------------------------------
+
+    def add_memory(self, owner_user_id, scope, kind, content,
+                   project_id=None, source_thread_id=None):
+        if scope not in _SCOPES:
+            raise ValueError("scope must be 'user' or 'project'")
+        if kind not in _KINDS:
+            raise ValueError("kind must be one of %s" % (", ".join(_KINDS),))
+        content = _clean_str(content, _MAX_CONTENT, "content")
+        if scope == "user":
+            if project_id:
+                raise ValueError(
+                    "user-scope memory must not carry a project_id")
+            project_id = ""
+        else:
+            if not project_id or not isinstance(project_id, str):
+                raise ValueError(
+                    "project-scope memory requires a project_id")
+            project_id = project_id.strip()
+        if source_thread_id is not None:
+            source_thread_id = _clean_str(
+                source_thread_id, 128, "source_thread_id")
+        mem_id = uuid.uuid4().hex
+        now = time.time()
+        self.conn.execute(
+            "INSERT INTO identity_memories(id, owner_user_id, scope, "
+            "project_id, kind, content, source_thread_id, status, "
+            "superseded_by, created_at, updated_at) "
+            "VALUES(?, ?, ?, ?, ?, ?, ?, 'active', '', ?, ?)",
+            (mem_id, owner_user_id, scope, project_id, kind, content,
+             source_thread_id or "", now, now))
+        self.conn.commit()
+        return mem_id
+
+    def get_memory(self, mem_id):
+        row = self.conn.execute(
+            "SELECT * FROM identity_memories WHERE id = ?",
+            (mem_id,)).fetchone()
+        return dict(row) if row is not None else None
+
+    def list_user_memories(self, user_id, include_superseded=False):
+        sql = ("SELECT * FROM identity_memories "
+               "WHERE owner_user_id = ? AND scope = 'user'")
+        if not include_superseded:
+            sql += " AND status = 'active'"
+        sql += " ORDER BY created_at DESC, id DESC"
+        rows = self.conn.execute(sql, (user_id,)).fetchall()
+        return [dict(r) for r in rows]
+
+    def list_project_memories(self, project_id, include_superseded=False):
+        """Caller must already hold project ``read`` permission."""
+        sql = ("SELECT * FROM identity_memories "
+               "WHERE scope = 'project' AND project_id = ?")
+        if not include_superseded:
+            sql += " AND status = 'active'"
+        sql += " ORDER BY created_at DESC, id DESC"
+        rows = self.conn.execute(sql, (project_id,)).fetchall()
+        return [dict(r) for r in rows]
+
+    def supersede_memory(self, mem_id, actor_user_id,
+                         replacement_content=None, replacement_kind=None):
+        """Mark old row superseded; optionally create its replacement.
+
+        User-scope rows may only be superseded by their owner
+        (PermissionError otherwise). Project-scope membership is
+        enforced by the server before calling this method.
+        Returns (old_id, new_id_or_None).
+        """
+        row = self.get_memory(mem_id)
+        if row is None:
+            raise KeyError("memory not found")
+        if row["scope"] == "user" and row["owner_user_id"] != actor_user_id:
+            raise PermissionError("not the owner of this memory")
+        new_id = None
+        if replacement_content is not None:
+            kind = replacement_kind or row["kind"]
+            new_id = self.add_memory(
+                row["owner_user_id"], row["scope"], kind,
+                replacement_content,
+                project_id=(row["project_id"] or None)
+                if row["scope"] == "project" else None,
+                source_thread_id=(row["source_thread_id"] or None))
+        self.conn.execute(
+            "UPDATE identity_memories SET status = 'superseded', "
+            "superseded_by = ?, updated_at = ? WHERE id = ?",
+            (new_id or "", time.time(), mem_id))
+        self.conn.commit()
+        return mem_id, new_id
+
+    def delete_memory(self, mem_id, actor_user_id):
+        """Physical removal (user-initiated forgetting only)."""
+        row = self.get_memory(mem_id)
+        if row is None:
+            raise KeyError("memory not found")
+        if row["scope"] == "user" and row["owner_user_id"] != actor_user_id:
+            raise PermissionError("not the owner of this memory")
+        self.conn.execute(
+            "DELETE FROM identity_memories WHERE id = ?", (mem_id,))
+        self.conn.commit()
+        return True
+
+    def build_identity_context(self, user_id, project_id=None,
+                               top_k_user=5, top_k_project=5):
+        """Format active memories for an explicit caller.
+
+        Returns "" when there is nothing active. Never returns the
+        full history: at most top_k per scope. Project memories are
+        included only when project_id is given; the caller must have
+        checked project ``read`` permission first.
+        """
+        sections = []
+        user_rows = self.list_user_memories(user_id)[:top_k_user]
+        if user_rows:
+            lines = ["[用户记忆]"]
+            for r in user_rows:
+                lines.append("- [%s] %s" % (r["kind"], r["content"]))
+            sections.append("\n".join(lines))
+        if project_id:
+            proj_rows = self.list_project_memories(
+                project_id)[:top_k_project]
+            if proj_rows:
+                lines = ["[项目记忆]"]
+                for r in proj_rows:
+                    lines.append("- [%s] %s" % (r["kind"], r["content"]))
+                sections.append("\n".join(lines))
+        return "\n".join(sections)

cloud/migrations.py

@@ -68,6 +68,13 @@ MIGRATION_TABLES = (
     "task_attachments",
     "task_followups",
     "environments",
+    # Cloud identity batch: TEXT primary keys, no sequence to sync.
+    "user_profiles",
+    "identity_memories",
+    # Auto memory batch: same TEXT-primary-key pattern.
+    "memory_settings",
+    "task_memory_flags",
+    "memory_extraction_log",
 )
 
 _SKIP_TABLES = frozenset({"kb_fts", "sqlite_sequence"})

cloud/server.py

@@ -110,6 +110,9 @@ from cloud.approvals import ensure_approvals_table as _appr_ensure
 from cloud.approvals import expire_overdue as _appr_expire
 from cloud.approvals import get_approval as _appr_get
 from cloud.approvals import list_approvals as _appr_list
+from cloud.auto_memory import AutoMemoryStore
+from cloud.auto_memory import process_due_tasks as _auto_process_due
+from cloud.identity import IdentityStore
 from cloud.kb import KBStore
 from cloud.memory import MemoryStore
 from cloud.pr import PrSkipped, PrStepError
@@ -1664,6 +1667,11 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         # 项目补仓库是第一个 PATCH 路由；与其它方法一样走 _dispatch。
         self._dispatch_safe("PATCH")
 
+    def do_PUT(self):
+        # PUT 与 PATCH 同理：此前无入口，/users/me/profile 是
+        # 第一个 PUT 路由；未知 PUT 路径仍由 _dispatch 回 404。
+        self._dispatch_safe("PUT")
+
     def _dispatch_safe(self, method):
         """分发兜底：_dispatch 里任何未捕获异常都回 500 JSON，而不是
         让请求线程直接断连（2026-10-27 死锁事件就是这么变成前端
@@ -1879,6 +1887,34 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 self._send_json(404, {"error": "not found"})
         elif path == "/users/me/quota" and method == "GET":
             self._user_quota_self(user_id)
+        elif path == "/users/me/profile" and method == "GET":
+            self._identity_profile_get(user_id)
+        elif path == "/users/me/profile" and method in ("PUT", "POST"):
+            self._identity_profile_upsert(user_id)
+        elif path == "/users/me/memories" and method == "GET":
+            self._identity_user_memories_list(user_id)
+        elif path == "/users/me/memories" and method == "POST":
+            self._identity_user_memory_create(user_id)
+        elif path == "/users/me/identity-context" and method == "GET":
+            self._identity_context(user_id, query)
+        elif path == "/users/me/auto-memory-settings" and method == "GET":
+            self._auto_memory_settings_get(user_id)
+        elif path == "/users/me/auto-memory-settings" and \
+                method in ("PUT", "POST"):
+            self._auto_memory_settings_upsert(user_id)
+        elif path == "/users/me/auto-memory-log" and method == "GET":
+            self._auto_memory_log(user_id, query)
+        elif path == "/users/me/auto-memory/run" and method == "POST":
+            self._auto_memory_run(user_id)
+        elif path.startswith("/users/me/memories/"):
+            _mid = path[len("/users/me/memories/"):]
+            if _mid and "/" not in _mid and method == "DELETE":
+                self._identity_user_memory_delete(user_id, _mid)
+            elif _mid.endswith("/supersede") and method == "POST":
+                self._identity_user_memory_supersede(
+                    user_id, _mid[: -len("/supersede")].rstrip("/"))
+            else:
+                self._send_json(404, {"error": "not found"})
         elif path == "/tasks" and method == "POST":
             self._create_task(user_id)
         elif path == "/uploads" and method == "POST":
@@ -1918,6 +1954,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 self._retry_task(user_id, rest[: -len("/retry")])
             elif rest.endswith("/goal-resume") and method == "POST":
                 self._goal_resume_task(user_id, rest[: -len("/goal-resume")])
+            elif rest.endswith("/auto-memory-flags") and method == "POST":
+                self._task_auto_memory_flags(
+                    user_id, rest[: -len("/auto-memory-flags")])
             elif rest.endswith("/heartbeat") and method == "POST":
                 self._heartbeat_task(user_id, rest[: -len("/heartbeat")])
             elif rest.endswith("/followups") and method == "POST":
@@ -2688,6 +2727,48 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             else:
                 self._send_json(404, {"error": "not found"})
             return
+        # /projects/{id}/identity-memories (project-scope system memory)
+        if len(parts) == 2 and parts[1] == "identity-memories":
+            if method == "POST":
+                err = self._check_project(user_id, project_id, "create_task")
+                if err:
+                    self._send_json(err[0], {"error": err[1]})
+                    return
+                self._project_identity_memory_create(user_id, project_id)
+            elif method == "GET":
+                err = self._check_project(user_id, project_id, "read")
+                if err:
+                    self._send_json(err[0], {"error": err[1]})
+                    return
+                self._project_identity_memories_list(project_id, query)
+            else:
+                self._send_json(404, {"error": "not found"})
+            return
+        # /projects/{id}/identity-memories/{mid}/supersede
+        if len(parts) == 4 and parts[1] == "identity-memories" \
+                and parts[3] == "supersede":
+            if method != "POST":
+                self._send_json(404, {"error": "not found"})
+                return
+            err = self._check_project(user_id, project_id, "create_task")
+            if err:
+                self._send_json(err[0], {"error": err[1]})
+                return
+            self._project_identity_memory_supersede(
+                user_id, project_id, parts[2])
+            return
+        # /projects/{id}/identity-memories/{mid}
+        if len(parts) == 3 and parts[1] == "identity-memories":
+            if method != "DELETE":
+                self._send_json(404, {"error": "not found"})
+                return
+            err = self._check_project(user_id, project_id, "create_task")
+            if err:
+                self._send_json(err[0], {"error": err[1]})
+                return
+            self._project_identity_memory_delete(
+                user_id, project_id, parts[2])
+            return
         # /projects/{id}/memory
         if len(parts) == 2 and parts[1] == "memory":
             if method == "POST":
@@ -3036,6 +3117,365 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             auth.audit(user_id, project_id, "supersede_memory", str(mid))
         self._send_json(200, {"id": mid, "status": "deprecated"})
 
+    # -- System-level identity / memory (cloud only) ------------------
+    # These handlers never touch host.py, CODEX_HOME or any thread:
+    # they only read/write the control-plane database. The context
+    # endpoint returns text for an explicit caller; it does not
+    # inject anything into a task or a thread on its own.
+
+    def _identity_store(self):
+        return IdentityStore(self.server.db_path)
+
+    def _identity_close(self, store):
+        try:
+            store.conn.close()
+        except Exception:
+            pass
+
+    def _identity_profile_get(self, user_id):
+        store = self._identity_store()
+        try:
+            self._send_json(200, store.get_profile(user_id))
+        finally:
+            self._identity_close(store)
+
+    def _identity_profile_upsert(self, user_id):
+        body, err = self._read_json()
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        store = self._identity_store()
+        try:
+            try:
+                profile = store.upsert_profile(
+                    user_id,
+                    display_name=body.get("display_name"),
+                    timezone=body.get("timezone"),
+                    preferences=body.get("preferences"))
+            except ValueError as exc:
+                self._send_json(400, {"error": str(exc)})
+                return
+        finally:
+            self._identity_close(store)
+        with _auth_store(self.server.db_path) as auth:
+            auth.audit(user_id, "", "profile_updated", user_id)
+        self._send_json(200, profile)
+
+    def _identity_user_memories_list(self, user_id):
+        store = self._identity_store()
+        try:
+            rows = store.list_user_memories(user_id)
+        finally:
+            self._identity_close(store)
+        self._send_json(200, {"user_id": user_id, "memories": rows})
+
+    def _identity_user_memory_create(self, user_id):
+        body, err = self._read_json()
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        store = self._identity_store()
+        try:
+            try:
+                mem_id = store.add_memory(
+                    user_id, "user", body.get("kind"),
+                    body.get("content"),
+                    source_thread_id=body.get("source_thread_id"))
+            except ValueError as exc:
+                self._send_json(400, {"error": str(exc)})
+                return
+        finally:
+            self._identity_close(store)
+        with _auth_store(self.server.db_path) as auth:
+            auth.audit(user_id, "", "identity_memory_created", mem_id)
+        self._send_json(201, {"id": mem_id, "scope": "user"})
+
+    def _identity_user_memory_supersede(self, user_id, mem_id):
+        body, err = self._read_json()
+        if err:
+            body = {}
+        store = self._identity_store()
+        try:
+            row = store.get_memory(mem_id)
+            if row is None or row["scope"] != "user" \
+                    or row["owner_user_id"] != user_id:
+                self._send_json(404, {"error": "memory not found"})
+                return
+            try:
+                _old, new_id = store.supersede_memory(
+                    mem_id, user_id,
+                    replacement_content=body.get("replacement_content"),
+                    replacement_kind=body.get("replacement_kind"))
+            except (KeyError, PermissionError):
+                self._send_json(404, {"error": "memory not found"})
+                return
+            except ValueError as exc:
+                self._send_json(400, {"error": str(exc)})
+                return
+        finally:
+            self._identity_close(store)
+        with _auth_store(self.server.db_path) as auth:
+            auth.audit(user_id, "", "identity_memory_superseded", mem_id)
+        self._send_json(200, {"id": mem_id, "status": "superseded",
+                              "replacement_id": new_id})
+
+    def _identity_user_memory_delete(self, user_id, mem_id):
+        store = self._identity_store()
+        try:
+            row = store.get_memory(mem_id)
+            if row is None or row["scope"] != "user" \
+                    or row["owner_user_id"] != user_id:
+                self._send_json(404, {"error": "memory not found"})
+                return
+            try:
+                store.delete_memory(mem_id, user_id)
+            except (KeyError, PermissionError):
+                self._send_json(404, {"error": "memory not found"})
+                return
+        finally:
+            self._identity_close(store)
+        with _auth_store(self.server.db_path) as auth:
+            auth.audit(user_id, "", "identity_memory_deleted", mem_id)
+        self._send_json(200, {"deleted": True, "id": mem_id})
+
+    def _identity_context(self, user_id, query):
+        project_id = None
+        if query:
+            vals = query.get("project_id")
+            if vals:
+                project_id = vals[0]
+        if project_id:
+            err = self._check_project(user_id, project_id, "read")
+            if err:
+                self._send_json(err[0], {"error": err[1]})
+                return
+        store = self._identity_store()
+        try:
+            profile = store.get_profile(user_id)
+            user_rows = store.list_user_memories(user_id)
+            proj_rows = store.list_project_memories(project_id) \
+                if project_id else []
+            context = store.build_identity_context(user_id, project_id)
+        finally:
+            self._identity_close(store)
+        summaries_context = ""
+        if project_id:
+            mem_store = MemoryStore(self.server.db_path)
+            try:
+                summaries_context = mem_store.build_prompt_context(
+                    project_id)
+            finally:
+                try:
+                    mem_store.conn.close()
+                except Exception:
+                    pass
+        self._send_json(200, {
+            "user_id": user_id,
+            "project_id": project_id,
+            "profile": profile,
+            "user_memories": user_rows,
+            "project_memories": proj_rows,
+            "identity_context": context,
+            # Kept as a separate field on purpose: the existing
+            # project summaries are NOT merged into identity_context
+            # and neither field is injected into any task/thread here.
+            "project_summaries_context": summaries_context,
+        })
+
+    # -- 自动记忆（官方 memories 对齐：设置 + 提取触发） ------------
+    # 设置与提取日志只归属本人；任务级开关走项目 create_task 门。
+
+    def _auto_memory_store(self):
+        return AutoMemoryStore(self.server.db_path)
+
+    def _auto_memory_close(self, store):
+        try:
+            store.close()
+        except Exception:
+            pass
+
+    def _auto_memory_settings_get(self, user_id):
+        store = self._auto_memory_store()
+        try:
+            self._send_json(200, store.get_settings(user_id))
+        finally:
+            self._auto_memory_close(store)
+
+    def _auto_memory_settings_upsert(self, user_id):
+        body, err = self._read_json()
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        if not isinstance(body, dict):
+            self._send_json(400, {"error": "body must be an object"})
+            return
+        store = self._auto_memory_store()
+        try:
+            try:
+                settings = store.upsert_settings(user_id, **body)
+            except (ValueError, TypeError) as exc:
+                self._send_json(400, {"error": str(exc)})
+                return
+        finally:
+            self._auto_memory_close(store)
+        with _auth_store(self.server.db_path) as auth:
+            auth.audit(user_id, "", "auto_memory_settings_updated", "")
+        self._send_json(200, settings)
+
+    def _auto_memory_log(self, user_id, query):
+        try:
+            limit = int((query.get("limit") or ["50"])[0])
+        except (TypeError, ValueError):
+            self._send_json(400, {"error": "limit must be an integer"})
+            return
+        store = self._auto_memory_store()
+        try:
+            rows = store.list_log(user_id, limit=limit)
+        finally:
+            self._auto_memory_close(store)
+        self._send_json(200, {"user_id": user_id, "log": rows})
+
+    def _auto_memory_run(self, user_id):
+        body, err = self._read_json()
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        force = False
+        if isinstance(body, dict) and body.get("force") is not None:
+            if body.get("force") not in (True, False, 0, 1):
+                self._send_json(400, {"error": "force must be a boolean"})
+                return
+            force = bool(body.get("force"))
+        results = _auto_process_due(
+            self.server.db_path, force=force, owner_id=user_id)
+        for item in results:
+            if item.get("status") == "extracted":
+                # 提取日志已在 process_due_tasks 里落库，这里只补
+                # 一条审计（不查任务归属：run 只扫本人的任务）。
+                task_id = item.get("task_id") or ""
+                with _auth_store(self.server.db_path) as auth:
+                    auth.audit(user_id, "", "auto_memory_extracted",
+                               task_id)
+        self._send_json(200, {"user_id": user_id, "results": results})
+
+    def _task_auto_memory_flags(self, user_id, task_id):
+        body, err = self._read_json()
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        if not isinstance(body, dict):
+            self._send_json(400, {"error": "body must be an object"})
+            return
+        task = self._fetch_task(task_id)
+        if task is None:
+            self._send_json(404, {"error": "task not found"})
+            return
+        err = self._check_project(
+            user_id, task["project_id"], "create_task")
+        if err:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        store = self._auto_memory_store()
+        try:
+            try:
+                flags = store.set_task_flags(
+                    task_id,
+                    use_memories=body.get("use_memories"),
+                    generate_memories=body.get("generate_memories"))
+            except ValueError as exc:
+                self._send_json(400, {"error": str(exc)})
+                return
+        finally:
+            self._auto_memory_close(store)
+        with _auth_store(self.server.db_path) as auth:
+            auth.audit(user_id, task["project_id"],
+                       "task_auto_memory_flags", task_id)
+        self._send_json(200, {"task_id": task_id, "flags": flags})
+
+    def _project_identity_memory_create(self, user_id, project_id):
+        body, err = self._read_json()
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        store = self._identity_store()
+        try:
+            try:
+                mem_id = store.add_memory(
+                    user_id, "project", body.get("kind"),
+                    body.get("content"), project_id=project_id,
+                    source_thread_id=body.get("source_thread_id"))
+            except ValueError as exc:
+                self._send_json(400, {"error": str(exc)})
+                return
+        finally:
+            self._identity_close(store)
+        with _auth_store(self.server.db_path) as auth:
+            auth.audit(user_id, project_id,
+                       "identity_memory_created", mem_id)
+        self._send_json(201, {"id": mem_id, "scope": "project",
+                              "project_id": project_id})
+
+    def _project_identity_memories_list(self, project_id, query):
+        try:
+            top_k = int((query.get("top_k") or ["20"])[0])
+        except (TypeError, ValueError):
+            self._send_json(400, {"error": "top_k must be an integer"})
+            return
+        store = self._identity_store()
+        try:
+            rows = store.list_project_memories(project_id)
+        finally:
+            self._identity_close(store)
+        self._send_json(200, {"project_id": project_id,
+                              "memories": rows[:top_k]})
+
+    def _project_identity_memory_supersede(self, user_id, project_id,
+                                           mem_id):
+        body, err = self._read_json()
+        if err:
+            body = {}
+        store = self._identity_store()
+        try:
+            row = store.get_memory(mem_id)
+            if row is None or row["scope"] != "project" \
+                    or row["project_id"] != project_id:
+                self._send_json(404, {"error": "memory not found"})
+                return
+            try:
+                _old, new_id = store.supersede_memory(
+                    mem_id, user_id,
+                    replacement_content=body.get("replacement_content"),
+                    replacement_kind=body.get("replacement_kind"))
+            except (KeyError, PermissionError, ValueError) as exc:
+                if isinstance(exc, ValueError):
+                    self._send_json(400, {"error": str(exc)})
+                else:
+                    self._send_json(404, {"error": "memory not found"})
+                return
+        finally:
+            self._identity_close(store)
+        with _auth_store(self.server.db_path) as auth:
+            auth.audit(user_id, project_id,
+                       "identity_memory_superseded", mem_id)
+        self._send_json(200, {"id": mem_id, "status": "superseded",
+                              "replacement_id": new_id})
+
+    def _project_identity_memory_delete(self, user_id, project_id, mem_id):
+        store = self._identity_store()
+        try:
+            row = store.get_memory(mem_id)
+            if row is None or row["scope"] != "project" \
+                    or row["project_id"] != project_id:
+                self._send_json(404, {"error": "memory not found"})
+                return
+            store.delete_memory(mem_id, user_id)
+        finally:
+            self._identity_close(store)
+        with _auth_store(self.server.db_path) as auth:
+            auth.audit(user_id, project_id,
+                       "identity_memory_deleted", mem_id)
+        self._send_json(200, {"deleted": True, "id": mem_id})
+
     def _kb_index(self, user_id, project_id):
         body, err = self._read_json()
         if err:

cloud/tests/test_auto_memory.py

@@ -0,0 +1,482 @@
+"""Tests for cloud/auto_memory.py + worker injection + HTTP settings.
+
+Scope, stated so a future reader does not over-read it: v1
+deterministic extraction only. No test here calls a model for
+extraction, touches host.py, CODEX_HOME or any Codex thread; the
+store, the worker hook and the HTTP endpoints only read/write the
+control-plane database, and injection only changes the in-memory
+task copy the worker hands to its executor.
+
+Run from the repo root:
+    python3 -m unittest discover -s cloud/tests -p "test_auto_memory.py" -v
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.auto_memory import AutoMemoryStore
+    from cloud.auto_memory import build_injection_prefix
+    from cloud.auto_memory import process_due_tasks
+    from cloud.auto_memory import redact_text
+    from cloud.db import POSTGRES_SCHEMA
+    from cloud.identity import IdentityStore
+    from cloud.migrations import MIGRATION_TABLES
+    from cloud.server import create_server
+    from cloud.server import init_db
+    from cloud.worker import Worker
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.auto_memory import AutoMemoryStore
+    from cloud.auto_memory import build_injection_prefix
+    from cloud.auto_memory import process_due_tasks
+    from cloud.auto_memory import redact_text
+    from cloud.db import POSTGRES_SCHEMA
+    from cloud.identity import IdentityStore
+    from cloud.migrations import MIGRATION_TABLES
+    from cloud.server import create_server
+    from cloud.server import init_db
+    from cloud.worker import Worker
+
+
+def _tmp_db():
+    fd, path = tempfile.mkstemp(suffix=".db")
+    os.close(fd)
+    os.unlink(path)
+    return path
+
+
+def _insert_completed(db_path, task_id, owner="u1", project="p1",
+                      prompt="做登录页", result=None, started_ago=2000,
+                      duration=120, goal=None):
+    now = time.time()
+    finished = now - 1000
+    started = finished - duration
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute(
+            "INSERT INTO tasks(id, project_id, prompt, status,"
+            " idempotency_key, depends_on, priority, created_by,"
+            " created_at, updated_at, last_heartbeat, goal,"
+            " started_at, finished_at, result_text)"
+            " VALUES(?, ?, ?, 'completed', NULL, '[]', 0, ?, ?, ?,"
+            " NULL, ?, ?, ?, ?)",
+            (task_id, project, prompt, owner, now, finished,
+             goal, started, finished,
+             result if result is not None else "登录页已完成。" + "x" * 300),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+    return finished
+
+
+class RedactTest(unittest.TestCase):
+    def test_key_value_and_bearer_redacted(self):
+        out = redact_text("api_key=abc123 and password: hunter2")
+        self.assertNotIn("abc123", out)
+        self.assertNotIn("hunter2", out)
+        self.assertIn("[REDACTED]", out)
+
+    def test_plain_text_untouched(self):
+        self.assertEqual(redact_text("hello world"), "hello world")
+        self.assertEqual(redact_text(None), "")
+
+
+class StoreTest(unittest.TestCase):
+    def setUp(self):
+        self.db = _tmp_db()
+        self.store = AutoMemoryStore(self.db)
+
+    def tearDown(self):
+        try:
+            self.store.close()
+        finally:
+            if os.path.exists(self.db):
+                os.unlink(self.db)
+
+    def test_defaults_off(self):
+        settings = self.store.get_settings("u1")
+        self.assertEqual(settings["generate_memories"], 0)
+        self.assertEqual(settings["use_memories"], 0)
+        self.assertEqual(settings["disable_on_external_context"], 1)
+        self.assertEqual(settings["min_rate_limit_remaining_percent"], 20)
+        self.assertEqual(settings["idle_seconds"], 900)
+
+    def test_upsert_validates_ranges(self):
+        with self.assertRaises(ValueError):
+            self.store.upsert_settings("u1",
+                                       min_rate_limit_remaining_percent=101)
+        with self.assertRaises(ValueError):
+            self.store.upsert_settings("u1", idle_seconds=-1)
+        with self.assertRaises(ValueError):
+            self.store.upsert_settings("u1", bogus_field=1)
+        got = self.store.upsert_settings(
+            "u1", generate_memories=True, use_memories=1,
+            idle_seconds=0, min_duration_seconds=0, min_result_chars=0)
+        self.assertEqual(got["generate_memories"], 1)
+        self.assertEqual(got["idle_seconds"], 0)
+
+    def test_task_flag_overrides_settings(self):
+        settings = self.store.upsert_settings("u1", generate_memories=False)
+        self.assertFalse(AutoMemoryStore.effective(settings, None,
+                                                    "generate_memories"))
+        self.store.set_task_flags("t1", generate_memories=True)
+        flags = self.store.get_task_flags("t1")
+        self.assertTrue(AutoMemoryStore.effective(settings, flags,
+                                                  "generate_memories"))
+        with self.assertRaises(ValueError):
+            self.store.set_task_flags("t2")
+
+    def test_pg_schema_and_migration_lists_include_new_tables(self):
+        schema_text = "\n".join(POSTGRES_SCHEMA)
+        for table in ("memory_settings", "task_memory_flags",
+                      "memory_extraction_log"):
+            self.assertIn("CREATE TABLE IF NOT EXISTS %s" % table,
+                          schema_text)
+            self.assertIn(table, MIGRATION_TABLES)
+
+
+class ExtractionTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        init_db(self.db_path)
+        store = AutoMemoryStore(self.db_path)
+        store.upsert_settings(
+            "u1", generate_memories=True, use_memories=True,
+            idle_seconds=900, min_duration_seconds=30,
+            min_result_chars=200, min_rate_limit_remaining_percent=0)
+        store.close()
+        # Unlimited quota: explicit NULL row, so the quota gate passes.
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO user_quotas(user_id, quota_usd_micros,"
+                " updated_at) VALUES('u1', NULL, ?)", (time.time(),))
+            conn.commit()
+        finally:
+            conn.close()
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _memories(self):
+        store = IdentityStore(self.db_path)
+        try:
+            return store.list_project_memories("p1")
+        finally:
+            store.conn.close()
+
+    def test_happy_path_creates_project_memory_and_is_idempotent(self):
+        _insert_completed(self.db_path, "t1")
+        results = process_due_tasks(self.db_path)
+        self.assertEqual(results[0]["status"], "extracted")
+        self.assertEqual(results[0]["reason"], "created")
+        memories = self._memories()
+        self.assertEqual(len(memories), 1)
+        self.assertEqual(memories[0]["scope"], "project")
+        self.assertIn("做登录页", memories[0]["content"])
+        # Second sweep: no duplicate, same memory reused.
+        results2 = process_due_tasks(self.db_path)
+        self.assertEqual(results2[0]["reason"], "already_extracted")
+        self.assertEqual(len(self._memories()), 1)
+
+    def test_secret_in_result_is_redacted(self):
+        _insert_completed(self.db_path, "t1",
+                          result="完成，api_key=should-not-leak " + "y" * 300)
+        process_due_tasks(self.db_path)
+        self.assertNotIn("should-not-leak", self._memories()[0]["content"])
+
+    def test_idle_gate_and_force(self):
+        now = time.time()
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO tasks(id, project_id, prompt, status,"
+                " idempotency_key, depends_on, priority, created_by,"
+                " created_at, updated_at, last_heartbeat, started_at,"
+                " finished_at, result_text)"
+                " VALUES('t1', 'p1', '做登录页', 'completed', NULL,"
+                " '[]', 0, 'u1', ?, ?, NULL, ?, ?, ?)",
+                (now, now, now - 120, now, "z" * 300))
+            conn.commit()
+        finally:
+            conn.close()
+        results = process_due_tasks(self.db_path)
+        self.assertEqual(results[0]["reason"], "idle_not_reached")
+        self.assertEqual(self._memories(), [])
+        results = process_due_tasks(self.db_path, force=True)
+        self.assertEqual(results[0]["status"], "extracted")
+
+    def test_too_short_skipped(self):
+        _insert_completed(self.db_path, "t1", duration=5)
+        results = process_due_tasks(self.db_path)
+        self.assertEqual(results[0]["reason"], "too_short")
+        self.assertEqual(self._memories(), [])
+
+    def test_generate_disabled_skipped(self):
+        store = AutoMemoryStore(self.db_path)
+        store.upsert_settings("u1", generate_memories=False)
+        store.close()
+        _insert_completed(self.db_path, "t1")
+        results = process_due_tasks(self.db_path)
+        self.assertEqual(results[0]["reason"], "generate_disabled")
+        self.assertEqual(self._memories(), [])
+
+    def test_external_context_gate(self):
+        _insert_completed(self.db_path, "t1")
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "CREATE TABLE IF NOT EXISTS task_followups("
+                "id TEXT PRIMARY KEY, task_id TEXT, project_id TEXT,"
+                " body TEXT, status TEXT, created_at REAL,"
+                " sender_task_id TEXT)")
+            conn.execute(
+                "INSERT INTO task_followups(id, task_id, project_id,"
+                " body, status, created_at, sender_task_id)"
+                " VALUES('f1', 't1', 'p1', 'hi', 'done', ?, 'other')",
+                (time.time(),))
+            conn.commit()
+        finally:
+            conn.close()
+        results = process_due_tasks(self.db_path)
+        self.assertEqual(results[0]["reason"], "external_context")
+        store = AutoMemoryStore(self.db_path)
+        store.upsert_settings("u1", disable_on_external_context=False)
+        store.close()
+        results = process_due_tasks(self.db_path)
+        self.assertEqual(results[0]["status"], "extracted")
+
+    def test_quota_gate(self):
+        import datetime
+        month = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m")
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "UPDATE user_quotas SET quota_usd_micros = 10000000"
+                " WHERE user_id = 'u1'")
+            conn.execute(
+                "CREATE TABLE IF NOT EXISTS cost_records("
+                "id INTEGER PRIMARY KEY, task_id TEXT, project_id TEXT,"
+                " model TEXT, input_tokens INTEGER, output_tokens INTEGER,"
+                " cached_input_tokens INTEGER, total_tokens INTEGER,"
+                " usd_micros INTEGER, source TEXT, fallback_count INTEGER,"
+                " day TEXT, created_at REAL)")
+            conn.execute(
+                "INSERT INTO cost_records(task_id, project_id, model,"
+                " input_tokens, output_tokens, cached_input_tokens,"
+                " total_tokens, usd_micros, source, fallback_count, day,"
+                " created_at) VALUES('t1', 'p1', 'm', 0, 0, 0, 0, 9500000,"
+                " 'actual', 0, ?, ?)", (month + "-01", time.time()))
+            conn.commit()
+        finally:
+            conn.close()
+        store = AutoMemoryStore(self.db_path)
+        store.upsert_settings(
+            "u1", min_rate_limit_remaining_percent=20)
+        store.close()
+        _insert_completed(self.db_path, "t1")
+        results = process_due_tasks(self.db_path)
+        self.assertEqual(results[0]["reason"], "quota_low")
+        self.assertEqual(self._memories(), [])
+
+    def test_same_goal_consolidates_instead_of_duplicating(self):
+        _insert_completed(self.db_path, "t1", result="第一版完成。" + "a" * 300)
+        process_due_tasks(self.db_path)
+        self.assertEqual(len(self._memories()), 1)
+        _insert_completed(self.db_path, "t2", result="第二版完成。" + "b" * 300)
+        results = process_due_tasks(self.db_path)
+        extracted = [r for r in results if r["task_id"] == "t2"][0]
+        self.assertEqual(extracted["reason"], "consolidated")
+        memories = self._memories()
+        self.assertEqual(len(memories), 1)
+        self.assertIn("第二版完成", memories[0]["content"])
+
+
+class InjectionTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        init_db(self.db_path)
+        identity = IdentityStore(self.db_path)
+        identity.add_memory("u1", "project", "convention", "提交前先跑测试",
+                            project_id="p1")
+        identity.conn.close()
+        store = AutoMemoryStore(self.db_path)
+        store.upsert_settings("u1", use_memories=True)
+        store.close()
+        self.env = {"XCM_API_KEY": "test-placeholder"}
+        self.calls = []
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _completed_runner(self, cmd):
+        if "dispatch" in cmd:
+            name = cmd[cmd.index("--name") + 1]
+        else:
+            name = cmd[cmd.index("--task") + 1]
+        return SimpleNamespace(
+            returncode=0,
+            stdout=json.dumps({"task": name, "status": "completed",
+                               "text": "done"}),
+            stderr="")
+
+    def _recording_runner(self, inner):
+        def fake(cmd):
+            self.calls.append(list(cmd))
+            return inner(cmd)
+        return fake
+
+    def _insert_queued(self, task_id, prompt="做新功能", owner="u1"):
+        now = time.time()
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO tasks(id, project_id, prompt, status,"
+                " idempotency_key, depends_on, priority, created_by,"
+                " created_at, updated_at, last_heartbeat)"
+                " VALUES(?, 'p1', ?, 'queued', NULL, '[]', 0, ?, ?, ?,"
+                " NULL)",
+                (task_id, prompt, owner, now, now),
+            )
+            conn.commit()
+        finally:
+            conn.close()
+
+    def test_worker_prepends_memory_to_first_round_only(self):
+        self._insert_queued("n1")
+        worker = Worker(self.db_path, "/fake/host.py",
+                        runner=self._recording_runner(self._completed_runner),
+                        env=self.env)
+        results = worker.run_all()
+        self.assertEqual([r["status"] for r in results], ["completed"])
+        dispatch = [c for c in self.calls if "dispatch" in c][0]
+        task_arg = dispatch[dispatch.index("--task") + 1]
+        self.assertIn("提交前先跑测试", task_arg)
+        self.assertIn("做新功能", task_arg)
+        # DB prompt is unchanged: injection only touched the copy.
+        conn = sqlite3.connect(self.db_path)
+        try:
+            row = conn.execute(
+                "SELECT prompt FROM tasks WHERE id = 'n1'").fetchone()
+        finally:
+            conn.close()
+        self.assertEqual(row[0], "做新功能")
+
+    def test_injection_off_by_default(self):
+        store = AutoMemoryStore(self.db_path)
+        store.upsert_settings("u1", use_memories=False)
+        store.close()
+        self.assertEqual(
+            build_injection_prefix(self.db_path, "u1", "p1"), "")
+
+
+class EndpointsTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        status, self.project = self._req(
+            "POST", "/projects", {"name": "demo"}, self.owner_key)
+        self.assertEqual(status, 201)
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _req(self, method, path, body=None, key=None):
+        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
+        hdrs = {}
+        payload = None
+        if body is not None:
+            payload = json.dumps(body).encode("utf-8")
+            hdrs["Content-Type"] = "application/json"
+        if key is not None:
+            hdrs["X-API-Key"] = key
+        try:
+            conn.request(method, path, body=payload, headers=hdrs)
+            resp = conn.getresponse()
+            data = resp.read()
+        finally:
+            conn.close()
+        try:
+            parsed = json.loads(data.decode("utf-8")) if data else None
+        except ValueError:
+            parsed = None
+        return resp.status, parsed
+
+    def test_settings_get_put_and_validation(self):
+        status, body = self._req(
+            "GET", "/users/me/auto-memory-settings", key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(body["generate_memories"], 0)
+        status, body = self._req(
+            "PUT", "/users/me/auto-memory-settings",
+            {"generate_memories": True, "use_memories": True},
+            self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["generate_memories"], 1)
+        status, _ = self._req(
+            "PUT", "/users/me/auto-memory-settings",
+            {"idle_seconds": -5}, self.owner_key)
+        self.assertEqual(status, 400)
+
+    def test_run_and_log_and_task_flags(self):
+        status, body = self._req(
+            "POST", "/users/me/auto-memory/run", {"force": True},
+            self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["results"], [])
+        status, body = self._req(
+            "GET", "/users/me/auto-memory-log", key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(body["log"], [])
+        # Task flags: unknown task 404, real task 200.
+        status, _ = self._req(
+            "POST", "/tasks/nope/auto-memory-flags",
+            {"use_memories": False}, self.owner_key)
+        self.assertEqual(status, 404)
+        status, task = self._req(
+            "POST", "/tasks",
+            {"project_id": self.project["id"], "prompt": "做事"},
+            self.owner_key)
+        self.assertEqual(status, 201, task)
+        status, body = self._req(
+            "POST", "/tasks/%s/auto-memory-flags" % task["id"],
+            {"use_memories": False}, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["flags"]["use_memories"], 0)
+        status, _ = self._req(
+            "POST", "/tasks/%s/auto-memory-flags" % task["id"],
+            {}, self.owner_key)
+        self.assertEqual(status, 400)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_identity_memory.py

@@ -0,0 +1,259 @@
+"""Tests for cloud/identity.py (system-level identity + memory).
+
+Scope of this batch, stated so a future reader does not over-read
+it: cloud layer only. No test here touches host.py, CODEX_HOME or
+any Codex thread; the store and the HTTP endpoints only read/write
+the control-plane database.
+
+Run from the repo root:
+    python3 -m unittest discover -s cloud/tests -p "test_identity_memory.py" -v
+"""
+
+import http.client
+import json
+import os
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.db import POSTGRES_SCHEMA
+    from cloud.identity import IdentityStore
+    from cloud.migrations import MIGRATION_TABLES
+    from cloud.server import create_server
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.db import POSTGRES_SCHEMA
+    from cloud.identity import IdentityStore
+    from cloud.migrations import MIGRATION_TABLES
+    from cloud.server import create_server
+
+
+def _tmp_db():
+    fd, path = tempfile.mkstemp(suffix=".db")
+    os.close(fd)
+    os.unlink(path)
+    return path
+
+
+class IdentityStoreTest(unittest.TestCase):
+    def setUp(self):
+        self.db = _tmp_db()
+        self.store = IdentityStore(self.db)
+
+    def tearDown(self):
+        try:
+            self.store.conn.close()
+        finally:
+            if os.path.exists(self.db):
+                os.unlink(self.db)
+
+    def test_profile_partial_update_keeps_other_fields(self):
+        self.store.upsert_profile(
+            "u1", display_name="Alice", timezone="Asia/Shanghai",
+            preferences={"lang": "zh"})
+        got = self.store.upsert_profile("u1", display_name="Alicia")
+        self.assertEqual(got["display_name"], "Alicia")
+        self.assertEqual(got["timezone"], "Asia/Shanghai")
+        self.assertEqual(got["preferences"], {"lang": "zh"})
+
+    def test_profile_rejects_secret_preferences(self):
+        with self.assertRaises(ValueError):
+            self.store.upsert_profile("u1", preferences={"api_key": "x"})
+        with self.assertRaises(ValueError):
+            self.store.upsert_profile("u1", preferences={"token": "x"})
+
+    def test_user_memory_visible_only_to_owner(self):
+        self.store.add_memory("u1", "user", "preference", "喜欢简洁中文")
+        self.assertEqual(len(self.store.list_user_memories("u1")), 1)
+        self.assertEqual(self.store.list_user_memories("u2"), [])
+        self.assertNotIn("喜欢简洁中文",
+                         self.store.build_identity_context("u2"))
+
+    def test_scope_project_validation(self):
+        with self.assertRaises(ValueError):
+            self.store.add_memory("u1", "user", "fact", "x",
+                                  project_id="p1")
+        with self.assertRaises(ValueError):
+            self.store.add_memory("u1", "project", "fact", "x")
+        with self.assertRaises(ValueError):
+            self.store.add_memory("u1", "project", "nope", "x",
+                                  project_id="p1")
+        with self.assertRaises(ValueError):
+            self.store.add_memory("u1", "project", "fact", "",
+                                  project_id="p1")
+
+    def test_supersede_chain_hides_old_from_context(self):
+        old = self.store.add_memory("u1", "user", "preference", "旧偏好")
+        _old, new = self.store.supersede_memory(
+            old, "u1", replacement_content="新偏好")
+        self.assertIsNotNone(new)
+        ctx = self.store.build_identity_context("u1")
+        self.assertIn("新偏好", ctx)
+        self.assertNotIn("旧偏好", ctx)
+        row = self.store.get_memory(old)
+        self.assertEqual(row["status"], "superseded")
+        self.assertEqual(row["superseded_by"], new)
+
+    def test_supersede_user_memory_by_non_owner_forbidden(self):
+        mid = self.store.add_memory("u1", "user", "fact", "我的事实")
+        with self.assertRaises(PermissionError):
+            self.store.supersede_memory(mid, "u2")
+        with self.assertRaises(PermissionError):
+            self.store.delete_memory(mid, "u2")
+
+    def test_context_caps_and_project_section(self):
+        for i in range(7):
+            self.store.add_memory("u1", "user", "fact", "用户事实%d" % i)
+        self.store.add_memory("u1", "project", "convention", "项目约定",
+                              project_id="p1")
+        ctx = self.store.build_identity_context("u1", "p1")
+        self.assertIn("[用户记忆]", ctx)
+        self.assertIn("[项目记忆]", ctx)
+        self.assertIn("项目约定", ctx)
+        # Only the newest 5 user facts may appear.
+        self.assertNotIn("用户事实0", ctx)
+        self.assertNotIn("用户事实1", ctx)
+        self.assertIn("用户事实6", ctx)
+        # Without a project, no project section leaks in.
+        self.assertNotIn("[项目记忆]",
+                         self.store.build_identity_context("u1"))
+
+    def test_pg_schema_and_migration_lists_include_new_tables(self):
+        schema_text = "\n".join(POSTGRES_SCHEMA)
+        self.assertIn("CREATE TABLE IF NOT EXISTS user_profiles", schema_text)
+        self.assertIn("CREATE TABLE IF NOT EXISTS identity_memories",
+                      schema_text)
+        self.assertIn("user_profiles", MIGRATION_TABLES)
+        self.assertIn("identity_memories", MIGRATION_TABLES)
+
+
+class IdentityEndpointsTest(unittest.TestCase):
+    """HTTP-level: routes, self-only user scope, project permission."""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        self.viewer_id, self.viewer_key = self.auth.create_user("viewer")
+        self.outsider_id, self.outsider_key = self.auth.create_user("outsider")
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        status, self.project = self._req(
+            "POST", "/projects", {"name": "demo"}, self.owner_key)
+        self.assertEqual(status, 201)
+        self.auth.set_member(
+            self.project["id"], self.owner_id, self.viewer_id, "viewer")
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _req(self, method, path, body=None, key=None):
+        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
+        hdrs = {}
+        payload = None
+        if body is not None:
+            payload = json.dumps(body).encode("utf-8")
+            hdrs["Content-Type"] = "application/json"
+        if key is not None:
+            hdrs["X-API-Key"] = key
+        try:
+            conn.request(method, path, body=payload, headers=hdrs)
+            resp = conn.getresponse()
+            data = resp.read()
+        finally:
+            conn.close()
+        try:
+            parsed = json.loads(data.decode("utf-8")) if data else None
+        except ValueError:
+            parsed = None
+        return resp.status, parsed
+
+    def _pid(self):
+        return self.project["id"]
+
+    def test_profile_and_user_memory_roundtrip(self):
+        status, body = self._req(
+            "PUT", "/users/me/profile",
+            {"display_name": "Alice", "timezone": "Asia/Shanghai",
+             "preferences": {"lang": "zh"}}, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["display_name"], "Alice")
+        status, body = self._req(
+            "GET", "/users/me/profile", key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(body["preferences"], {"lang": "zh"})
+        status, body = self._req(
+            "POST", "/users/me/memories",
+            {"kind": "preference", "content": "喜欢简洁中文"},
+            self.owner_key)
+        self.assertEqual(status, 201, body)
+        status, body = self._req(
+            "GET", "/users/me/memories", key=self.viewer_key)
+        self.assertEqual(status, 200)
+        # Viewer asks for *their own* memories: owner's must not leak.
+        self.assertEqual(body["memories"], [])
+        status, body = self._req(
+            "GET", "/users/me/identity-context", key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertIn("喜欢简洁中文", body["identity_context"])
+
+    def test_project_memory_permission_and_outsider_403(self):
+        status, body = self._req(
+            "POST", "/projects/%s/identity-memories" % self._pid(),
+            {"kind": "convention", "content": "提交前先跑测试"},
+            self.owner_key)
+        self.assertEqual(status, 201, body)
+        status, body = self._req(
+            "GET", "/projects/%s/identity-memories" % self._pid(),
+            key=self.viewer_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(len(body["memories"]), 1)
+        status, _ = self._req(
+            "GET", "/projects/%s/identity-memories" % self._pid(),
+            key=self.outsider_key)
+        self.assertEqual(status, 403)
+        status, _ = self._req(
+            "GET", "/users/me/identity-context?project_id=%s" % self._pid(),
+            key=self.outsider_key)
+        self.assertEqual(status, 403)
+
+    def test_user_memory_supersede_and_delete_via_http(self):
+        status, created = self._req(
+            "POST", "/users/me/memories",
+            {"kind": "fact", "content": "旧事实"}, self.owner_key)
+        self.assertEqual(status, 201)
+        mid = created["id"]
+        status, body = self._req(
+            "POST", "/users/me/memories/%s/supersede" % mid,
+            {"replacement_content": "新事实"}, self.viewer_key)
+        # Another user's memory: 404, not a leak-shaped 403.
+        self.assertEqual(status, 404)
+        status, body = self._req(
+            "POST", "/users/me/memories/%s/supersede" % mid,
+            {"replacement_content": "新事实"}, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["status"], "superseded")
+        status, body = self._req(
+            "DELETE", "/users/me/memories/%s" % body["replacement_id"],
+            key=self.owner_key)
+        self.assertEqual(status, 200)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/worker.py

@@ -145,6 +145,9 @@ from cloud.cost import CostStore, estimate_tokens, price_usd_micros, \
 from cloud.db import connect as _db_connect
 from cloud.logging_config import get_logger, log_event, setup_logging
 from cloud.models import ModelRouter, is_fallbackable, record_fallback
+from cloud.auto_memory import AutoMemoryStore
+from cloud.auto_memory import build_injection_prefix as _auto_prefix
+from cloud.auto_memory import process_due_tasks as _auto_process_due
 from cloud.peer_messages import deliver_peer_message as _deliver_peer_message
 from cloud.peer_messages import peer_message_prefix as _peer_prefix
 from cloud.peer_messages import row_get as _row_get
@@ -977,6 +980,78 @@ class Worker:
                       status=event_type,
                       message="notify exception: %s" % exc)
 
+    # -- 自动记忆 ------------------------------------------------------
+    # 注入失败绝不挡任务（fail-open：无记忆跑也比不跑强）；提取
+    # 失败只记日志（run_loop 轮次边界已包一层，见 process 方法）。
+
+    def _apply_auto_memory_injection(self, task):
+        """首轮注入记忆：只改内存副本，原 prompt/goal 不落库。
+
+        跳过：协调员任务（不跑模型回合）、续跑与断点续跑（已有
+        会话历史）、无属主、use_memories 关、无记忆。goal 任务
+        同时改内存里的 goal（执行器取 goal 优先于 prompt），
+        前缀里写明“仅供参考，不得改写本任务目标”。
+        """
+        if task.get("coordinator_parts"):
+            return task
+        if task.get("_resume") or task.get("_goal_resume_breakpoint"):
+            return task
+        owner = task.get("created_by")
+        if not owner:
+            return task
+        try:
+            store = AutoMemoryStore(self.db_path)
+            try:
+                flags = store.get_task_flags(task["id"])
+            finally:
+                store.close()
+            prefix = _auto_prefix(
+                self.db_path, owner, task["project_id"], flags=flags)
+        except Exception as exc:
+            try:
+                self.logger.warning("auto memory injection failed: %s",
+                                    exc)
+            except Exception:
+                pass
+            return task
+        if not prefix:
+            return task
+        new_task = dict(task)
+        prompt = task.get("prompt") or ""
+        new_task["prompt"] = prefix + "\n\n" + prompt
+        if task.get("goal_mode") or task.get("verify_command"):
+            new_task["goal"] = prefix + "\n\n" + (
+                task.get("goal") or prompt)
+        new_task["_memory_injected"] = True
+        self._audit(task["project_id"], "worker_memory_injected",
+                    task["id"])
+        return new_task
+
+    def process_auto_memories(self, now=None, force=False,
+                              owner_id=None):
+        """扫 completed 任务并提取记忆（run_loop 轮次边界调用）。
+
+        返回 process_due_tasks 的结果列表；提取本身失败时返回
+        空列表并只记日志，不炸循环。提取成功的任务各补一条
+        worker 侧审计（服务端手动触发路径另有自己的审计）。
+        """
+        try:
+            results = _auto_process_due(
+                self.db_path, now=now, force=force, owner_id=owner_id)
+        except Exception as exc:
+            try:
+                self.logger.warning("auto memory sweep failed: %s", exc)
+            except Exception:
+                pass
+            return []
+        for item in results:
+            if item.get("status") == "extracted":
+                # project_id 不在结果里：审计 detail 只带任务 id，
+                # project 字段用空串（与系统审计同款占位）。
+                self._audit("", "worker_memory_extracted",
+                            item.get("task_id") or "")
+        return results
+
     # -- 认领 ----------------------------------------------------------
     def _budget_exceeded(self):
         return budget_exceeded(self.usage_turns, self.max_turns,
@@ -3158,6 +3233,9 @@ class Worker:
             else:
                 task["prompt"] = resume_message
                 task["_resume"] = True
+        # 自动记忆注入（官方 use_memories）：首轮才注入，见
+        # _apply_auto_memory_injection 的跳过规则；只改内存副本。
+        task = self._apply_auto_memory_injection(task)
         self._audit(task["project_id"], "worker_claim",
                     "%s executor=%s" % (task_id, self._executor_name()))
         # log_event 只接受 task_id/project_id/status/message 四个
@@ -4122,6 +4200,9 @@ class Worker:
                 # G11/G12 后台投递：摘要（每小时）与 outbox 重试，
                 # 轮次边界做，不中断在跑任务，失败只记日志。
                 self._process_push_due()
+                # 自动记忆提取：同款轮次边界后台扫，失败只记日志
+                # （process_auto_memories 内部已兜底，不炸循环）。
+                self.process_auto_memories()
                 # 自动陈旧回收：死 worker 的 running 任务在这里被
                 # 标 interrupted；活任务心跳新鲜，不会被扫到。
                 if sweep_threshold is not None:

docs/auto-memory.md

@@ -0,0 +1,93 @@
+# 自动记忆（官方 Codex memories 的云端对齐版，v1）
+
+官方功能见 learn.chatgpt.com/docs/customization/memories：开启后，
+Codex 把先前聊天的有用上下文转成记忆，供未来工作用；后台生成、
+闲置后才生成、剩余额度低时跳过、有 generate / use /
+disable_on_external_context 等设置、密钥脱敏。
+
+本批把这套行为搬到云端控制面（`cloud/auto_memory.py`），不改
+host.py、不改 CODEX_HOME、不改 thread 的创建/续跑/fork/归档。
+云端记忆与官方本地记忆的区别要先说清：官方记忆存在单机
+`~/.codex/memories/` 里、跟一台机器走；云端记忆存在控制面
+数据库里，跟用户/项目走，换 worker、换执行器都还在。
+
+## 开关（默认全关，与官方“本地记忆默认关”一致）
+
+- `GET/PUT /users/me/auto-memory-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`：只存设置，v1 不调
+    模型（见“未做”）。
+- `POST /tasks/{id}/auto-memory-flags`：单个任务覆盖 use/generate
+  （NULL = 跟随用户设置），权限为项目 create_task。不改 tasks
+  表结构，建任务的各条 INSERT 路径都不用动。
+- `POST /users/me/auto-memory/run`：手动触发本人任务的提取扫荡
+  （body 可带 `force: true` 跳过闲置/时长/长度门，不跳过开关、
+  外部上下文门与额度门）；`GET /users/me/auto-memory-log`：看
+  本人任务的提取日志（按任务主键覆盖，不无限增长）。
+
+## 提取：什么时候、从哪来、写到哪
+
+触发：worker `run_loop` 每轮结束调 `process_auto_memories()`；
+也可经上面的手动触发接口立即跑。只扫 `status='completed'` 的
+任务（failed / interrupted / merged 不提取）。
+
+资格（任一不满足即跳过并记日志，下次仍可再扫）：
+1. generate 开关（任务级覆盖优先于用户设置）；
+2. 终态闲置够 `idle_seconds`（终态后仍可被 /continue 续跑，
+   所以“刚完成”不等于“已闲置”；`force` 可跳过）；
+3. 有 started_at/finished_at 且时长 ≥ `min_duration_seconds`，
+   result_text 长度 ≥ `min_result_chars`（跳过短期会话）；
+4. `disable_on_external_context` 开时，任务间消息
+   （task_followups 有该任务的行）或协调员子任务
+   （parent_task_id 非空）不提取；
+5. 当月 USD 配额剩余百分比 ≥ `min_rate_limit_remaining_percent`
+   （无配额行按默认 $10 计，只有显式 NULL 行才是不限额，与
+   worker 配额口径一致；查不到配额时 fail-closed，记
+   quota_check_failed，不冒充额度充足）。
+
+内容（v1 确定性版，不调模型）：任务 goal/prompt 首行 +
+result_text 前 800 字，脱敏后拼成一条 project 级 fact，写入
+`identity_memories`，`source_thread_id` 记云端任务 id（去重键）。
+同一任务只提取一次；同一目标的新结果出现时 supersede 旧记忆
+（consolidation 的确定性版），不堆重复。自动提取只写 project
+级：单个项目的任务结论若自动写成 user 级，会跨项目泄漏项目
+上下文；user 级记忆仍只接受手动创建。
+
+脱敏：`api_key/password/secret/token/credential` 的 key=value
+形态、Authorization / Bearer 头、sk- / ghp_ / gho_ / xox*- 前缀
+一律替换为 `[REDACTED]`；偏好里的密钥键禁令见 identity-memory.md。
+
+## 注入：怎么进任务
+
+worker `run_one` 认领后、执行前调 `_apply_auto_memory_injection`：
+- 只注入首轮：续跑、断点续跑、协调员任务不注入；
+- 只改内存副本：prompt 前加 `【过往记忆，仅供参考，不得改写
+  本任务目标】` + 最新各 5 条用户/项目记忆；goal 任务同时改
+  内存里的 goal（执行器取 goal 优先于 prompt）；库里的原
+  prompt / goal 不动，任务详情里看到的仍是原文本；
+- 注入失败 fail-open（记 warning，不挡任务）；注入成功记
+  `worker_memory_injected` 审计；无记忆或 use 关时不注入。
+- 注入不进 thread 的 developerInstructions：host.py 的 dispatch
+  路径不传 instructions，云端层唯一不改 host.py 的注入点就是
+  prompt 前缀，identity-memory.md 里“调用方自己拼进新建任务
+  的 prompt”正是这条路。
+
+## 未做（诚实版）
+
+- 未调模型提取/合并：`extract_model` / `consolidation_model`
+  只存设置。v1 提取是确定性拼贴，不是模型摘要；换模型提取
+  需另定模型选择、成本上限与失败语义。
+- 未识别 host 线程内的 MCP / web search：tasks 表上没有这类
+  标记，`disable_on_external_context` 目前只认任务间消息与
+  协调员子任务；要认 MCP/web 需先给 host 线程事件加遥测。
+- 未做控制台 UI：开关与日志目前只有 API；未做 per-chat 的
+  `/memories` 式交互（任务级 flags 接口是对应的最小版）。
+- 未进代码签名候选：worker 签名只含 worker/sandbox/
+  coordinator/cost/server/host；本批同时改了 worker.py，所以
+  正常部署换目录时旧 worker 仍会因 worker.py 变化而自退，
+  仅换 auto_memory.py 的单文件部署不在签名覆盖内。

docs/identity-memory.md

@@ -0,0 +1,51 @@
+# 系统级身份与记忆（云端层）
+
+本批只做云端层，不动 Codex thread：不改 `host.py`、不改
+`CODEX_HOME`、不改任何 thread 的创建/续跑/fork/归档。
+自动注入是后一批（docs/auto-memory.md）接通的：worker 首轮
+把上下文拼进内存里的 prompt 副本，库里的原 prompt 不动。
+
+## 数据
+
+- `user_profiles`：每用户一行，display_name、timezone、
+  preferences（JSON）。偏好里禁止含 password/secret/token/
+  api_key/credential 的键，绝不能拿它当第二个凭证库。
+- `identity_memories`：分级记忆，TEXT 主键（不走 IDENTITY
+  序列，见 `cloud/migrations.py` 对齐说明）。
+  - `user`：只归属用户一人可见，跨项目；不得带 project_id。
+  - `project`：归属用户创建、项目成员可见；读/写权限由
+    server 的项目 `read` / `create_task` 门先卡，store 层
+    不认项目成员关系。
+  - `kind`：fact / preference / decision / convention。
+  - `source_thread_id`：仅作来源标注，不建立 thread 依赖。
+- 旧项目记忆 `summaries` 表不动；`identity-context` 返回的
+  项目摘要上下文是独立字段，不与身份上下文合并。
+
+## 语义
+
+- 修正用 supersede 链：旧行变 `superseded` 并指向新行，
+  旧内容不再进上下文，但可审计；删除是物理删除，仅用于
+  用户主动遗忘。
+- 上下文只取最新 active 各 5 条，不全量倒历史。
+- 他人用户级记忆在 HTTP 层回 404（不是 403），避免用
+  状态码探出记忆存在。
+
+## 接口
+
+- `GET/PUT /users/me/profile`（本人）
+- `GET/POST /users/me/memories`、`POST /users/me/memories/{id}/supersede`、`DELETE /users/me/memories/{id}`（本人）
+- `GET/POST /projects/{id}/identity-memories`、`POST /projects/{id}/identity-memories/{id}/supersede`、`DELETE /projects/{id}/identity-memories/{id}`（项目门）
+- `GET /users/me/identity-context?project_id=...`：显式取用，
+  返回 profile、两级记忆、身份上下文、项目摘要上下文四个
+  独立字段；调用方要把上下文放进新任务时，应自己拼进
+  新建任务的 prompt 并留痕，本接口不代劳。
+
+## 未做
+
+- 自动提取与注入已由 docs/auto-memory.md 一批接通（worker
+  首轮 prompt 前缀注入 + run_loop 轮次边界提取扫荡）：注入点
+  定为首轮 prompt（host.py dispatch 不传 developerInstructions，
+  云端层不改 host.py 的唯一注入点就是 prompt 前缀），去重与
+  合并用确定性 supersede，成本上限用配额剩余百分比门。
+- 未做控制台 UI；未做跨引擎记忆（本批记忆与 Codex
+  thread 无关，换引擎时云端记忆仍可用）。