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 无关，换引擎时云端记忆仍可用）。