Add personal memory and admin dashboard

backend/api/server.py

@@ -8,6 +8,7 @@ from __future__ import annotations
 
 import asyncio
 import csv
+import hashlib
 import json
 import os
 import queue
@@ -48,6 +49,11 @@ from src.jupyter_runtime import (
 )
 from src.mcp_runtime import MCPRuntime, MCPServerProfile
 from src.openai_compat import OpenAICompatClient, OpenAICompatError
+from src.personal_memory import (
+    SKILL_MEMORY_DIRNAME,
+    USER_MEMORY_FILENAME,
+    PersonalMemoryManager,
+)
 from src.session_store import (
     DEFAULT_AGENT_SESSION_DIR,
     StoredAgentSession,
@@ -514,6 +520,11 @@ class AgentState:
             allow_shell=allow_shell,
             allow_write=allow_write,
         )
+        self.memory_manager = PersonalMemoryManager(
+            self.session_directory.parent / 'accounts',
+            self.config_for,
+        )
+        self.memory_manager.start()
 
     @property
     def cwd(self) -> Path:
@@ -582,6 +593,7 @@ class AgentState:
                 'uploads': self.session_directory,
                 'outputs': self.session_directory,
                 'python_env': base / 'python' / '.venv',
+                'memory': base / 'memory',
             }
         base = self._account_base(account_id)
         return {
@@ -591,6 +603,7 @@ class AgentState:
             'uploads': base / 'sessions',
             'outputs': base / 'sessions',
             'python_env': base / 'python' / '.venv',
+            'memory': base / 'memory',
         }
 
     def _skill_settings_path(self, account_id: str | None) -> Path:
@@ -955,6 +968,21 @@ class FeishuOnlineDocRequest(BaseModel):
     account_id: str | None = None
 
 
+class AdminLoginRequest(BaseModel):
+    username: str = Field(min_length=1)
+    password: str = Field(min_length=1)
+
+
+class AdminAccountCreateRequest(BaseModel):
+    account_id: str = Field(min_length=2)
+
+
+class MemoryUpdateRequest(BaseModel):
+    account_id: str = Field(min_length=1)
+    content: str
+    skill: str | None = None
+
+
 def _append_runtime_context(current: str | None, addition: str) -> str:
     parts = [part.strip() for part in (current, addition) if part and part.strip()]
     return '\n\n'.join(parts)
@@ -1163,6 +1191,158 @@ def create_app(state: AgentState) -> FastAPI:
         )
         return result
 
+    # ------------- memory ----------------------------------------------------
+    @app.get('/api/memory/user')
+    async def get_user_memory(account_id: str) -> dict[str, Any]:
+        try:
+            return state.memory_manager.list_user_memory(_safe_account_id(account_id))
+        except Exception as exc:
+            raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+    @app.put('/api/memory/user')
+    async def put_user_memory(payload: MemoryUpdateRequest) -> dict[str, Any]:
+        try:
+            return state.memory_manager.update_user_memory(
+                _safe_account_id(payload.account_id),
+                payload.content,
+            )
+        except Exception as exc:
+            raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+    @app.get('/api/memory/skills')
+    async def get_skill_memories(account_id: str) -> list[dict[str, Any]]:
+        try:
+            return state.memory_manager.list_skill_memories(_safe_account_id(account_id))
+        except Exception as exc:
+            raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+    @app.get('/api/memory/skills/{skill_name}')
+    async def get_skill_memory(skill_name: str, account_id: str) -> dict[str, Any]:
+        try:
+            return state.memory_manager.read_skill_memory(
+                _safe_account_id(account_id),
+                skill_name,
+            )
+        except Exception as exc:
+            raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+    @app.put('/api/memory/skills/{skill_name}')
+    async def put_skill_memory(
+        skill_name: str,
+        payload: MemoryUpdateRequest,
+    ) -> dict[str, Any]:
+        try:
+            return state.memory_manager.update_skill_memory(
+                _safe_account_id(payload.account_id),
+                skill_name,
+                payload.content,
+            )
+        except Exception as exc:
+            raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+    @app.delete('/api/memory/skills/{skill_name}')
+    async def delete_skill_memory(skill_name: str, account_id: str) -> dict[str, Any]:
+        try:
+            return state.memory_manager.delete_skill_memory(
+                _safe_account_id(account_id),
+                skill_name,
+            )
+        except Exception as exc:
+            raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+    @app.get('/api/memory/queue')
+    async def get_memory_queue(account_id: str | None = None) -> dict[str, Any]:
+        return state.memory_manager.queue_snapshot(
+            _safe_account_id(account_id) if account_id else None
+        )
+
+    @app.get('/api/memory/events')
+    async def get_memory_events(
+        account_id: str,
+        limit: int = 50,
+    ) -> list[dict[str, Any]]:
+        return state.memory_manager.list_recent_events(
+            _safe_account_id(account_id),
+            limit=limit,
+        )
+
+    # ------------- admin -----------------------------------------------------
+    @app.post('/api/admin/login')
+    async def admin_login(payload: AdminLoginRequest) -> dict[str, Any]:
+        if payload.username == 'admin' and payload.password == 'admin':
+            return {'token': _admin_token(), 'username': 'admin'}
+        raise HTTPException(status_code=401, detail='管理员账号或密码错误')
+
+    @app.get('/api/admin/summary')
+    async def admin_summary(token: str) -> dict[str, Any]:
+        _require_admin_token(token)
+        return _build_admin_summary(state)
+
+    @app.get('/api/admin/accounts')
+    async def admin_accounts(token: str) -> list[dict[str, Any]]:
+        _require_admin_token(token)
+        return _list_admin_accounts(state)
+
+    @app.post('/api/admin/accounts')
+    async def admin_create_account(
+        token: str,
+        payload: AdminAccountCreateRequest,
+    ) -> dict[str, Any]:
+        _require_admin_token(token)
+        return _admin_create_account(state, payload.account_id)
+
+    @app.delete('/api/admin/accounts/{account_id}')
+    async def admin_delete_account(account_id: str, token: str) -> dict[str, Any]:
+        _require_admin_token(token)
+        return _admin_delete_account(state, account_id)
+
+    @app.get('/api/admin/memory')
+    async def admin_memory(token: str, account_id: str | None = None) -> dict[str, Any]:
+        _require_admin_token(token)
+        safe_account = _safe_account_id(account_id) if account_id else None
+        return {
+            'queue': state.memory_manager.queue_snapshot(safe_account),
+            'events': (
+                state.memory_manager.list_recent_events(safe_account, limit=100)
+                if safe_account
+                else []
+            ),
+        }
+
+    @app.get('/api/admin/memory/user')
+    async def admin_user_memory(token: str, account_id: str) -> dict[str, Any]:
+        _require_admin_token(token)
+        return state.memory_manager.list_user_memory(_safe_account_id(account_id))
+
+    @app.put('/api/admin/memory/user')
+    async def admin_update_user_memory(
+        token: str,
+        payload: MemoryUpdateRequest,
+    ) -> dict[str, Any]:
+        _require_admin_token(token)
+        return state.memory_manager.update_user_memory(
+            _safe_account_id(payload.account_id),
+            payload.content,
+        )
+
+    @app.get('/api/admin/memory/skills')
+    async def admin_skill_memories(token: str, account_id: str) -> list[dict[str, Any]]:
+        _require_admin_token(token)
+        return state.memory_manager.list_skill_memories(_safe_account_id(account_id))
+
+    @app.put('/api/admin/memory/skills/{skill_name}')
+    async def admin_update_skill_memory(
+        skill_name: str,
+        token: str,
+        payload: MemoryUpdateRequest,
+    ) -> dict[str, Any]:
+        _require_admin_token(token)
+        return state.memory_manager.update_skill_memory(
+            _safe_account_id(payload.account_id),
+            skill_name,
+            payload.content,
+        )
+
     @app.get('/api/models')
     async def list_models_get(account_id: str | None = None) -> dict[str, Any]:
         config = state.config_for(account_id)
@@ -1450,6 +1630,12 @@ def create_app(state: AgentState) -> FastAPI:
             requested_session_id,
         )
         runtime_context = request.runtime_context
