From 6e9d6dc5e81fba8f4744d972cc35e56b5f3fbece Mon Sep 17 00:00:00 2001 From: host Date: Wed, 28 Oct 2026 03:43:02 +0800 Subject: [PATCH 1/2] =?UTF-8?q?=E4=BA=91=E7=AB=AF=E8=BA=AB=E4=BB=BD?= =?UTF-8?q?=E4=B8=8E=E8=AE=B0=E5=BF=86=EF=BC=9A=E7=94=A8=E6=88=B7=E7=94=BB?= =?UTF-8?q?=E5=83=8F+=E5=88=86=E7=BA=A7=E8=AE=B0=E5=BF=86=EF=BC=8C?= =?UTF-8?q?=E4=B8=8D=E5=8A=A8=20thread?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 cloud/identity.py：user_profiles（显示名/时区/偏好，偏好禁密钥键） 与 identity_memories（user 级仅本人可见、project 级项目成员可见， kind fact/preference/decision/convention，supersede 链+物理删除）。 - server 新增 /users/me/profile、/users/me/memories、/projects/{id}/ identity-memories、/users/me/identity-context，项目记忆先过项目 read/create_task 门；上下文只显式取用，不自动注入任务/thread。 - PG schema 与迁移清单同步新增两张 TEXT 主键表，序列清单不动； server 补 do_PUT 入口（/users/me/profile 首个 PUT 路由）。 - 新测试 11 个；三套件 cloud 937 OK（4 跳过）、host 89 OK、deploy 29 OK。 - 未动：host.py、worker、CODEX_HOME、旧 summaries 表、控制台 UI； 未推、未建 PR、未部署。 --- cloud/db.py | 21 ++ cloud/identity.py | 301 ++++++++++++++++++++++++++ cloud/migrations.py | 3 + cloud/server.py | 316 ++++++++++++++++++++++++++++ cloud/tests/test_identity_memory.py | 259 +++++++++++++++++++++++ docs/identity-memory.md | 49 +++++ 6 files changed, 949 insertions(+) create mode 100644 cloud/identity.py create mode 100644 cloud/tests/test_identity_memory.py create mode 100644 docs/identity-memory.md diff --git a/cloud/db.py b/cloud/db.py index 80345c6..608d7e4 100644 --- a/cloud/db.py +++ b/cloud/db.py @@ -945,6 +945,27 @@ 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)", ) diff --git a/cloud/identity.py b/cloud/identity.py new file mode 100644 index 0000000..ed1aec7 --- /dev/null +++ b/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) diff --git a/cloud/migrations.py b/cloud/migrations.py index 468de77..4a6d2de 100644 --- a/cloud/migrations.py +++ b/cloud/migrations.py @@ -68,6 +68,9 @@ MIGRATION_TABLES = ( "task_attachments", "task_followups", "environments", + # Cloud identity batch: TEXT primary keys, no sequence to sync. + "user_profiles", + "identity_memories", ) _SKIP_TABLES = frozenset({"kb_fts", "sqlite_sequence"}) diff --git a/cloud/server.py b/cloud/server.py index 94af7a0..6ab199e 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -82,6 +82,7 @@ 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.identity import IdentityStore from cloud.kb import KBStore from cloud.memory import MemoryStore from cloud.pr import PrSkipped, PrStepError @@ -1591,6 +1592,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 死锁事件就是这么变成前端 @@ -1806,6 +1812,25 @@ 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.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": @@ -2567,6 +2592,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": @@ -2863,6 +2930,255 @@ 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, + }) + + 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: diff --git a/cloud/tests/test_identity_memory.py b/cloud/tests/test_identity_memory.py new file mode 100644 index 0000000..7bf92ec --- /dev/null +++ b/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() diff --git a/docs/identity-memory.md b/docs/identity-memory.md new file mode 100644 index 0000000..eb2a014 --- /dev/null +++ b/docs/identity-memory.md @@ -0,0 +1,49 @@ +# 系统级身份与记忆（云端层） + +本批只做云端层，不动 Codex thread：不改 `host.py`、不改 +`CODEX_HOME`、不改任何 thread 的创建/续跑/fork/归档， +也不向任何任务或 thread 自动注入记忆。 + +## 数据 + +- `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 并留痕，本接口不代劳。 + +## 未做 + +- 未接 worker：任务创建时不会自动拼身份上下文，thread + 行为与升级前一致；自动注入是下一步的显式决策，需另 + 定注入点（新建 thread 的 developerInstructions 还是首轮 + prompt）、去重与成本上限。 +- 未做控制台 UI；未做跨引擎记忆（本批记忆与 Codex + thread 无关，换引擎时云端记忆仍可用）。 From cbc90f9d305a6d5dbc2a793c7c8c2ffc66963e5c Mon Sep 17 00:00:00 2001 From: host Date: Wed, 28 Oct 2026 12:12:10 +0800 Subject: [PATCH 2/2] =?UTF-8?q?=E8=87=AA=E5=8A=A8=E8=AE=B0=E5=BF=86?= =?UTF-8?q?=EF=BC=9A=E5=AE=98=E6=96=B9=20memories=20=E5=AF=B9=E9=BD=90?= =?UTF-8?q?=E7=9A=84=E6=8F=90=E5=8F=96+=E6=B3=A8=E5=85=A5=E9=97=AD?= =?UTF-8?q?=E7=8E=AF=EF=BC=88=E7=A1=AE=E5=AE=9A=E6=80=A7=20v1=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 开关默认关（generate/use），设置接口 GET/PUT /users/me/auto-memory-settings，任务级覆盖 POST /tasks/{id}/auto-memory-flags，手动触发 POST /users/me/auto-memory/run，日志 GET /users/me/auto-memory-log。 提取：worker run_loop 轮次边界扫 completed 任务，闲置/时长/长度/外部上下文/配额剩余百分比五道门，脱敏后写 project 级 identity_memories，同一任务去重、同一目标 supersede 合并；只写 project 级，不跨项目泄漏。 注入：worker run_one 首轮把最新各 5 条用户/项目记忆拼进内存 prompt 副本（goal 任务同步改内存 goal），库原 prompt 不动；续跑/断点/协调员不注入，注入失败 fail-open。 v1 诚实边界：extract/consolidation 模型只存设置不调模型；MCP/web 线程内调用暂无 tasks 表标记，外部上下文门只认任务间消息与协调员子任务；无控制台 UI。 --- cloud/auto_memory.py | 550 ++++++++++++++++++++++++++++++++ cloud/db.py | 28 ++ cloud/migrations.py | 4 + cloud/server.py | 124 +++++++ cloud/tests/test_auto_memory.py | 482 ++++++++++++++++++++++++++++ cloud/worker.py | 81 +++++ docs/auto-memory.md | 93 ++++++ docs/identity-memory.md | 14 +- 8 files changed, 1370 insertions(+), 6 deletions(-) create mode 100644 cloud/auto_memory.py create mode 100644 cloud/tests/test_auto_memory.py create mode 100644 docs/auto-memory.md diff --git a/cloud/auto_memory.py b/cloud/auto_memory.py new file mode 100644 index 0000000..9e4f6cd --- /dev/null +++ b/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 diff --git a/cloud/db.py b/cloud/db.py index 54a75f7..9b08d57 100644 --- a/cloud/db.py +++ b/cloud/db.py @@ -967,6 +967,34 @@ POSTGRES_SCHEMA = ( "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)", ) diff --git a/cloud/migrations.py b/cloud/migrations.py index b82d3b6..c7a0cd1 100644 --- a/cloud/migrations.py +++ b/cloud/migrations.py @@ -71,6 +71,10 @@ MIGRATION_TABLES = ( # 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"}) diff --git a/cloud/server.py b/cloud/server.py index e286a7f..9b39570 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -110,6 +110,8 @@ 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 @@ -1895,6 +1897,15 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): 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": @@ -1943,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": @@ -3268,6 +3282,116 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): "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: diff --git a/cloud/tests/test_auto_memory.py b/cloud/tests/test_auto_memory.py new file mode 100644 index 0000000..7df4871 --- /dev/null +++ b/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() diff --git a/cloud/worker.py b/cloud/worker.py index 8076283..b3dfaa9 100644 --- a/cloud/worker.py +++ b/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: diff --git a/docs/auto-memory.md b/docs/auto-memory.md new file mode 100644 index 0000000..eec4057 --- /dev/null +++ b/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 的单文件部署不在签名覆盖内。 diff --git a/docs/identity-memory.md b/docs/identity-memory.md index eb2a014..ec5b438 100644 --- a/docs/identity-memory.md +++ b/docs/identity-memory.md @@ -1,8 +1,9 @@ # 系统级身份与记忆（云端层） 本批只做云端层，不动 Codex thread：不改 `host.py`、不改 -`CODEX_HOME`、不改任何 thread 的创建/续跑/fork/归档， -也不向任何任务或 thread 自动注入记忆。 +`CODEX_HOME`、不改任何 thread 的创建/续跑/fork/归档。 +自动注入是后一批（docs/auto-memory.md）接通的：worker 首轮 +把上下文拼进内存里的 prompt 副本，库里的原 prompt 不动。 ## 数据 @@ -41,9 +42,10 @@ ## 未做 -- 未接 worker：任务创建时不会自动拼身份上下文，thread - 行为与升级前一致；自动注入是下一步的显式决策，需另 - 定注入点（新建 thread 的 developerInstructions 还是首轮 - prompt）、去重与成本上限。 +- 自动提取与注入已由 docs/auto-memory.md 一批接通（worker + 首轮 prompt 前缀注入 + run_loop 轮次边界提取扫荡）：注入点 + 定为首轮 prompt（host.py dispatch 不传 developerInstructions， + 云端层不改 host.py 的唯一注入点就是 prompt 前缀），去重与 + 合并用确定性 supersede，成本上限用配额剩余百分比门。 - 未做控制台 UI；未做跨引擎记忆（本批记忆与 Codex thread 无关，换引擎时云端记忆仍可用）。