+        memory_context = state.memory_manager.render_injection(
+            request.account_id,
+            state.enabled_skill_names(request.account_id),
+        )
+        if memory_context:
+            runtime_context = _append_runtime_context(runtime_context, memory_context)
         if jupyter_runtime is not None:
             runtime_context = _append_runtime_context(
                 runtime_context,
@@ -1575,6 +1761,9 @@ def create_app(state: AgentState) -> FastAPI:
             elapsed_ms = max(0, int((time.perf_counter() - started_at) * 1000))
             payload = _serialize_run_result(result)
             payload['elapsed_ms'] = elapsed_ms
+            used_skills = _extract_used_skill_names(payload.get('transcript') or ())
+            if not used_skills:
+                used_skills = tuple(state.enabled_skill_names(request.account_id))
             if result.session_id:
                 _annotate_last_assistant_elapsed(
                     session_directory,
@@ -1591,6 +1780,17 @@ def create_app(state: AgentState) -> FastAPI:
                     fallback_title=fallback_title,
                 )
                 _annotate_transcript_elapsed(payload, elapsed_ms)
+                try:
+                    state.memory_manager.enqueue_interaction(
+                        account_id=request.account_id,
+                        session_id=result.session_id,
+                        user_prompt=prompt,
+                        assistant_output=result.final_output,
+                        skills=used_skills,
+                        model=config.model,
+                    )
+                except Exception:
+                    pass
             state.run_manager.finish(
                 run_record.run_id,
                 'cancelled' if run_record.cancel_event.is_set() else 'completed',
@@ -3288,6 +3488,213 @@ def _safe_account_id(account_id: str | None) -> str:
     return normalized[:80] or 'default'
 
 
+def _admin_token() -> str:
+    return os.environ.get('ZK_ADMIN_TOKEN') or 'admin'
+
+
+def _require_admin_token(token: str) -> None:
+    if token != _admin_token():
+        raise HTTPException(status_code=401, detail='Admin unauthorized')
+
+
+def _accounts_root(state: AgentState) -> Path:
+    return state.session_directory.parent / 'accounts'
+
+
+def _users_json_path(state: AgentState) -> Path:
+    return _accounts_root(state) / 'users.json'
+
+
+def _auth_sessions_json_path(state: AgentState) -> Path:
+    return _accounts_root(state) / 'auth_sessions.json'
+
+
+def _load_users_file(state: AgentState) -> dict[str, Any]:
+    path = _users_json_path(state)
+    try:
+        payload = json.loads(path.read_text(encoding='utf-8'))
+    except (OSError, json.JSONDecodeError):
+        payload = {'users': []}
+    if not isinstance(payload, dict) or not isinstance(payload.get('users'), list):
+        return {'users': []}
+    return payload
+
+
+def _save_users_file(state: AgentState, payload: dict[str, Any]) -> None:
+    path = _users_json_path(state)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
+
+
+def _hash_admin_created_password(password: str = '123456') -> str:
+    salt = os.urandom(16).hex()
+    digest = hashlib.scrypt(
+        password.encode('utf-8'),
+        salt=salt.encode('utf-8'),
+        n=16384,
+        r=8,
+        p=1,
+        dklen=64,
+    ).hex()
+    return f'{salt}:{digest}'
+
+
+def _build_admin_summary(state: AgentState) -> dict[str, Any]:
+    accounts = _list_admin_accounts(state)
+    totals = {
+        'accounts': len(accounts),
+        'sessions': sum(item['session_count'] for item in accounts),
+        'tool_calls': sum(item['tool_calls'] for item in accounts),
+        'tokens': sum(item['total_tokens'] for item in accounts),
+    }
+    return {
+        'totals': totals,
+        'memory_queue': state.memory_manager.queue_snapshot(),
+        'accounts': accounts[:20],
+    }
+
+
+def _list_admin_accounts(state: AgentState) -> list[dict[str, Any]]:
+    users_payload = _load_users_file(state)
+    known_users = {
+        str(item.get('id') or item.get('username') or '').strip()
+        for item in users_payload.get('users', [])
+        if isinstance(item, dict)
+    }
+    accounts_root = _accounts_root(state)
+    directory_users = (
+        {
+            path.name
+            for path in accounts_root.iterdir()
+            if path.is_dir() and path.name not in {'__pycache__'}
+        }
+        if accounts_root.exists()
+        else set()
+    )
+    account_ids = sorted({item for item in known_users | directory_users if item})
+    result: list[dict[str, Any]] = []
+    for account_id in account_ids:
+        base = accounts_root / account_id
+        sessions_dir = base / 'sessions'
+        session_files = _iter_session_files(sessions_dir)
+        session_count = 0
+        tool_calls = 0
+        total_tokens = 0
+        input_tokens = 0
+        output_tokens = 0
+        latest_mtime = 0.0
+        models: dict[str, int] = {}
+        for path in session_files:
+            try:
+                data = json.loads(path.read_text(encoding='utf-8'))
+            except (OSError, json.JSONDecodeError):
+                continue
+            session_count += 1
+            tool_calls += int(data.get('tool_calls') or 0)
+            usage = data.get('usage') if isinstance(data.get('usage'), dict) else {}
+            input_tokens += int(usage.get('input_tokens') or 0)
+            output_tokens += int(usage.get('output_tokens') or 0)
+            total_tokens += int(
+                usage.get('total_tokens')
+                or (usage.get('input_tokens') or 0) + (usage.get('output_tokens') or 0)
+            )
+            model_config = data.get('model_config') if isinstance(data.get('model_config'), dict) else {}
+            model = str(model_config.get('model') or data.get('model') or 'unknown')
+            models[model] = models.get(model, 0) + 1
+            latest_mtime = max(latest_mtime, _session_mtime(path))
+        memory_dir = base / 'memory'
+        result.append(
+            {
+                'account_id': account_id,
+                'registered': account_id in known_users,
+                'session_count': session_count,
+                'tool_calls': tool_calls,
+                'input_tokens': input_tokens,
+                'output_tokens': output_tokens,
+                'total_tokens': total_tokens,
+                'latest_session_at': latest_mtime,
+                'models': models,
+                'user_memory_lines': _count_optional_lines(memory_dir / USER_MEMORY_FILENAME),
+                'skill_memory_count': len(list((memory_dir / SKILL_MEMORY_DIRNAME).glob('*.md')))
+                if (memory_dir / SKILL_MEMORY_DIRNAME).exists()
+                else 0,
+            }
+        )
+    return sorted(result, key=lambda item: item['latest_session_at'], reverse=True)
+
+
+def _admin_create_account(state: AgentState, account_id: str) -> dict[str, Any]:
+    safe_id = _safe_account_id(account_id)
+    if safe_id == 'default':
+        raise HTTPException(status_code=400, detail='账号名不合法')
+    users = _load_users_file(state)
+    user_rows = users.setdefault('users', [])
+    if any(
+        isinstance(item, dict)
+        and str(item.get('id') or item.get('username') or '').strip() == safe_id
+        for item in user_rows
+    ):
+        raise HTTPException(status_code=400, detail='账号已存在')
+    user_rows.append(
+        {
+            'id': safe_id,
+            'username': safe_id,
+            'passwordHash': _hash_admin_created_password(),
+            'createdAt': datetime_utc_iso(),
+        }
+    )
+    _save_users_file(state, users)
+    base = _accounts_root(state) / safe_id
+    (base / 'sessions').mkdir(parents=True, exist_ok=True)
+    state.memory_manager.ensure_account(safe_id)
+    return {'account_id': safe_id, 'created': True, 'initial_password': '123456'}
+
+
+def _admin_delete_account(state: AgentState, account_id: str) -> dict[str, Any]:
+    safe_id = _safe_account_id(account_id)
+    users = _load_users_file(state)
+    users['users'] = [
+        item
+        for item in users.get('users', [])
+        if not (
+            isinstance(item, dict)
+            and str(item.get('id') or item.get('username') or '').strip() == safe_id
+        )
+    ]
+    _save_users_file(state, users)
+    sessions_path = _auth_sessions_json_path(state)
+    try:
+        auth_sessions = json.loads(sessions_path.read_text(encoding='utf-8'))
+    except (OSError, json.JSONDecodeError):
+        auth_sessions = {'sessions': {}}
+    if isinstance(auth_sessions.get('sessions'), dict):
+        auth_sessions['sessions'] = {
+            token: session
+            for token, session in auth_sessions['sessions'].items()
+            if not (isinstance(session, dict) and session.get('accountId') == safe_id)
+        }
+        sessions_path.write_text(
+            json.dumps(auth_sessions, ensure_ascii=False, indent=2) + '\n',
+            encoding='utf-8',
+        )
+    base = _accounts_root(state) / safe_id
+    if base.exists():
+        shutil.rmtree(base)
+    state._clear_agents_for_account(safe_id)
+    return {'account_id': safe_id, 'deleted': True}
+
+
+def _count_optional_lines(path: Path) -> int:
+    try:
+        return len(path.read_text(encoding='utf-8').splitlines())
+    except OSError:
+        return 0
+
+
+def datetime_utc_iso() -> str:
+    return time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
+
+
 def _is_internal_message(content: str) -> bool:
     return content.lstrip().startswith('<system-reminder>')
 
@@ -3302,6 +3709,40 @@ def _strip_session_context(content: str) -> str:
     return content.strip()
 
 
+def _extract_used_skill_names(transcript: Any) -> tuple[str, ...]:
+    """从 transcript 中尽量提取实际调用过的 Skill 名称。"""
+    names: list[str] = []
+    if not isinstance(transcript, (list, tuple)):
+        return ()
+    for entry in transcript:
+        if not isinstance(entry, dict):
+            continue
+        tool_calls = entry.get('tool_calls')
+        if not isinstance(tool_calls, list):
+            continue
+        for call in tool_calls:
+            if not isinstance(call, dict):
+                continue
+            function = call.get('function') if isinstance(call.get('function'), dict) else {}
+            tool_name = call.get('name') or function.get('name')
+            if str(tool_name).lower() != 'skill':
+                continue
+            arguments = call.get('arguments') or function.get('arguments')
+            if isinstance(arguments, str):
+                try:
+                    arguments = json.loads(arguments)
+                except json.JSONDecodeError:
+                    arguments = {}
+            if not isinstance(arguments, dict):
+                continue
+            for key in ('skill', 'skill_name', 'name'):
+                value = arguments.get(key)
+                if isinstance(value, str) and value.strip():
+                    names.append(value.strip())
+                    break
+    return tuple(dict.fromkeys(names))
+
+
 def _iter_session_files(directory: Path) -> list[Path]:
     if not directory.exists():
         return []

frontend/app/app/admin/page.tsx

@@ -0,0 +1,583 @@
+"use client";
+
+import {
+	ActivityIcon,
+	DatabaseIcon,
+	RefreshCwIcon,
+	SaveIcon,
+	Trash2Icon,
+	UserPlusIcon,
+	UsersIcon,
+} from "lucide-react";
+import { useEffect, useMemo, useState } from "react";
+import { Button } from "@/components/ui/button";
+import { Input } from "@/components/ui/input";
+
+type AdminSummary = {
+	totals?: {
+		accounts: number;
+		sessions: number;
+		tool_calls: number;
+		tokens: number;
+	};
+	memory_queue?: MemoryQueue;
+	accounts?: AccountRow[];
+};
+
+type AccountRow = {
+	account_id: string;
+	registered: boolean;
+	session_count: number;
+	tool_calls: number;
+	input_tokens: number;
+	output_tokens: number;
+	total_tokens: number;
+	latest_session_at: number;
+	models: Record<string, number>;
+	user_memory_lines: number;
+	skill_memory_count: number;
+};
+
+type MemoryQueue = {
+	totals: {
+		events: number;
+		pending: number;
+		processing: number;
+		done: number;
+		failed: number;
+	};
+	accounts: Array<{
+		account_id: string;
+		events: number;
+		pending: number;
+		processing: number;
+		done: number;
+		failed: number;
+	}>;
+};
+
+type MemoryEvent = {
+	id: string;
+	session_id: string;
+	skills: string[];
+	signals: string[];
+	priority: number;
+	status: string;
+	error?: string | null;
+	created_at: string;
+	updated_at: string;
+};
+
+type UserMemory = {
+	content: string;
+	line_count: number;
+	updated_at?: string | null;
+};
+
+type SkillMemory = {
+	skill: string;
+	content: string;
+	line_count: number;
+	updated_at?: string | null;
+};
+
+const TOKEN_KEY = "zk-admin-token";
+
+export default function AdminPage() {
+	const [token, setToken] = useState("");
+	const [username, setUsername] = useState("admin");
+	const [password, setPassword] = useState("");
+	const [error, setError] = useState("");
+	const [isLoading, setIsLoading] = useState(false);
+
+	useEffect(() => {
+		setToken(localStorage.getItem(TOKEN_KEY) ?? "");
+	}, []);
+
+	async function login() {
+		setError("");
+		setIsLoading(true);
+		try {
+			const response = await fetch("/api/admin/login", {
+				method: "POST",
+				headers: { "content-type": "application/json" },
+				body: JSON.stringify({ username, password }),
+			});
+			const payload = (await response.json()) as {
+				token?: string;
+				detail?: string;
+			};
+			if (!response.ok || !payload.token) {
+				throw new Error(payload.detail ?? "登录失败");
+			}
+			localStorage.setItem(TOKEN_KEY, payload.token);
+			setToken(payload.token);
+		} catch (err) {
+			setError(err instanceof Error ? err.message : "登录失败");
+		} finally {
+			setIsLoading(false);
+		}
+	}
+
+	if (!token) {
+		return (
+			<main className="flex min-h-dvh items-center justify-center bg-background px-4">
+				<div className="w-full max-w-sm rounded-lg border bg-card p-6 shadow-sm">
+					<div className="mb-5">
+						<h1 className="font-semibold text-xl">管理后台</h1>
+						<p className="mt-1 text-muted-foreground text-sm">
+							查看用户、用量、会话、工具调用和记忆队列。
+						</p>
+					</div>
+					<div className="grid gap-3">
+						<Input
+							value={username}
+							onChange={(event) => setUsername(event.target.value)}
+							placeholder="管理员账号"
+						/>
+						<Input
+							type="password"
+							value={password}
+							onChange={(event) => setPassword(event.target.value)}
+							placeholder="管理员密码"
+							onKeyDown={(event) => {
+								if (event.key === "Enter") login();
+							}}
+						/>
+						{error ? <p className="text-destructive text-sm">{error}</p> : null}
+						<Button onClick={login} disabled={isLoading}>
+							{isLoading ? "登录中..." : "登录"}
+						</Button>
+					</div>
+				</div>
+			</main>
+		);
+	}
+
+	return <AdminDashboard token={token} onLogout={() => setToken("")} />;
+}
+
+function AdminDashboard({
+	token,
+	onLogout,
+}: {
+	token: string;
+	onLogout: () => void;
+}) {
+	const [summary, setSummary] = useState<AdminSummary | null>(null);
+	const [accounts, setAccounts] = useState<AccountRow[]>([]);
+	const [selectedAccountId, setSelectedAccountId] = useState("");
+	const [events, setEvents] = useState<MemoryEvent[]>([]);
+	const [userMemory, setUserMemory] = useState<UserMemory | null>(null);
+	const [skillMemories, setSkillMemories] = useState<SkillMemory[]>([]);
+	const [selectedSkill, setSelectedSkill] = useState("");
+	const [memoryDraft, setMemoryDraft] = useState("");
+	const [newAccountId, setNewAccountId] = useState("");
+	const [notice, setNotice] = useState("");
+
+	const selectedAccount = useMemo(
+		() => accounts.find((item) => item.account_id === selectedAccountId),
+		[accounts, selectedAccountId],
+	);
+	const selectedSkillMemory = useMemo(
+		() => skillMemories.find((item) => item.skill === selectedSkill),
+		[skillMemories, selectedSkill],
+	);
+
+	// biome-ignore lint/correctness/useExhaustiveDependencies: 首次进入后台时加载一次即可，后续由刷新按钮触发。
+	useEffect(() => {
+		refresh();
+	}, []);
+
+	// biome-ignore lint/correctness/useExhaustiveDependencies: 账号切换时加载对应记忆；函数体依赖当前 token。
+	useEffect(() => {
+		if (!selectedAccountId) return;
+		loadAccountMemory(selectedAccountId);
+	}, [selectedAccountId]);
+
+	useEffect(() => {
+		if (selectedSkillMemory) {
+			setMemoryDraft(selectedSkillMemory.content);
+		} else if (userMemory) {
+			setMemoryDraft(userMemory.content);
+		}
+	}, [selectedSkillMemory, userMemory]);
+
+	async function refresh() {
+		const [summaryPayload, accountPayload] = await Promise.all([
+			adminFetch<AdminSummary>(token, "/summary"),
+			adminFetch<AccountRow[]>(token, "/accounts"),
+		]);
+		setSummary(summaryPayload);
+		setAccounts(accountPayload);
+		const firstAccount =
+			selectedAccountId || accountPayload[0]?.account_id || "";
+		setSelectedAccountId(firstAccount);
+		if (firstAccount) await loadAccountMemory(firstAccount);
+	}
+
+	async function loadAccountMemory(accountId: string) {
+		const [memoryPayload, userPayload, skillPayload] = await Promise.all([
+			adminFetch<{ events: MemoryEvent[] }>(
+				token,
+				`/memory?account_id=${encodeURIComponent(accountId)}`,
+			),
+			adminFetch<UserMemory>(
+				token,
+				`/memory/user?account_id=${encodeURIComponent(accountId)}`,
+			),
+			adminFetch<SkillMemory[]>(
+				token,
+				`/memory/skills?account_id=${encodeURIComponent(accountId)}`,
+			),
+		]);
+		setEvents(memoryPayload.events ?? []);
+		setUserMemory(userPayload);
+		setSkillMemories(skillPayload);
+		setSelectedSkill("");
+		setMemoryDraft(userPayload.content);
+	}
+
+	async function createAccount() {
+		if (!newAccountId.trim()) return;
+		const payload = await adminFetch<{
+			account_id: string;
+			initial_password: string;
+		}>(token, "/accounts", {
+			method: "POST",
+			body: JSON.stringify({ account_id: newAccountId.trim() }),
+		});
+		setNotice(
+			`已创建 ${payload.account_id}，初始密码 ${payload.initial_password}`,
+		);
+		setNewAccountId("");
+		await refresh();
+	}
+
+	async function deleteAccount(accountId: string) {
+		if (!confirm(`确定删除账号 ${accountId} 及其本地数据吗？`)) return;
+		await adminFetch(token, `/accounts/${encodeURIComponent(accountId)}`, {
+			method: "DELETE",
+		});
+		setNotice(`已删除 ${accountId}`);
+		setSelectedAccountId("");
+		await refresh();
+	}
+
+	async function saveMemory() {
+		if (!selectedAccountId) return;
+		if (selectedSkill) {
+			await adminFetch(
+				token,
+				`/memory/skills/${encodeURIComponent(selectedSkill)}`,
+				{
+					method: "PUT",
+					body: JSON.stringify({
+						account_id: selectedAccountId,
+						content: memoryDraft,
+					}),
+				},
+			);
+			setNotice(`已保存 ${selectedSkill} 记忆`);
+		} else {
+			await adminFetch(token, "/memory/user", {
+				method: "PUT",
+				body: JSON.stringify({
+					account_id: selectedAccountId,
+					content: memoryDraft,
+				}),
+			});
+			setNotice("已保存用户记忆");
+		}
+		await loadAccountMemory(selectedAccountId);
+	}
+
+	function logout() {
+		localStorage.removeItem(TOKEN_KEY);
+		onLogout();
+	}
+
+	return (
+		<main className="min-h-dvh bg-background text-foreground">
+			<header className="sticky top-0 z-10 border-b bg-background/90 backdrop-blur">
+				<div className="mx-auto flex max-w-7xl items-center gap-3 px-5 py-3">
+					<div className="flex size-9 items-center justify-center rounded-lg bg-primary text-primary-foreground">
+						<DatabaseIcon className="size-4" />
+					</div>
+					<div className="min-w-0 flex-1">
+						<h1 className="font-semibold text-lg">ZK Data Agent 管理后台</h1>
+						<p className="text-muted-foreground text-xs">
+							用户、会话、工具用量和个性化记忆队列。
+						</p>
+					</div>
+					<Button variant="outline" size="sm" onClick={refresh}>
+						<RefreshCwIcon className="size-4" />
+						刷新
+					</Button>
+					<Button variant="ghost" size="sm" onClick={logout}>
+						退出
+					</Button>
+				</div>
+			</header>
+
+			<div className="mx-auto grid max-w-7xl gap-4 px-5 py-5">
+				{notice ? (
+					<div className="rounded-md border bg-muted/40 px-3 py-2 text-sm">
+						{notice}
+					</div>
+				) : null}
+
+				<div className="grid gap-3 md:grid-cols-4">
+					<Metric title="用户" value={summary?.totals?.accounts ?? 0} />
+					<Metric title="会话" value={summary?.totals?.sessions ?? 0} />
+					<Metric title="工具调用" value={summary?.totals?.tool_calls ?? 0} />
+					<Metric
+						title="Token"
+						value={formatNumber(summary?.totals?.tokens ?? 0)}
+					/>
+				</div>
+
+				<div className="grid min-h-[640px] gap-4 lg:grid-cols-[320px_1fr]">
+					<section className="rounded-lg border bg-card">
+						<div className="border-b p-3">
+							<div className="mb-3 flex items-center gap-2 font-medium">
+								<UsersIcon className="size-4" />
+								用户
+							</div>
+							<div className="flex gap-2">
+								<Input
+									value={newAccountId}
+									placeholder="新账号"
+									onChange={(event) => setNewAccountId(event.target.value)}
+								/>
+								<Button size="icon" variant="outline" onClick={createAccount}>
+									<UserPlusIcon className="size-4" />
+								</Button>
+							</div>
+						</div>
+						<div className="max-h-[560px] overflow-y-auto p-2">
+							{accounts.map((account) => (
+								<button
+									key={account.account_id}
+									type="button"
+									className={`mb-1 flex w-full flex-col rounded-md px-3 py-2 text-left transition-colors hover:bg-muted ${
+										selectedAccountId === account.account_id ? "bg-muted" : ""
+									}`}
+									onClick={() => setSelectedAccountId(account.account_id)}
+								>
+									<span className="flex items-center justify-between gap-2">
+										<span className="truncate font-medium text-sm">
+											{account.account_id}
+										</span>
+										<span className="text-muted-foreground text-xs">
+											{account.session_count} 会话
+										</span>
+									</span>
+									<span className="mt-1 text-muted-foreground text-xs">
+										{account.tool_calls} 工具 ·{" "}
+										{formatNumber(account.total_tokens)} tokens
+									</span>
+								</button>
+							))}
+						</div>
+					</section>
+
+					<section className="grid gap-4">
+						<div className="grid gap-3 md:grid-cols-4">
+							<Metric title="账号" value={selectedAccount?.account_id ?? "-"} />
+							<Metric
+								title="用户记忆行"
+								value={selectedAccount?.user_memory_lines ?? 0}
+							/>
+							<Metric
+								title="Skill 记忆"
+								value={selectedAccount?.skill_memory_count ?? 0}
+							/>
+							<Metric
+								title="队列 pending"
+								value={
+									summary?.memory_queue?.accounts.find(
+										(item) => item.account_id === selectedAccountId,
+									)?.pending ?? 0
+								}
+							/>
+						</div>
+
+						<div className="grid gap-4 xl:grid-cols-[1fr_360px]">
+							<div className="rounded-lg border bg-card">
+								<div className="flex items-center justify-between border-b p-3">
+									<div>
+										<div className="font-medium">记忆编辑</div>
+										<div className="text-muted-foreground text-xs">
+											默认编辑用户记忆；选择 Skill 后编辑该 Skill 记忆。
+										</div>
+									</div>
+									<Button size="sm" onClick={saveMemory}>
+										<SaveIcon className="size-4" />
+										保存
+									</Button>
+								</div>
+								<div className="flex gap-2 border-b p-3">
+									<Button
+										variant={selectedSkill ? "outline" : "secondary"}
+										size="sm"
+										onClick={() => {
+											setSelectedSkill("");
+											setMemoryDraft(userMemory?.content ?? "");
+										}}
+									>
+										用户记忆
+									</Button>
+									<div className="flex min-w-0 flex-1 gap-2 overflow-x-auto">
+										{skillMemories.map((item) => (
+											<Button
+												key={item.skill}
+												variant={
+													selectedSkill === item.skill ? "secondary" : "outline"
+												}
+												size="sm"
+												onClick={() => setSelectedSkill(item.skill)}
+											>
+												{item.skill}
+											</Button>
+										))}
+									</div>
+								</div>
+								<textarea
+									className="min-h-[420px] w-full resize-y bg-transparent p-4 font-mono text-sm outline-none"
+									value={memoryDraft}
+									onChange={(event) => setMemoryDraft(event.target.value)}
+								/>
+							</div>
+
+							<div className="grid gap-4">
+								<div className="rounded-lg border bg-card">
+									<div className="flex items-center gap-2 border-b p-3 font-medium">
+										<ActivityIcon className="size-4" />
+										记忆队列
+									</div>
+									<div className="grid grid-cols-2 gap-2 p-3 text-sm">
+										<SmallStat
+											label="全部"
+											value={summary?.memory_queue?.totals.events ?? 0}
+										/>
+										<SmallStat
+											label="待处理"
+											value={summary?.memory_queue?.totals.pending ?? 0}
+										/>
+										<SmallStat
+											label="处理中"
+											value={summary?.memory_queue?.totals.processing ?? 0}
+										/>
+										<SmallStat
+											label="失败"
+											value={summary?.memory_queue?.totals.failed ?? 0}
+										/>
+									</div>
+								</div>
+
+								<div className="rounded-lg border bg-card">
+									<div className="border-b p-3 font-medium">最近记忆事件</div>
+									<div className="max-h-[360px] overflow-y-auto p-2">
+										{events.length ? (
+											events.map((event) => (
+												<div
+													key={event.id}
+													className="mb-2 rounded-md border bg-muted/20 p-2 text-xs"
+												>
+													<div className="flex items-center justify-between gap-2">
+														<span className="font-medium">{event.status}</span>
+														<span className="text-muted-foreground">
+															P{event.priority}
+														</span>
+													</div>
+													<div className="mt-1 truncate text-muted-foreground">
+														{event.session_id}
+													</div>
+													<div className="mt-1">
+														{event.signals.join(" / ") || "无信号"}
+													</div>
+													{event.error ? (
+														<div className="mt-1 text-destructive">
+															{event.error}
+														</div>
+													) : null}
+												</div>
+											))
+										) : (
+											<p className="p-4 text-center text-muted-foreground text-sm">
+												暂无事件。
+											</p>
+										)}
+									</div>
+								</div>
+
+								{selectedAccountId ? (
+									<Button
+										variant="destructive"
+										onClick={() => deleteAccount(selectedAccountId)}
+									>
+										<Trash2Icon className="size-4" />
+										删除当前用户
+									</Button>
+								) : null}
+							</div>
+						</div>
+					</section>
+				</div>
+			</div>
+		</main>
+	);
+}
+
+function Metric({ title, value }: { title: string; value: string | number }) {
+	return (
+		<div className="rounded-lg border bg-card p-3">
+			<div className="text-muted-foreground text-xs">{title}</div>
+			<div className="mt-1 truncate font-semibold text-xl">{value}</div>
+		</div>
+	);
+}
+
+function SmallStat({
+	label,
+	value,
+}: {
+	label: string;
+	value: string | number;
+}) {
+	return (
+		<div className="rounded-md border bg-muted/20 p-2">
+			<div className="font-medium">{value}</div>
+			<div className="text-muted-foreground text-xs">{label}</div>
+		</div>
+	);
+}
+
+async function adminFetch<T>(
+	token: string,
+	path: string,
+	init?: RequestInit,
+): Promise<T> {
+	const separator = path.includes("?") ? "&" : "?";
+	const response = await fetch(`/api/admin${path}${separator}token=${token}`, {
+		...init,
+		headers: {
+			"content-type": "application/json",
+			...(init?.headers ?? {}),
+		},
+		cache: "no-store",
+	});
+	const payload = await response.json().catch(() => ({}));
+	if (!response.ok) {
+		throw new Error(payload.detail ?? payload.error ?? "请求失败");
+	}
+	return payload as T;
+}
+
+function formatNumber(value: number) {
+	return new Intl.NumberFormat("zh-CN", {
+		notation: value > 10000 ? "compact" : "standard",
+		maximumFractionDigits: 1,
+	}).format(value);
+}

frontend/app/app/api/admin/[...path]/route.ts

@@ -0,0 +1,52 @@
+const CLAW_API_URL = process.env.CLAW_API_URL ?? "http://127.0.0.1:8765";
+
+type RouteContext = {
+	params: Promise<{ path?: string[] }>;
+};
+
+export async function GET(request: Request, context: RouteContext) {
+	return proxyAdminRequest(request, context);
+}
+
+export async function POST(request: Request, context: RouteContext) {
+	return proxyAdminRequest(request, context);
+}
+
+export async function PUT(request: Request, context: RouteContext) {
+	return proxyAdminRequest(request, context);
+}
+
+export async function DELETE(request: Request, context: RouteContext) {
+	return proxyAdminRequest(request, context);
+}
+
+async function proxyAdminRequest(request: Request, context: RouteContext) {
+	const params = await context.params;
+	const path = (params.path ?? []).map(encodeURIComponent).join("/");
+	const sourceUrl = new URL(request.url);
+	const targetUrl = new URL(`${CLAW_API_URL}/api/admin/${path}`);
+	for (const [key, value] of sourceUrl.searchParams.entries()) {
+		targetUrl.searchParams.set(key, value);
+	}
+
+	const init: RequestInit = {
+		method: request.method,
+		headers: {
+			"content-type": request.headers.get("content-type") ?? "application/json",
+		},
+		cache: "no-store",
+	};
+	if (!["GET", "HEAD"].includes(request.method)) {
+		init.body = await request.text();
+	}
+
+	const response = await fetch(targetUrl, init);
+	const payload = await response.text();
+	return new Response(payload, {
+		status: response.status,
+		headers: {
+			"content-type":
+				response.headers.get("content-type") ?? "application/json",
+		},
+	});
+}

frontend/app/app/api/claw/memory/skills/[skillName]/route.ts

@@ -0,0 +1,55 @@
+import { getCurrentAccount } from "@/lib/claw-auth";
+
+const CLAW_API_URL = process.env.CLAW_API_URL ?? "http://127.0.0.1:8765";
+
+type RouteContext = {
+	params: Promise<{ skillName: string }>;
+};
+
+export async function GET(_request: Request, context: RouteContext) {
+	const account = await getCurrentAccount();
+	if (!account)
+		return Response.json({ error: "unauthorized" }, { status: 401 });
+	const { skillName } = await context.params;
+	const url = new URL(
+		`${CLAW_API_URL}/api/memory/skills/${encodeURIComponent(skillName)}`,
+	);
+	url.searchParams.set("account_id", account.id);
+	const response = await fetch(url, { cache: "no-store" });
+	const payload = await response.text();
+	return new Response(payload, {
+		status: response.status,
+		headers: {
+			"content-type":
+				response.headers.get("content-type") ?? "application/json",
+		},
+	});
+}
+
+export async function PUT(request: Request, context: RouteContext) {
+	const account = await getCurrentAccount();
+	if (!account)
+		return Response.json({ error: "unauthorized" }, { status: 401 });
+	const { skillName } = await context.params;
+	const body = (await request.json()) as { content?: string };
+	const response = await fetch(
+		`${CLAW_API_URL}/api/memory/skills/${encodeURIComponent(skillName)}`,
+		{
+			method: "PUT",
+			headers: { "content-type": "application/json" },
+			body: JSON.stringify({
+				account_id: account.id,
+				content: body.content ?? "",
+			}),
+			cache: "no-store",
+		},
+	);
+	const payload = await response.text();
+	return new Response(payload, {
+		status: response.status,
+		headers: {
+			"content-type":
+				response.headers.get("content-type") ?? "application/json",
+		},
+	});
+}

frontend/app/app/api/claw/memory/skills/route.ts

@@ -0,0 +1,19 @@
+import { getCurrentAccount } from "@/lib/claw-auth";
+
+const CLAW_API_URL = process.env.CLAW_API_URL ?? "http://127.0.0.1:8765";
+
+export async function GET() {
+	const account = await getCurrentAccount();
+	if (!account) return Response.json([], { status: 401 });
+	const url = new URL(`${CLAW_API_URL}/api/memory/skills`);
+	url.searchParams.set("account_id", account.id);
+	const response = await fetch(url, { cache: "no-store" });
+	const payload = await response.text();
+	return new Response(payload, {
+		status: response.status,
+		headers: {
+			"content-type":
+				response.headers.get("content-type") ?? "application/json",
+		},
+	});
+}

frontend/app/app/api/claw/memory/user/route.ts

@@ -0,0 +1,44 @@
+import { getCurrentAccount } from "@/lib/claw-auth";
+
+const CLAW_API_URL = process.env.CLAW_API_URL ?? "http://127.0.0.1:8765";
+
+export async function GET() {
+	const account = await getCurrentAccount();
+	if (!account)
+		return Response.json({ error: "unauthorized" }, { status: 401 });
+	const url = new URL(`${CLAW_API_URL}/api/memory/user`);
+	url.searchParams.set("account_id", account.id);
+	const response = await fetch(url, { cache: "no-store" });
+	const payload = await response.text();
+	return new Response(payload, {
+		status: response.status,
+		headers: {
+			"content-type":
+				response.headers.get("content-type") ?? "application/json",
+		},
+	});
+}
+
+export async function PUT(request: Request) {
+	const account = await getCurrentAccount();
+	if (!account)
+		return Response.json({ error: "unauthorized" }, { status: 401 });
+	const body = (await request.json()) as { content?: string };
+	const response = await fetch(`${CLAW_API_URL}/api/memory/user`, {
+		method: "PUT",
+		headers: { "content-type": "application/json" },
+		body: JSON.stringify({
+			account_id: account.id,
+			content: body.content ?? "",
+		}),
+		cache: "no-store",
+	});
+	const payload = await response.text();
+	return new Response(payload, {
+		status: response.status,
+		headers: {
+			"content-type":
+				response.headers.get("content-type") ?? "application/json",
+		},
+	});
+}

frontend/app/components/assistant-ui/threadlist-sidebar.tsx

@@ -3,10 +3,12 @@ import { ThreadListPrimitive } from "@assistant-ui/react";
 import {
 	BarChart3Icon,
 	BotIcon,
+	BrainIcon,
 	ChevronDownIcon,
 	Clock3Icon,
 	LogOutIcon,
 	PanelLeftOpenIcon,
+	SaveIcon,
 	SearchIcon,
 	SquarePenIcon,
 	UserIcon,
@@ -86,6 +88,19 @@ type ClawStoredSession = {
 	model?: string;
 };
 
+type UserMemoryPayload = {
+	content: string;
+	line_count?: number;
+	updated_at?: string | null;
+};
+
+type SkillMemoryPayload = {
+	skill: string;
+	content: string;
+	line_count?: number;
+	updated_at?: string | null;
+};
+
 type ReplaySessionPayload = Parameters<typeof toReplayRepository>[0];
 
 type SidebarSession = ClawSessionSummary & {
@@ -424,6 +439,12 @@ function AccountMenu({
 		null,
 	);
 	const [modelUsageOpen, setModelUsageOpen] = useState(false);
+	const [memoryOpen, setMemoryOpen] = useState(false);
+	const [userMemory, setUserMemory] = useState<UserMemoryPayload | null>(null);
+	const [skillMemories, setSkillMemories] = useState<SkillMemoryPayload[]>([]);
+	const [memoryDraft, setMemoryDraft] = useState("");
+	const [selectedMemorySkill, setSelectedMemorySkill] = useState("");
+	const [memorySaving, setMemorySaving] = useState(false);
 
 	useEffect(() => {
 		if (!open) return;
@@ -462,12 +483,67 @@ function AccountMenu({
 			});
 	}, [open]);
 
+	useEffect(() => {
+		if (!memoryOpen) return;
+		Promise.all([
+			fetch("/api/claw/memory/user", { cache: "no-store" }).then((res) =>
+				res.ok ? res.json() : null,
+			),
+			fetch("/api/claw/memory/skills", { cache: "no-store" }).then((res) =>
+				res.ok ? res.json() : [],
+			),
+		]).then(([userPayload, skillPayload]) => {
+			const nextUser = userPayload as UserMemoryPayload | null;
+			setUserMemory(nextUser);
+			setMemoryDraft(nextUser?.content ?? "");
+			setSelectedMemorySkill("");
+			setSkillMemories(
+				Array.isArray(skillPayload)
+					? (skillPayload as SkillMemoryPayload[])
+					: [],
+			);
+		});
+	}, [memoryOpen]);
+
 	async function logout() {
 		await fetch("/api/claw/auth/logout", { method: "POST" });
 		clearActiveSessionId();
 		onLogout();
 	}
 
+	async function saveUserMemory() {
+		setMemorySaving(true);
+		try {
+			const response = await fetch(
+				selectedMemorySkill
+					? `/api/claw/memory/skills/${encodeURIComponent(selectedMemorySkill)}`
+					: "/api/claw/memory/user",
+				{
+					method: "PUT",
+					headers: { "content-type": "application/json" },
+					body: JSON.stringify({ content: memoryDraft }),
+				},
+			);
+			if (response.ok) {
+				if (selectedMemorySkill) {
+					const payload = (await response.json()) as SkillMemoryPayload;
+					setSkillMemories((current) =>
+						current.map((item) =>
+							item.skill === payload.skill ? payload : item,
+						),
+					);
+					setMemoryDraft(payload.content);
+				} else {
+					const payload = (await response.json()) as UserMemoryPayload;
+					setUserMemory(payload);
+					setMemoryDraft(payload.content);
+				}
+			}
+		} finally {
+			setMemorySaving(false);
+		}
+	}
+
 	const totalToolCalls = sessions.reduce(
 		(sum, session) => sum + (session.tool_calls ?? 0),
 		0,
@@ -608,6 +684,80 @@ function AccountMenu({
 							<div className="text-muted-foreground">暂无可统计会话</div>
 						) : null}
 					</div>
+					<div className="mt-3 rounded-md border bg-muted/30 p-2 text-xs">
+						<button
+							type="button"
+							className="flex w-full items-center justify-between gap-2 text-left"
+							onClick={() => setMemoryOpen((value) => !value)}
+						>
+							<span className="flex items-center gap-1.5 font-medium">
+								<BrainIcon className="size-3.5" />
+								记忆
+							</span>
+							<span className="flex items-center gap-1 text-muted-foreground">
+								用户 {userMemory?.line_count ?? 0} 行 · Skill{" "}
+								{skillMemories.length}
+								<ChevronDownIcon
+									className={`size-3.5 transition-transform ${
+										memoryOpen ? "rotate-180" : ""
+									}`}
+								/>
+							</span>
+						</button>
+						{memoryOpen ? (
+							<div className="mt-2 grid gap-2">
+								<div className="flex gap-1 overflow-x-auto">
+									<Button
+										size="xs"
+										variant={selectedMemorySkill ? "outline" : "secondary"}
+										onClick={() => {
+											setSelectedMemorySkill("");
+											setMemoryDraft(userMemory?.content ?? "");
+										}}
+									>
+										用户记忆
+									</Button>
+									{skillMemories.map((item) => (
+										<Button
+											key={item.skill}
+											size="xs"
+											variant={
+												selectedMemorySkill === item.skill
+													? "secondary"
+													: "outline"
+											}
+											onClick={() => {
+												setSelectedMemorySkill(item.skill);
+												setMemoryDraft(item.content);
+											}}
+										>
+											{item.skill}
+										</Button>
+									))}
+								</div>
+								<textarea
+									className="min-h-32 w-full resize-y rounded-md border bg-background p-2 font-mono text-xs outline-none"
+									value={memoryDraft}
+									onChange={(event) => setMemoryDraft(event.target.value)}
+								/>
+								<div className="flex items-center justify-between gap-2">
+									<span className="truncate text-muted-foreground">
+										Skill 记忆：
+										{skillMemories.map((item) => item.skill).join("、") ||
+											"暂无"}
+									</span>
+									<Button
+										size="xs"
+										onClick={saveUserMemory}
+										disabled={memorySaving}
+									>
+										<SaveIcon className="size-3" />
+										保存
+									</Button>
+								</div>
+							</div>
+						) : null}
+					</div>
 					<div className="mt-3 flex items-center justify-between gap-2 border-t pt-3">
 						<span className="text-muted-foreground text-xs">
 							界面与模型配置

src/personal_memory.py

@@ -0,0 +1,737 @@
+"""账号级个性化记忆。
+
+这个模块只做轻量、可替换的第一版实现：
+- Markdown 是人可编辑的记忆正文。
+- SQLite 是后台事件队列、状态和 revision 账本。
+- 主对话链路只追加事件和读取已有记忆，不在同步路径里整理记忆。
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import sqlite3
+import threading
+import time
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+from uuid import uuid4
+
+from .agent_types import ModelConfig
+from .openai_compat import OpenAICompatClient
+
+
+MEMORY_DB_FILENAME = 'memory.db'
+USER_MEMORY_FILENAME = 'user.md'
+SKILL_MEMORY_DIRNAME = 'skills'
+MAX_MEMORY_LINES_BEFORE_COMPACT = 300
+
+EXPLICIT_MEMORY_PATTERNS = (
+    '记住',
+    '以后',
+    '下次',
+    '默认',
+    '总是',
+    '不要',
+    '应该',
+    '固定',
+)
+
+CORRECTION_PATTERNS = (
+    '不对',
+    '不是这样',
+    '格式错',
+    '之前说过',
+    '还是不行',
+    '这个不对',
+    '需要改成',
+)
+
+
+@dataclass(frozen=True)
+class MemoryEvent:
+    account_id: str
+    session_id: str
+    interaction_id: str
+    user_prompt: str
+    assistant_output: str
+    skills: tuple[str, ...]
+    model: str
+    priority: int
+    signals: tuple[str, ...]
+    created_at: str
+
+
+class PersonalMemoryManager:
+    """管理用户记忆、Skill 记忆和异步整理队列。"""
+
+    def __init__(self, accounts_root: Path, model_config_getter: Any) -> None:
+        self.accounts_root = accounts_root
+        self._model_config_getter = model_config_getter
+        self._stop_event = threading.Event()
+        self._worker_thread: threading.Thread | None = None
+        self._init_lock = threading.Lock()
+
+    def start(self) -> None:
+        if self._worker_thread and self._worker_thread.is_alive():
+            return
+        self._worker_thread = threading.Thread(
+            target=self._worker_loop,
+            name='personal-memory-worker',
+            daemon=True,
+        )
+        self._worker_thread.start()
+
+    def stop(self) -> None:
+        self._stop_event.set()
+
+    def ensure_account(self, account_id: str) -> Path:
+        base = self._account_memory_root(account_id)
+        (base / SKILL_MEMORY_DIRNAME).mkdir(parents=True, exist_ok=True)
+        self._ensure_db(account_id)
+        user_path = base / USER_MEMORY_FILENAME
+        if not user_path.exists():
+            self._atomic_write_text(
+                user_path,
+                '# 用户记忆\n\n暂无用户记忆。\n',
+            )
+        return base
+
+    def render_injection(
+        self,
+        account_id: str | None,
+        enabled_skill_names: tuple[str, ...] | None,
+    ) -> str:
+        """渲染给模型的个性化记忆上下文。"""
+        if not account_id:
+            return ''
+        base = self.ensure_account(account_id)
+        sections: list[str] = []
+        user_text = self._read_memory_file(base / USER_MEMORY_FILENAME)
+        if user_text:
+            sections.extend(['## 用户记忆', user_text])
+
+        skill_root = base / SKILL_MEMORY_DIRNAME
+        skill_sections: list[str] = []
+        skill_names = enabled_skill_names or ()
+        for skill_name in skill_names:
+            safe_name = _safe_memory_name(skill_name)
+            if not safe_name:
+                continue
+            text = self._read_memory_file(skill_root / f'{safe_name}.md')
+            if text:
+                skill_sections.extend([f'### {skill_name}', text])
+        if skill_sections:
+            sections.extend(['## Skill 使用记忆', *skill_sections])
+        if not sections:
+            return ''
+        return '\n'.join(
+            [
+                '# 个性化记忆',
+                '以下是该账号长期保存的偏好和 Skill 使用经验。若与用户本轮明确要求冲突，以用户本轮要求为准。',
+                *sections,
+            ]
+        ).strip()
+
+    def enqueue_interaction(
+        self,
+        *,
+        account_id: str | None,
+        session_id: str | None,
+        user_prompt: str,
+        assistant_output: str,
+        skills: tuple[str, ...],
+        model: str,
+    ) -> None:
+        if not account_id or not session_id:
+            return
+        signals = detect_memory_signals(user_prompt, assistant_output, skills)
+        if not signals:
+            return
+        event = MemoryEvent(
+            account_id=account_id,
+            session_id=session_id,
+            interaction_id=uuid4().hex,
+            user_prompt=user_prompt[-6000:],
+            assistant_output=assistant_output[-6000:],
+            skills=skills,
+            model=model,
+            priority=10 if 'explicit' in signals else 3,
+            signals=signals,
+            created_at=_now_iso(),
+        )
+        self._insert_event(event)
+
+    def list_user_memory(self, account_id: str) -> dict[str, Any]:
+        base = self.ensure_account(account_id)
+        return {
+            'account_id': account_id,
+            'kind': 'user',
+            'content': (base / USER_MEMORY_FILENAME).read_text(encoding='utf-8'),
+            'line_count': _line_count(base / USER_MEMORY_FILENAME),
+            'updated_at': _mtime_iso(base / USER_MEMORY_FILENAME),
+        }
+
+    def update_user_memory(self, account_id: str, content: str) -> dict[str, Any]:
+        base = self.ensure_account(account_id)
+        path = base / USER_MEMORY_FILENAME
+        self._atomic_write_text(path, _normalize_memory_markdown(content, '用户记忆'))
+        self._record_revision(account_id, 'user', None)
+        return self.list_user_memory(account_id)
+
+    def list_skill_memories(self, account_id: str) -> list[dict[str, Any]]:
+        base = self.ensure_account(account_id)
+        skill_root = base / SKILL_MEMORY_DIRNAME
+        result: list[dict[str, Any]] = []
+        for path in sorted(skill_root.glob('*.md')):
+            result.append(
+                {
+                    'skill': path.stem,
+                    'content': path.read_text(encoding='utf-8'),
+                    'line_count': _line_count(path),
+                    'updated_at': _mtime_iso(path),
+                }
+            )
+        return result
+
+    def read_skill_memory(self, account_id: str, skill_name: str) -> dict[str, Any]:
+        base = self.ensure_account(account_id)
+        safe_name = _safe_memory_name(skill_name)
+        if not safe_name:
+            raise ValueError('skill name must not be empty')
+        path = base / SKILL_MEMORY_DIRNAME / f'{safe_name}.md'
+        if not path.exists():
+            self._atomic_write_text(path, f'# {safe_name} 使用记忆\n\n暂无 Skill 使用记忆。\n')
+        return {
+            'account_id': account_id,
+            'skill': safe_name,
+            'content': path.read_text(encoding='utf-8'),
+            'line_count': _line_count(path),
+            'updated_at': _mtime_iso(path),
+        }
+
+    def update_skill_memory(
+        self,
+        account_id: str,
+        skill_name: str,
+        content: str,
+    ) -> dict[str, Any]:
+        base = self.ensure_account(account_id)
+        safe_name = _safe_memory_name(skill_name)
+        if not safe_name:
+            raise ValueError('skill name must not be empty')
+        path = base / SKILL_MEMORY_DIRNAME / f'{safe_name}.md'
+        self._atomic_write_text(
+            path,
+            _normalize_memory_markdown(content, f'{safe_name} 使用记忆'),
+        )
+        self._record_revision(account_id, 'skill', safe_name)
+        return self.read_skill_memory(account_id, safe_name)
+
+    def delete_skill_memory(self, account_id: str, skill_name: str) -> dict[str, Any]:
+        base = self.ensure_account(account_id)
+        safe_name = _safe_memory_name(skill_name)
+        if not safe_name:
+            raise ValueError('skill name must not be empty')
+        path = base / SKILL_MEMORY_DIRNAME / f'{safe_name}.md'
+        if path.exists():
+            path.unlink()
+        self._record_revision(account_id, 'skill', safe_name)
+        return {'deleted': True, 'skill': safe_name}
+
+    def queue_snapshot(self, account_id: str | None = None) -> dict[str, Any]:
+        accounts = [account_id] if account_id else self._list_account_ids()
+        totals = {
+            'pending': 0,
+            'processing': 0,
+            'done': 0,
+            'failed': 0,
+            'events': 0,
+        }
+        account_rows: list[dict[str, Any]] = []
+        for item in accounts:
+            self.ensure_account(item)
+            conn = self._connect(item)
+            try:
+                rows = conn.execute(
+                    'select status, count(*) from memory_events group by status'
+                ).fetchall()
+                counts = {str(status): int(count) for status, count in rows}
+                events = sum(counts.values())
+                account_rows.append(
+                    {
+                        'account_id': item,
+                        'events': events,
+                        'pending': counts.get('pending', 0),
+                        'processing': counts.get('processing', 0),
+                        'done': counts.get('done', 0),
+                        'failed': counts.get('failed', 0),
+                    }
+                )
+                totals['events'] += events
+                for key in ('pending', 'processing', 'done', 'failed'):
+                    totals[key] += counts.get(key, 0)
+            finally:
+                conn.close()
+        return {'totals': totals, 'accounts': account_rows}
+
+    def list_recent_events(self, account_id: str, limit: int = 50) -> list[dict[str, Any]]:
+        self.ensure_account(account_id)
+        conn = self._connect(account_id)
+        try:
+            rows = conn.execute(
+                '''
+                select id, session_id, skills_json, signals_json, priority, status,
+                       error, created_at, updated_at
+                from memory_events
+                order by created_at desc
+                limit ?
+                ''',
+                (max(1, min(limit, 200)),),
+            ).fetchall()
+            return [
+                {
+                    'id': row['id'],
+                    'session_id': row['session_id'],
+                    'skills': _loads_json_list(row['skills_json']),
+                    'signals': _loads_json_list(row['signals_json']),
+                    'priority': row['priority'],
+                    'status': row['status'],
+                    'error': row['error'],
+                    'created_at': row['created_at'],
+                    'updated_at': row['updated_at'],
+                }
+                for row in rows
+            ]
+        finally:
+            conn.close()
+
+    def _worker_loop(self) -> None:
+        while not self._stop_event.wait(5.0):
+            for account_id in self._list_account_ids():
+                try:
+                    self._process_account_events(account_id)
+                except Exception:
+                    continue
+
+    def _process_account_events(self, account_id: str) -> None:
+        self.ensure_account(account_id)
+        conn = self._connect(account_id)
+        try:
+            conn.execute('begin immediate')
+            rows = conn.execute(
+                '''
+                select * from memory_events
+                where status = 'pending'
+                order by priority desc, created_at asc
+                limit 8
+                '''
+            ).fetchall()
+            if not rows:
+                conn.commit()
+                return
+            now = _now_iso()
+            ids = [row['id'] for row in rows]
+            conn.executemany(
+                'update memory_events set status = ?, updated_at = ? where id = ?',
+                [('processing', now, row_id) for row_id in ids],
+            )
+            conn.commit()
+        except Exception:
+            conn.rollback()
+            raise
+        finally:
+            conn.close()
+
+        try:
+            self._consolidate_events(account_id, rows)
+        except Exception as exc:
+            conn = self._connect(account_id)
+            try:
+                now = _now_iso()
+                conn.executemany(
+                    '''
+                    update memory_events
+                    set status = 'failed', error = ?, updated_at = ?
+                    where id = ?
+                    ''',
+                    [(str(exc), now, row_id) for row_id in ids],
+                )
+                conn.commit()
+            finally:
+                conn.close()
+            return
+
+        conn = self._connect(account_id)
+        try:
+            now = _now_iso()
+            conn.executemany(
+                "update memory_events set status = 'done', updated_at = ? where id = ?",
+                [(now, row_id) for row_id in ids],
+            )
+            conn.commit()
+        finally:
+            conn.close()
+
+    def _consolidate_events(self, account_id: str, rows: list[sqlite3.Row]) -> None:
+        events = [
+            {
+                'user_prompt': row['user_prompt'],
+                'assistant_output': row['assistant_output'],
+                'skills': _loads_json_list(row['skills_json']),
+                'signals': _loads_json_list(row['signals_json']),
+            }
+            for row in rows
+        ]
+        skills = sorted(
+            {
+                skill
+                for event in events
+                for skill in event['skills']
+                if isinstance(skill, str) and skill.strip()
+            }
+        )
+        base = self.ensure_account(account_id)
+        existing_user = self._read_memory_file(base / USER_MEMORY_FILENAME)
+        existing_skills = {
+            skill: self._read_memory_file(
+                base / SKILL_MEMORY_DIRNAME / f'{_safe_memory_name(skill)}.md'
+            )
+            for skill in skills
+            if _safe_memory_name(skill)
+        }
+        updates = self._generate_memory_updates(
+            account_id=account_id,
+            events=events,
+            existing_user=existing_user,
+            existing_skills=existing_skills,
+        )
+        if updates.get('user_memory'):
+            self.update_user_memory(account_id, str(updates['user_memory']))
+        skill_updates = updates.get('skill_memories')
+        if isinstance(skill_updates, dict):
+            for skill, content in skill_updates.items():
+                if isinstance(skill, str) and isinstance(content, str) and content.strip():
+                    self.update_skill_memory(account_id, skill, content)
+
+    def _generate_memory_updates(
+        self,
+        *,
+        account_id: str,
+        events: list[dict[str, Any]],
+        existing_user: str,
+        existing_skills: dict[str, str],
+    ) -> dict[str, Any]:
+        config = self._model_config_getter(account_id)
+        messages = [
+            {
+                'role': 'system',
+                'content': (
+                    '你是 ZK Data Agent 的记忆整理后台。'
+                    '请只沉淀长期稳定的用户偏好和 Skill 使用经验。'
+                    '不要记录一次性任务目标、临时文件名、普通聊天内容。'
+                    '输出必须是 JSON 对象。'
+                ),
+            },
+            {
+                'role': 'user',
+                'content': json.dumps(
+                    {
+                        'existing_user_memory': existing_user,
+                        'existing_skill_memories': existing_skills,
+                        'new_events': events,
+                        'requirements': [
+                            '保留 Markdown，可整理合并，不要追加流水账。',
+                            '如果没有值得更新的用户记忆，user_memory 返回空字符串。',
+                            'skill_memories 只返回需要更新的 skill。',
+                        ],
+                        'output_schema': {
+                            'user_memory': '完整的用户记忆 Markdown，或空字符串',
+                            'skill_memories': {
+                                'skill-name': '完整的 Skill 记忆 Markdown'
+                            },
+                        },
+                    },
+                    ensure_ascii=False,
+                ),
+            },
+        ]
+        turn = OpenAICompatClient(config).complete(messages, tools=[])
+        content = turn.content.strip()
+        try:
+            parsed = json.loads(_extract_json_object(content))
+        except Exception:
+            parsed = self._fallback_memory_updates(events, existing_user, existing_skills)
+        if not isinstance(parsed, dict):
+            return {}
+        return parsed
+
+    def _fallback_memory_updates(
+        self,
+        events: list[dict[str, Any]],
+        existing_user: str,
+        existing_skills: dict[str, str],
+    ) -> dict[str, Any]:
+        user_lines = _memory_body_lines(existing_user)
+        skill_updates: dict[str, str] = {}
+        for event in events:
+            prompt = str(event.get('user_prompt') or '').strip()
+            if any(pattern in prompt for pattern in EXPLICIT_MEMORY_PATTERNS):
+                candidate = _shorten_memory_line(prompt)
+                if candidate and candidate not in user_lines:
+                    user_lines.append(candidate)
+            for skill in event.get('skills') or []:
+                if not isinstance(skill, str) or not skill.strip():
+                    continue
+                body = _memory_body_lines(existing_skills.get(skill, ''))
+                if any(pattern in prompt for pattern in CORRECTION_PATTERNS):
+                    candidate = _shorten_memory_line(prompt)
+                    if candidate and candidate not in body:
+                        body.append(candidate)
+                    skill_updates[skill] = _render_memory_doc(
+                        f'{skill} 使用记忆',
+                        body,
+                    )
+        result: dict[str, Any] = {'skill_memories': skill_updates}
+        if user_lines != _memory_body_lines(existing_user):
+            result['user_memory'] = _render_memory_doc('用户记忆', user_lines)
+        return result
+
+    def _insert_event(self, event: MemoryEvent) -> None:
+        self.ensure_account(event.account_id)
+        conn = self._connect(event.account_id)
+        try:
+            conn.execute(
+                '''
+                insert into memory_events (
+                    id, account_id, session_id, interaction_id, user_prompt,
+                    assistant_output, skills_json, signals_json, model, priority,
+                    status, created_at, updated_at
+                )
+                values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?)
+                ''',
+                (
+                    uuid4().hex,
+                    event.account_id,
+                    event.session_id,
+                    event.interaction_id,
+                    event.user_prompt,
+                    event.assistant_output,
+                    json.dumps(list(event.skills), ensure_ascii=False),
+                    json.dumps(list(event.signals), ensure_ascii=False),
+                    event.model,
+                    event.priority,
+                    event.created_at,
+                    event.created_at,
+                ),
+            )
+            conn.commit()
+        finally:
+            conn.close()
+
+    def _record_revision(
+        self,
+        account_id: str,
+        kind: str,
+        skill_name: str | None,
+    ) -> None:
+        self.ensure_account(account_id)
+        key = f'{kind}:{skill_name or ""}'
+        conn = self._connect(account_id)
+        try:
+            now = _now_iso()
+            conn.execute(
+                '''
+                insert into memory_revisions (key, kind, skill_name, revision, updated_at)
+                values (?, ?, ?, 1, ?)
+                on conflict(key) do update set
+                    revision = revision + 1,
+                    updated_at = excluded.updated_at
+                ''',
+                (key, kind, skill_name, now),
+            )
+            conn.commit()
+        finally:
+            conn.close()
+
+    def _ensure_db(self, account_id: str) -> None:
+        with self._init_lock:
+            conn = self._connect(account_id)
+            try:
+                conn.executescript(
+                    '''
+                    create table if not exists memory_events (
+                        id text primary key,
+                        account_id text not null,
+                        session_id text not null,
+                        interaction_id text not null,
+                        user_prompt text not null,
+                        assistant_output text not null,
+                        skills_json text not null,
+                        signals_json text not null,
+                        model text not null,
+                        priority integer not null default 0,
+                        status text not null default 'pending',
+                        error text,
+                        created_at text not null,
+                        updated_at text not null
+                    );
+                    create index if not exists idx_memory_events_status
+                        on memory_events(status, priority, created_at);
+                    create table if not exists memory_revisions (
+                        key text primary key,
+                        kind text not null,
+                        skill_name text,
+                        revision integer not null default 1,
+                        updated_at text not null
+                    );
+                    '''
+                )
+                conn.commit()
+            finally:
+                conn.close()
+
+    def _connect(self, account_id: str) -> sqlite3.Connection:
+        path = self._account_memory_root(account_id) / MEMORY_DB_FILENAME
+        path.parent.mkdir(parents=True, exist_ok=True)
+        conn = sqlite3.connect(path, timeout=10)
+        conn.row_factory = sqlite3.Row
+        return conn
+
+    def _account_memory_root(self, account_id: str) -> Path:
+        return self.accounts_root / _safe_account_id(account_id) / 'memory'
+
+    def _list_account_ids(self) -> list[str]:
+        if not self.accounts_root.exists():
+            return []
+        result: list[str] = []
+        for path in sorted(self.accounts_root.iterdir()):
+            if not path.is_dir():
+                continue
+            if (path / 'sessions').exists() or (path / 'memory').exists():
+                result.append(path.name)
+        return result
+
+    @staticmethod
+    def _read_memory_file(path: Path) -> str:
+        try:
+            text = path.read_text(encoding='utf-8')
+        except OSError:
+            return ''
+        if '暂无' in text and len(_memory_body_lines(text)) == 0:
+            return ''
+        return text.strip()
+
+    @staticmethod
+    def _atomic_write_text(path: Path, content: str) -> None:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        tmp = path.with_suffix(path.suffix + '.tmp')
+        tmp.write_text(content.rstrip() + '\n', encoding='utf-8')
+        tmp.replace(path)
+
+
+def detect_memory_signals(
+    user_prompt: str,
+    assistant_output: str,
+    skills: tuple[str, ...],
+) -> tuple[str, ...]:
+    signals: list[str] = []
+    prompt = user_prompt or ''
+    if any(pattern in prompt for pattern in EXPLICIT_MEMORY_PATTERNS):
+        signals.append('explicit')
+    if any(pattern in prompt for pattern in CORRECTION_PATTERNS):
+        signals.append('correction')
+    if skills and any(keyword in prompt for keyword in ('skill', 'Skill', '工具', '流程', '格式')):
+        signals.append('skill')
+    if '模型返回的工具参数不是合法 JSON' in assistant_output:
+        signals.append('tool_experience')
+    return tuple(dict.fromkeys(signals))
+
+
+def _extract_json_object(text: str) -> str:
+    text = text.strip()
+    if text.startswith('```'):
+        text = re.sub(r'^```(?:json)?\s*', '', text)
+        text = re.sub(r'\s*```$', '', text)
+    start = text.find('{')
+    end = text.rfind('}')
+    if start == -1 or end == -1 or end < start:
+        raise ValueError('no JSON object found')
+    return text[start : end + 1]
+
+
+def _normalize_memory_markdown(content: str, title: str) -> str:
+    text = content.strip()
+    if not text:
+        return f'# {title}\n\n暂无记忆。\n'
+    if not text.lstrip().startswith('#'):
+        text = f'# {title}\n\n{text}'
+    return text
+
+
+def _render_memory_doc(title: str, lines: list[str]) -> str:
+    clean_lines = [line.strip() for line in lines if line.strip()]
+    if not clean_lines:
+        return f'# {title}\n\n暂无记忆。\n'
+    return '\n'.join([f'# {title}', '', *[f'- {line}' for line in clean_lines]])
+
+
+def _memory_body_lines(text: str) -> list[str]:
+    lines: list[str] = []
+    for raw in text.splitlines():
+        line = raw.strip()
+        if not line or line.startswith('#') or '暂无' in line:
+            continue
+        if line.startswith('- '):
+            line = line[2:].strip()
+        lines.append(line)
+    return lines
+
+
+def _shorten_memory_line(text: str) -> str:
+    text = re.sub(r'\s+', ' ', text).strip()
+    if not text:
+        return ''
+    return text[:240]
+
+
+def _safe_account_id(value: str) -> str:
+    clean = re.sub(r'[^a-zA-Z0-9._-]+', '_', value.strip())
+    return clean[:80] or 'unknown'
+
+
+def _safe_memory_name(value: str) -> str:
+    clean = re.sub(r'[^a-zA-Z0-9._-]+', '-', value.strip().lower())
+    return clean[:100]
+
+
+def _line_count(path: Path) -> int:
+    try:
+        return len(path.read_text(encoding='utf-8').splitlines())
+    except OSError:
+        return 0
+
+
+def _mtime_iso(path: Path) -> str | None:
+    try:
+        return datetime.fromtimestamp(path.stat().st_mtime, tz=timezone.utc).isoformat()
+    except OSError:
+        return None
+
+
+def _loads_json_list(value: str | None) -> list[str]:
+    if not value:
+        return []
+    try:
+        parsed = json.loads(value)
+    except json.JSONDecodeError:
+        return []
+    if not isinstance(parsed, list):
+        return []
+    return [str(item) for item in parsed if str(item).strip()]
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()