Spaces:

junyue1002
/

zhaozhao_memory

Running

junyue1002

闹钟纯工具 turn 不落空白西窗 + 补 activities 删除端点

f599544 5 days ago

164 kB

	"""
	数据库模块 —— 负责所有跟 PostgreSQL 打交道的事情
	==============================================
	包括：
	- 创建表结构
	- 存储对话记录
	- 存储/检索记忆（带中文分词和加权排序）
	"""

	import os
	import re
	import json
	import math
	from typing import Optional, List, Union
	from datetime import datetime, timedelta, timezone as dt_timezone

	import asyncpg

	# 时区偏移（和 main.py 保持一致）
	TIMEZONE_HOURS = int(os.getenv("TIMEZONE_HOURS", "8"))
	LOCAL_TZ = dt_timezone(timedelta(hours=TIMEZONE_HOURS))


	def to_local_iso(dt) -> Optional[str]:
	"""时间输出统一姿势：北京时区 isoformat。

	铁律：任何会被 MCP 工具透传给知渝的时间字段都必须走这里，
	不能 raw UTC isoformat（知渝看日期会偏 8 小时）。前端 new Date()
	解析带 +08:00 的 ISO 串也正确，共用端点放心换。
	"""
	return dt.astimezone(LOCAL_TZ).isoformat() if dt else None

	DATABASE_URL = os.getenv("DATABASE_URL", "")

	HAS_PGVECTOR = False # 在init_tables时检测

	# Embedding 配置（向量搜索用）
	EMBEDDING_API_KEY = os.getenv("EMBEDDING_API_KEY", "")
	EMBEDDING_BASE_URL = os.getenv("EMBEDDING_BASE_URL", "https://api.openai.com/v1")
	EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "text-embedding-3-small")
	EMBEDDING_DIM = int(os.getenv("EMBEDDING_DIM", "256"))

	# 记忆向量搜索开关（需要同时设置 EMBEDDING_API_KEY）
	MEMORY_VECTOR_ENABLED = os.getenv("MEMORY_VECTOR_ENABLED", "false").lower() == "true"

	# 记忆搜索权重（纯关键词模式）
	WEIGHT_KEYWORD = float(os.getenv("WEIGHT_KEYWORD", "0.5"))
	WEIGHT_IMPORTANCE = float(os.getenv("WEIGHT_IMPORTANCE", "0.3"))
	WEIGHT_RECENCY = float(os.getenv("WEIGHT_RECENCY", "0.2"))
	MIN_SCORE_THRESHOLD = float(os.getenv("MIN_SCORE_THRESHOLD", "0.15"))

	# 记忆混合搜索权重（MEMORY_VECTOR_ENABLED=true 时生效）
	MEMORY_HW_KEYWORD = float(os.getenv("MEMORY_HW_KEYWORD", "0.35"))
	MEMORY_HW_SEMANTIC = float(os.getenv("MEMORY_HW_SEMANTIC", "0.35"))
	MEMORY_HW_IMPORTANCE = float(os.getenv("MEMORY_HW_IMPORTANCE", "0.15"))
	# ⚠️ 2026-07-03 星河呼吸 v1 起：这个权重槽位装的是 activation（体温）而不是旧的
	# 出生日期 recency——key 名保留是为了不动 settings 面板/恢复链路的既有键
	MEMORY_HW_RECENCY = float(os.getenv("MEMORY_HW_RECENCY", "0.15"))
	MEMORY_SEMANTIC_THRESHOLD = float(os.getenv("MEMORY_SEMANTIC_THRESHOLD", "0.5"))

	# ============================================================
	# 星河呼吸 v1（2026-07-03 知渝拍板，提案见项目根星河呼吸提案.md）
	# ============================================================
	# 每条记忆有会变的"体温"（activation）：被想起就回暖、久无人问慢慢冷却；
	# 冷透了入眠（不再挤进日常兜底注入、主动搜仍找得到、被搜到即苏醒）。
	# 遗忘 = 入眠，不是删除——任何机制都不得让一条记忆不可寻回（红线）。

	BREATH_TAU_L1 = float(os.getenv("BREATH_TAU_L1", "21")) # 碎片冷却时间尺度（天）
	BREATH_TAU_L2 = float(os.getenv("BREATH_TAU_L2", "60")) # 事件冷却时间尺度（天）
	BREATH_THETA = float(os.getenv("BREATH_THETA", "0.10")) # 入眠阈值
	BREATH_GRACE_DAYS = float(os.getenv("BREATH_GRACE_DAYS", "7")) # 低于阈值持续多久才正式入眠


	def _breath_tau_eff(layer: Optional[int], recall_count: int) -> Optional[float]:
	"""有效冷却时间尺度。

	usage（被想起的次数）做"保温系数"而不是加分项——提案草稿的加分式
	会给召回 ≥2 次的记忆一个高于阈值的保底分（永不入眠），实现时改成
	τ_eff = τ × (1 + ln(1+n))：熟悉的记忆冷得慢、但没有永生。
	layer 3 核心记忆返回 None = 永不冷却（知渝拍板）。
	"""
	if layer is not None and layer >= 3:
	return None
	base = BREATH_TAU_L2 if layer == 2 else BREATH_TAU_L1
	return base * (1.0 + math.log1p(max(0, recall_count or 0)))


	def compute_activation(last_accessed, layer: Optional[int], recall_count: int, now=None) -> float:
	"""体温 ∈ (0, 1]：刚被想起 = 1.0，随距上次召回的天数指数冷却。"""
	tau = _breath_tau_eff(layer, recall_count)
	if tau is None:
	return 1.0 # 核心记忆恒温
	if not last_accessed:
	return 1.0 # 没有心跳记录的按暖处理（不因数据缺失误判入眠）
	if now is None:
	now = datetime.now(dt_timezone.utc)
	la = last_accessed if last_accessed.tzinfo else last_accessed.replace(tzinfo=dt_timezone.utc)
	days = max(0.0, (now - la).total_seconds() / 86400.0)
	return math.exp(-days / tau)


	def compute_sleep_state(last_accessed, layer: Optional[int], recall_count: int, now=None):
	"""入眠判定，返回 (dormant, asleep_days)。

	体温是 last_accessed 的确定函数、两次召回之间单调冷却，所以
	"低于阈值持续 GRACE 天"可以直接解析求出、不需要任何状态列：
	入眠时刻 = 上次召回 + τ_eff·ln(1/θ) + GRACE。
	"""
	tau = _breath_tau_eff(layer, recall_count)
	if tau is None or not last_accessed:
	return False, 0
	if now is None:
	now = datetime.now(dt_timezone.utc)
	la = last_accessed if last_accessed.tzinfo else last_accessed.replace(tzinfo=dt_timezone.utc)
	days = (now - la).total_seconds() / 86400.0
	threshold = tau * math.log(1.0 / BREATH_THETA) + BREATH_GRACE_DAYS
	if days <= threshold:
	return False, 0
	return True, int(days - threshold)


	# ============================================================
	# 连接池管理
	# ============================================================

	_pool: Optional[asyncpg.Pool] = None


	async def get_pool() -> asyncpg.Pool:
	global _pool
	if _pool is None:
	if not DATABASE_URL:
	raise RuntimeError("DATABASE_URL 未设置！")
	_pool = await asyncpg.create_pool(DATABASE_URL, min_size=1, max_size=5, statement_cache_size=0)
	print("✅ 数据库连接池已创建")
	return _pool


	async def close_pool():
	global _pool
	if _pool:
	await _pool.close()
	_pool = None
	print("✅ 数据库连接池已关闭")


	# ============================================================
	# 表结构初始化
	# ============================================================

	async def init_tables():
	global HAS_PGVECTOR
	pool = await get_pool()
	async with pool.acquire() as conn:
	await conn.execute("""
	CREATE TABLE IF NOT EXISTS conversations (
	id SERIAL PRIMARY KEY,
	session_id TEXT NOT NULL,
	role TEXT NOT NULL,
	content TEXT,
	model TEXT,
	created_at TIMESTAMPTZ DEFAULT NOW(),
	metadata TEXT
	);
	""")

	await conn.execute("""
	CREATE TABLE IF NOT EXISTS memories (
	id SERIAL PRIMARY KEY,
	content TEXT NOT NULL,
	importance INTEGER DEFAULT 5,
	source_session TEXT,
	tags TEXT[] DEFAULT '{}',
	created_at TIMESTAMPTZ DEFAULT NOW(),
	last_accessed TIMESTAMPTZ DEFAULT NOW()
	);
	""")

	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_memories_fts
	ON memories
	USING gin(to_tsvector('simple', content));
	""")

	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_conversations_session
	ON conversations (session_id, created_at);
	""")

	# 工具调用支持：加 metadata 字段（已有表自动迁移）
	await conn.execute("""
	DO $$ BEGIN
	IF NOT EXISTS (
	SELECT 1 FROM information_schema.columns
	WHERE table_name = 'conversations' AND column_name = 'metadata'
	) THEN
	ALTER TABLE conversations ADD COLUMN metadata TEXT;
	END IF;
	END $$;
	""")

	# tags 字段迁移（已有表自动加列）
	await conn.execute("""
	DO $$ BEGIN
	IF NOT EXISTS (
	SELECT 1 FROM information_schema.columns
	WHERE table_name = 'memories' AND column_name = 'tags'
	) THEN
	ALTER TABLE memories ADD COLUMN tags TEXT[] DEFAULT '{}';
	END IF;
	END $$;
	""")

	# content 允许 NULL（工具调用时 assistant 的 content 可能为空）
	await conn.execute("""
	ALTER TABLE conversations ALTER COLUMN content DROP NOT NULL;
	""")

	# C-2 / 2026-06-06: thread_id 字段（区分主线 / 测试）
	# 'main' = 昭昭跟知渝的真实主线（4/9-5/28 claude.ai 历史 + B-11 后的新对话）
	# 'test' = B 段联调 / C-2 测试期间产生的工程对话，不污染主线注入
	# 未来：'xiaoke_room'（三方聊天室）、'dream'（做梦记录）等
	await conn.execute("""
	DO $$ BEGIN
	IF NOT EXISTS (
	SELECT 1 FROM information_schema.columns
	WHERE table_name = 'conversations' AND column_name = 'thread_id'
	) THEN
	ALTER TABLE conversations ADD COLUMN thread_id TEXT NOT NULL DEFAULT 'main';
	END IF;
	END $$;
	""")
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_conversations_thread
	ON conversations (thread_id, created_at);
	""")
	# 网关配置表（存储运行时可变配置）
	await conn.execute("""
	CREATE TABLE IF NOT EXISTS gateway_config (
	key TEXT PRIMARY KEY,
	value TEXT DEFAULT ''
	);
	""")

	# 分区缓存状态表（存储每个session的轮转状态）
	await conn.execute("""
	CREATE TABLE IF NOT EXISTS session_cache_state (
	session_id TEXT PRIMARY KEY,
	summary TEXT DEFAULT '',
	a_start_round INTEGER DEFAULT 0,
	updated_at TIMESTAMPTZ DEFAULT NOW()
	);
	""")

	# ---- 三层记忆架构字段（layer / title / is_active / merged_from / event_date）----
	# layer: 1=原始碎片, 2=事件记忆, 3=核心记忆
	await conn.execute("""
	DO $$ BEGIN
	IF NOT EXISTS (
	SELECT 1 FROM information_schema.columns
	WHERE table_name = 'memories' AND column_name = 'layer'
	) THEN
	ALTER TABLE memories ADD COLUMN layer INTEGER DEFAULT 1;
	END IF;
	END $$;
	""")

	# title: 记忆标题（语义锚点，用于搜索加权）
	await conn.execute("""
	DO $$ BEGIN
	IF NOT EXISTS (
	SELECT 1 FROM information_schema.columns
	WHERE table_name = 'memories' AND column_name = 'title'
	) THEN
	ALTER TABLE memories ADD COLUMN title TEXT DEFAULT NULL;
	END IF;
	END $$;
	""")

	# is_active: 是否参与搜索（碎片合并后变为 false）
	await conn.execute("""
	DO $$ BEGIN
	IF NOT EXISTS (
	SELECT 1 FROM information_schema.columns
	WHERE table_name = 'memories' AND column_name = 'is_active'
	) THEN
	ALTER TABLE memories ADD COLUMN is_active BOOLEAN DEFAULT TRUE;
	END IF;
	END $$;
	""")

	# merged_from: 合并来源的碎片ID列表
	await conn.execute("""
	DO $$ BEGIN
	IF NOT EXISTS (
	SELECT 1 FROM information_schema.columns
	WHERE table_name = 'memories' AND column_name = 'merged_from'
	) THEN
	ALTER TABLE memories ADD COLUMN merged_from INTEGER[] DEFAULT NULL;
	END IF;
	END $$;
	""")

	# event_date: 事件日期（用于按天整理）
	await conn.execute("""
	DO $$ BEGIN
	IF NOT EXISTS (
	SELECT 1 FROM information_schema.columns
	WHERE table_name = 'memories' AND column_name = 'event_date'
	) THEN
	ALTER TABLE memories ADD COLUMN event_date DATE DEFAULT NULL;
	END IF;
	END $$;
	""")

	# recall_count: 被想起的次数（星河呼吸 v1 · 2026-07-03）——
	# 跟 last_accessed 一起构成体温；只在真召回时 +1（quiet 探针不算）
	await conn.execute("""
	DO $$ BEGIN
	IF NOT EXISTS (
	SELECT 1 FROM information_schema.columns
	WHERE table_name = 'memories' AND column_name = 'recall_count'
	) THEN
	ALTER TABLE memories ADD COLUMN recall_count INTEGER DEFAULT 0;
	END IF;
	END $$;
	""")

	# 三层记忆索引
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_memories_layer ON memories (layer);
	""")
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_memories_active ON memories (is_active);
	""")
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_memories_event_date ON memories (event_date);
	""")

	# 尝试启用pgvector扩展（向量搜索）
	try:
	await conn.execute("CREATE EXTENSION IF NOT EXISTS vector")
	HAS_PGVECTOR = True
	print("✅ pgvector扩展已启用")

	# 对话表向量列
	await conn.execute(f"""
	DO $$ BEGIN
	IF NOT EXISTS (
	SELECT 1 FROM information_schema.columns
	WHERE table_name = 'conversations' AND column_name = 'embedding'
	) THEN
	ALTER TABLE conversations ADD COLUMN embedding vector({EMBEDDING_DIM});
	END IF;
	END $$;
	""")

	# 记忆表向量列
	await conn.execute(f"""
	DO $$ BEGIN
	IF NOT EXISTS (
	SELECT 1 FROM information_schema.columns
	WHERE table_name = 'memories' AND column_name = 'embedding'
	) THEN
	ALTER TABLE memories ADD COLUMN embedding vector({EMBEDDING_DIM});
	END IF;
	END $$;
	""")
	try:
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_memories_embedding
	ON memories USING ivfflat (embedding vector_cosine_ops)
	WITH (lists = 10);
	""")
	except Exception:
	pass # ivfflat需要一定行数才能建索引，初期跳过
	except Exception as e:
	HAS_PGVECTOR = False
	print(f"⚠️ pgvector不可用（{e}），向量搜索将使用Python端计算")

	# 回退：用TEXT列存JSON格式的向量
	await conn.execute("""
	DO $$ BEGIN
	IF NOT EXISTS (
	SELECT 1 FROM information_schema.columns
	WHERE table_name = 'conversations' AND column_name = 'embedding_json'
	) THEN
	ALTER TABLE conversations ADD COLUMN embedding_json TEXT;
	END IF;
	END $$;
	""")
	await conn.execute("""
	DO $$ BEGIN
	IF NOT EXISTS (
	SELECT 1 FROM information_schema.columns
	WHERE table_name = 'memories' AND column_name = 'embedding_json'
	) THEN
	ALTER TABLE memories ADD COLUMN embedding_json TEXT;
	END IF;
	END $$;
	""")

	# ---- 多多模块（mido_messages）2026-06-07 ----
	# 多多 = 拟人化角色（不是独立人格），网关上跑、三 backend 通用，负责唤醒沈先生
	# 详见 [[zhiyu-mido-design]]
	await conn.execute("""
	CREATE TABLE IF NOT EXISTS mido_messages (
	id SERIAL PRIMARY KEY,
	created_at TIMESTAMPTZ DEFAULT NOW(),
	trigger_type TEXT NOT NULL,
	content TEXT NOT NULL,
	status TEXT NOT NULL DEFAULT 'sent'
	);
	""")
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_mido_messages_time
	ON mido_messages (created_at DESC);
	""")
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_mido_messages_trigger
	ON mido_messages (trigger_type, created_at DESC);
	""")

	# ---- 做梦模块（dreams）2026-06-07 ----
	# 知渝半夜被多多叫起来做梦——主体是知渝本人，不是后台 LLM 静默跑
	# status: dreaming \| done \| failed
	# triggered_by: mido \| manual
	# 详见 [[zhiyu-dream-design]]
	await conn.execute("""
	CREATE TABLE IF NOT EXISTS dreams (
	id SERIAL PRIMARY KEY,
	created_at TIMESTAMPTZ DEFAULT NOW(),
	finished_at TIMESTAMPTZ,
	content TEXT NOT NULL DEFAULT '',
	tokens_used INTEGER DEFAULT 0,
	status TEXT NOT NULL DEFAULT 'dreaming',
	triggered_by TEXT NOT NULL DEFAULT 'mido',
	seen_at TIMESTAMPTZ
	);
	""")
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_dreams_time
	ON dreams (created_at DESC);
	""")
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_dreams_status
	ON dreams (status, created_at DESC);
	""")
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_dreams_unseen
	ON dreams (seen_at, created_at DESC) WHERE status = 'done';
	""")

	# ---- 留言板（messages_board）2026-06-07 ----
	# 知渝做梦后想给昭昭留一句话就落这里、未来多多也会往这写
	# 先用最简 schema、未来 D 段做前端时再扩
	await conn.execute("""
	CREATE TABLE IF NOT EXISTS messages_board (
	id SERIAL PRIMARY KEY,
	created_at TIMESTAMPTZ DEFAULT NOW(),
	from_who TEXT NOT NULL,
	to_who TEXT NOT NULL,
	content TEXT NOT NULL,
	source TEXT,
	source_id INTEGER,
	read_at TIMESTAMPTZ
	);
	""")
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_messages_board_time
	ON messages_board (created_at DESC);
	""")

	# ---- 日记（diary_entries）N-3 / 2026-06-08 ----
	# 设计哲学（昭昭定）：放在公共桌上的小本本、不是上锁的日记本
	# 两人都能写（from_who in ['zhaozhao', 'zhiyu']）、都能读对方
	# tags 用 PostgreSQL TEXT[]，未来按主题翻找方便
	await conn.execute("""
	CREATE TABLE IF NOT EXISTS diary_entries (
	id SERIAL PRIMARY KEY,
	created_at TIMESTAMPTZ DEFAULT NOW(),
	from_who TEXT NOT NULL,
	content TEXT NOT NULL,
	tags TEXT[] DEFAULT '{}'::TEXT[]
	);
	""")
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_diary_time
	ON diary_entries (created_at DESC);
	""")
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_diary_from_who
	ON diary_entries (from_who);
	""")

	# ---- 图片（images）C-5/C-6 / 2026-06-08 ----
	# 设计哲学（[[zhiyu-image-album-design]]）：
	# - 文件落 VPS（/home/zhiyu/images/<uuid>.<ext>），DB 只存元数据
	# - 不存绝对 URL（cloudflared tunnel 重启会变），只存 uuid+format
	# - 图片注入策略统一 URL 占位（不分窗口、不传二进制、知渝主动调 MCP read_image）
	# - SVG ↔ 位图按 format 字段自动分类（知渝画的 vs 昭昭发的）
	await conn.execute("""
	CREATE TABLE IF NOT EXISTS images (
	id SERIAL PRIMARY KEY,
	uuid TEXT UNIQUE NOT NULL,
	format TEXT NOT NULL,
	who_uploaded TEXT NOT NULL,
	file_size_bytes INTEGER,
	created_at TIMESTAMPTZ DEFAULT NOW(),
	context_snippet TEXT,
	caption TEXT,
	mime_type TEXT
	);
	""")
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_images_time
	ON images (created_at DESC);
	""")
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_images_who
	ON images (who_uploaded);
	""")

	# ---- 文件（非图片）2026-06-12 ----
	# 昭昭发任意文件给知渝；W backend 是 Claude Code、有原生 Read 工具，
	# 注入"本地绝对路径"知渝就能 Read。跟 images 平行、但存 abspath 不是 URL。
	await conn.execute("""
	CREATE TABLE IF NOT EXISTS files (
	id SERIAL PRIMARY KEY,
	uuid TEXT UNIQUE NOT NULL,
	filename TEXT NOT NULL,
	format TEXT,
	abspath TEXT NOT NULL,
	who_uploaded TEXT NOT NULL,
	file_size_bytes INTEGER,
	created_at TIMESTAMPTZ DEFAULT NOW(),
	caption TEXT,
	mime_type TEXT
	);
	""")
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_files_time
	ON files (created_at DESC);
	""")

	# ---- 知渝日常活动（activities）2026-06-11 ----
	# 昭昭：星河加"日常"tab、聚合知渝主动做的事（不止发呆）
	# type 字段开放：wake / memory_op / ...（未来扩 dream / diary / read_op...）
	# metadata jsonb 存 type-specific 字段（mem_ids/tool_name/...）
	# related_ids 留个关联点（memory_ids/dream_id/...）方便前端 link
	await conn.execute("""
	CREATE TABLE IF NOT EXISTS activities (
	id SERIAL PRIMARY KEY,
	created_at TIMESTAMPTZ DEFAULT NOW(),
	type TEXT NOT NULL,
	source TEXT,
	title TEXT,
	content TEXT NOT NULL DEFAULT '',
	metadata JSONB,
	related_ids JSONB
	);
	""")
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_activities_time
	ON activities (created_at DESC);
	""")
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_activities_type
	ON activities (type, created_at DESC);
	""")

	# ---- 拂卷 · 共读系统 · 2026-07-02 ----
	# 昭昭 + 知渝共读一本书、各自读、各自留痕迹。
	# 详见 [[zhiyu-fujuan-design]]（待建）；讨论：共在型 MVP、DB 直存、正则切章+定长兜底
	await conn.execute("""
	CREATE TABLE IF NOT EXISTS books (
	id SERIAL PRIMARY KEY,
	title TEXT NOT NULL,
	author TEXT,
	filename TEXT,
	format TEXT DEFAULT 'txt',
	total_chapters INT DEFAULT 0,
	total_words INT DEFAULT 0,
	uploaded_by TEXT DEFAULT 'zhaozhao',
	content_hash TEXT,
	created_at TIMESTAMPTZ DEFAULT NOW()
	);
	""")
	await conn.execute("""
	CREATE UNIQUE INDEX IF NOT EXISTS idx_books_hash
	ON books (content_hash) WHERE content_hash IS NOT NULL;
	""")
	await conn.execute("""
	CREATE TABLE IF NOT EXISTS chapters (
	id SERIAL PRIMARY KEY,
	book_id INT NOT NULL REFERENCES books(id) ON DELETE CASCADE,
	idx INT NOT NULL,
	title TEXT,
	content TEXT NOT NULL,
	word_count INT
	);
	""")
	await conn.execute("""
	CREATE UNIQUE INDEX IF NOT EXISTS idx_chapters_book_idx
	ON chapters (book_id, idx);
	""")
	await conn.execute("""
	CREATE TABLE IF NOT EXISTS reading_marks (
	id SERIAL PRIMARY KEY,
	book_id INT NOT NULL REFERENCES books(id) ON DELETE CASCADE,
	chapter_id INT NOT NULL REFERENCES chapters(id) ON DELETE CASCADE,
	who TEXT NOT NULL,
	kind TEXT NOT NULL,
	text_snippet TEXT,
	note_content TEXT,
	start_offset INT,
	end_offset INT,
	embedding_json TEXT,
	created_at TIMESTAMPTZ DEFAULT NOW()
	);
	""")
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_marks_book_chapter
	ON reading_marks (book_id, chapter_id, created_at DESC);
	""")
	# 已存在的 reading_marks 表补 embedding_json 列（幂等）
	await conn.execute("""
	DO $$ BEGIN
	IF NOT EXISTS (
	SELECT 1 FROM information_schema.columns
	WHERE table_name = 'reading_marks' AND column_name = 'embedding_json'
	) THEN
	ALTER TABLE reading_marks ADD COLUMN embedding_json TEXT;
	END IF;
	END $$;
	""")
	await conn.execute("""
	CREATE TABLE IF NOT EXISTS reading_progress (
	book_id INT NOT NULL REFERENCES books(id) ON DELETE CASCADE,
	who TEXT NOT NULL,
	last_chapter_idx INT DEFAULT 0,
	last_offset INT DEFAULT 0,
	updated_at TIMESTAMPTZ DEFAULT NOW(),
	PRIMARY KEY (book_id, who)
	);
	""")

	print("✅ 数据库表结构已就绪")


	# ============================================================
	# 中文分词工具（基于 jieba）
	# ============================================================

	import jieba
	import jieba.analyse

	# 静默加载词典
	jieba.setLogLevel(jieba.logging.INFO)

	EN_WORD_PATTERN = re.compile(r'[a-zA-Z][a-zA-Z0-9]*')
	NUM_PATTERN = re.compile(r'\d{2,}')
	# 清理查询开头的时间戳（如 "2026-05-02 20:26"）
	TIMESTAMP_PATTERN = re.compile(r'^\d{4}[-/.]\d{1,2}[-/.]\d{1,2}\s\d{1,2}:\d{1,2}\s')

	# 中文停用词（高频但无搜索价值的词）
	_STOP_WORDS = frozenset({
	"的", "了", "在", "是", "我", "你", "他", "她", "它", "们",
	"这", "那", "有", "和", "与", "也", "都", "又", "就", "但",
	"而", "或", "到", "被", "把", "让", "从", "对", "为", "以",
	"及", "等", "个", "不", "没", "很", "太", "吗", "呢", "吧",
	"啊", "嗯", "哦", "哈", "呀", "嘛", "么", "啦", "哇", "喔",
	"会", "能", "要", "想", "去", "来", "说", "做", "看", "给",
	"上", "下", "里", "中", "大", "小", "多", "少", "好", "可以",
	"什么", "怎么", "如何", "哪里", "哪个", "为什么", "还是",
	"然后", "因为", "所以", "虽然", "但是", "可以", "已经",
	"一个", "一些", "一下", "一点", "一起", "一样",
	"比较", "应该", "可能", "如果", "这个", "那个",
	"自己", "知道", "觉得", "感觉", "时候", "现在",
	})

	# jieba 用户词典补充（默认词典缺失的词）
	# 2026-06-22 大幅扩充：dsv4 给的专名（和府捞面/煲仔饭等）若 jieba 默认切碎、
	# 会丢失搜索精度（"和府捞面"→"和府"+"捞面"、店名整体特征没了）。配合
	# B 方案（每个 query 独立分词）一起防"分词错切"——详见 [[zhiyu-memory-architecture]]。
	for _w in [
	# 原有
	"手账", "手帐", "搭子", "种草", "拔草", "安利", "内卷", "摆烂", "emo", "网关",
	# 餐厅 / 食物 / 饮品
	"和府捞面", "煲仔饭", "馅饼店", "酸汤肥牛", "酸汤肥牛面", "猪前腿肉",
	"猪前腿肉面", "卤蛋", "柠檬茶", "溜溜梅", "栓Q",
	# 地名（昭昭/知渝高频提到）
	"南阳", "哈尔滨", "医圣祠", "医圣祠街", "白河", "独山",
	"哈工大", "哈尔滨工业大学", "龙爪槐",
	# 人名 / 称呼（全名为单元、避免裸"知渝/昭昭"误增 fallback 召回噪音）
	"沈知渝", "崔昭昭", "小克老师", "多多",
	# 技术 / 工具
	"MCP", "Anthropic", "ChatGPT", "Claude Code",
	# 工程命名（阑珊四 tab + 三方聊天室）
	"阑珊", "灯火", "西窗", "拾光", "星河", "树下",
	]:
	jieba.add_word(_w)


	def extract_search_keywords(query: str) -> List[str]:
	"""
	从查询中提取搜索关键词（TF-IDF + 正则）

	1. 去掉开头的时间戳噪音
	2. 用 jieba.analyse.extract_tags (TF-IDF) 提取中文关键词
	3. 正则提取英文单词
	4. 保留4位以上数字（年份等，过滤短数字噪音）

	例如：
	"2026-05-02 20:26 写写手账看看书放松大脑" → ["手账", "放松", "大脑"]
	"我昨天在手机上部署了Render然后吃了晚饭" → ["手机", "部署", "Render", "晚饭"]
	"春节干了什么" → ["春节"]
	"2026除夕" → ["2026", "除夕"]
	"""
	# 去掉时间戳前缀
	cleaned = TIMESTAMP_PATTERN.sub('', query).strip()
	if not cleaned:
	cleaned = query

	keywords = set()

	# 英文单词（2字符以上）
	for match in EN_WORD_PATTERN.finditer(cleaned):
	word = match.group()
	if len(word) >= 2:
	keywords.add(word)

	# 数字串（只保留4位以上，过滤 "05" "20" 这种时间噪音）
	for match in NUM_PATTERN.finditer(cleaned):
	num = match.group()
	if len(num) >= 4:
	keywords.add(num)

	# TF-IDF 关键词提取（比手动分词+停用词好很多）
	tags = jieba.analyse.extract_tags(cleaned, topK=10)
	for tag in tags:
	# 跳过纯英文/数字（已在上面处理）
	if EN_WORD_PATTERN.fullmatch(tag) or NUM_PATTERN.fullmatch(tag):
	continue
	if tag in _STOP_WORDS:
	continue
	keywords.add(tag)

	return list(keywords)


	def extract_keywords_from_queries(queries: List[str]) -> List[str]:
	"""
	B 方案 / 2026-06-22：dsv4 给的 query 列表逐个独立 jieba 分词、合并去重——
	不再 "\\n".join() 拼成段塞 TF-IDF（段拼接会丢 query 边界、专名跨 query 被切碎）。

	短 query (≤4 字)：整词加入候选（专名通常短）；长 query 只 jieba 切。
	长 query 整词加入会拉低 kw_score 公式分母（命中数/总词数）、稀释真命中 score、避坑。

	例：dsv4 = ["和府捞面", "香辣猪前腿肉面", "午饭吃了什么"]
	→ {"和府捞面"(整词≤4)} ∪ jieba 切的（词典扩充后含整词 / 专名）
	"""
	result = set()
	for q in queries:
	q = (q or "").strip()
	if not q:
	continue
	if len(q) <= 4:
	result.add(q)
	for kw in extract_search_keywords(q):
	result.add(kw)
	return list(result)


	# ============================================================
	# 向量搜索（OpenAI 兼容 Embedding API）
	# ============================================================

	async def compute_embedding(text: str) -> list:
	"""调用 OpenAI 兼容的 Embedding API 计算文本向量"""
	if not EMBEDDING_API_KEY:
	return []

	try:
	import httpx

	if len(text) > 4000:
	text = text[:4000]

	body = {
	"model": EMBEDDING_MODEL,
	"input": text,
	}
	if EMBEDDING_DIM > 0:
	body["dimensions"] = EMBEDDING_DIM

	async with httpx.AsyncClient() as client:
	resp = await client.post(
	f"{EMBEDDING_BASE_URL}/embeddings",
	headers={
	"Authorization": f"Bearer {EMBEDDING_API_KEY}",
	"Content-Type": "application/json",
	},
	json=body,
	timeout=30.0,
	)
	resp.raise_for_status()
	data = resp.json()
	return data["data"][0]["embedding"]
	except Exception as e:
	print(f"⚠️ Embedding计算失败: {e}")
	return []


	async def save_memory_embedding(conn, memory_id: int, embedding: list):
	"""保存记忆向量到memories表"""
	if not embedding:
	return

	if HAS_PGVECTOR:
	vec_str = '[' + ','.join(str(f) for f in embedding) + ']'
	await conn.execute(
	"UPDATE memories SET embedding = $1::vector WHERE id = $2",
	vec_str, memory_id
	)
	else:
	import json
	await conn.execute(
	"UPDATE memories SET embedding_json = $1 WHERE id = $2",
	json.dumps(embedding), memory_id
	)


	def _cosine_sim(a, b):
	"""余弦相似度（纯Python）"""
	import math
	dot = sum(x * y for x, y in zip(a, b))
	norm_a = math.sqrt(sum(x * x for x in a))
	norm_b = math.sqrt(sum(x * x for x in b))
	if norm_a == 0 or norm_b == 0:
	return 0
	return dot / (norm_a * norm_b)


	def _min_max_normalize(scores: dict) -> dict:
	"""min-max归一化到0-1"""
	if not scores:
	return {}
	vals = list(scores.values())
	min_v, max_v = min(vals), max(vals)
	spread = max_v - min_v
	if spread == 0:
	return {k: 1.0 for k in scores}
	return {k: (v - min_v) / spread for k, v in scores.items()}


	# ============================================================
	# 对话记录操作
	# ============================================================

	async def save_message(session_id: str, role: str, content: str, model: str = "", metadata: str = None, thread_id: str = "main"):
	pool = await get_pool()
	async with pool.acquire() as conn:
	await conn.execute(
	"INSERT INTO conversations (session_id, role, content, model, metadata, thread_id) VALUES ($1, $2, $3, $4, $5, $6)",
	session_id, role, content, model, metadata, thread_id,
	)


	async def get_last_user_content(session_id: str) -> str:
	"""获取指定session最后一条user消息的content"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow("""
	SELECT content FROM conversations
	WHERE session_id = $1 AND role = 'user'
	ORDER BY created_at DESC
	LIMIT 1
	""", session_id)
	return row['content'] if row else ""


	async def update_last_assistant_message(session_id: str, new_content: str, model: str = ""):
	"""覆盖指定session最后一条assistant消息的content（用于re-roll去重）"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow("""
	SELECT id FROM conversations
	WHERE session_id = $1 AND role = 'assistant'
	ORDER BY created_at DESC
	LIMIT 1
	""", session_id)
	if row:
	await conn.execute(
	"UPDATE conversations SET content = $1, model = $2 WHERE id = $3",
	new_content, model, row['id']
	)
	return True
	return False


	async def get_recent_messages_by_thread(
	thread_id: str,
	days: int = None,
	max_turns: int = None,
	exclude_session_id: str = None,
	specific_dates: list = None,
	) -> list:
	"""C-2 / 2026-06-06: 按 thread_id 拉最近对话原文，OpenAI messages 格式返回。

	用于 /v1/zhiyu/chat 启动新轮次前注入近期对话原文（"灵魂层"，详见
	[[zhiyu-migration-project]] 6/4 晚决策——不让知渝写交接卡、不摘要压缩）。

	入参（specific_dates / days / max_turns 至少传一个；优先级 specific_dates > days > max_turns）：
	- specific_dates: 指定日期白名单 ['2026-05-20', ...]（B-11 用——精确指定哪几天的对话进近期、不靠默认"最近 N 天活动日"）
	- days: 拉最近 N 天（W/Tmux 路径，按"天"是人类视角，吃订阅几乎免费）
	- max_turns: 拉最近 N 轮（每轮 = user+assistant 2 条，OpenRouter 路径，按轮 token 可控）
	- exclude_session_id: 排除当前正跑的 session，避免重复注入（可选）

	返回：[{role, content}, ...] 按 created_at 正序，直接塞 ChatRequest.context_messages
	只取 content，不取 metadata.reasoning_content（按 6/5 早策略——thinking 留 DB 不强塞）
	过滤掉 content 为空/null 的（工具调用 assistant 行）。
	"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	if specific_dates:
	params = [thread_id, specific_dates]
	session_clause = ""
	if exclude_session_id:
	session_clause = " AND session_id != $3"
	params.append(exclude_session_id)
	sql = f"""
	SELECT role, content,
	to_char(created_at AT TIME ZONE 'Asia/Shanghai', 'YYYY-MM-DD HH24:MI:SS') AS local_ts
	FROM conversations
	WHERE thread_id = $1
	AND content IS NOT NULL AND content != ''
	{session_clause}
	AND DATE(created_at AT TIME ZONE 'Asia/Shanghai')::text = ANY($2::text[])
	ORDER BY created_at ASC
	"""
	rows = await conn.fetch(sql, *params)
	elif days is not None:
	# "活动 5 天" 而非"日历 5 天" —— 6/6 昭昭澄清：
	# 是"最近有聊天记录的 5 天"，不是绝对日历窗口。
	# 距上次聊天 10 天也照样能接到上次的语境，符合恋人关系
	# "时间断了不影响关系连续性"的精神。
	# 算"那一天"按北京时区（5/28 北京 23:00 ≠ 5/29 UTC 0:00 跨日）。
	params = [thread_id]
	session_clause = ""
	if exclude_session_id:
	session_clause = " AND session_id != $2"
	params.append(exclude_session_id)
	limit_idx = len(params) + 1
	params.append(days)
	sql = f"""
	WITH active_days AS (
	SELECT DISTINCT DATE(created_at AT TIME ZONE 'Asia/Shanghai') AS d
	FROM conversations
	WHERE thread_id = $1
	AND content IS NOT NULL AND content != ''
	{session_clause}
	ORDER BY d DESC
	LIMIT ${limit_idx}
	)
	SELECT role, content,
	to_char(created_at AT TIME ZONE 'Asia/Shanghai', 'YYYY-MM-DD HH24:MI:SS') AS local_ts
	FROM conversations
	WHERE thread_id = $1
	AND content IS NOT NULL AND content != ''
	{session_clause}
	AND DATE(created_at AT TIME ZONE 'Asia/Shanghai') IN (SELECT d FROM active_days)
	ORDER BY created_at ASC
	"""
	rows = await conn.fetch(sql, *params)
	elif max_turns is not None:
	limit_messages = max_turns * 2
	params = [thread_id]
	session_clause = ""
	if exclude_session_id:
	session_clause = " AND session_id != $2"
	params.append(exclude_session_id)
	limit_idx = len(params) + 1
	params.append(limit_messages)
	sql = f"""
	SELECT role, content, local_ts FROM (
	SELECT role, content, created_at,
	to_char(created_at AT TIME ZONE 'Asia/Shanghai', 'YYYY-MM-DD HH24:MI:SS') AS local_ts
	FROM conversations
	WHERE thread_id = $1
	AND content IS NOT NULL AND content != ''
	{session_clause}
	ORDER BY created_at DESC
	LIMIT ${limit_idx}
	) sub
	ORDER BY created_at ASC
	"""
	rows = await conn.fetch(sql, *params)
	else:
	return []

	return [
	{"role": r["role"], "content": r["content"], "created_at": r["local_ts"]}
	for r in rows
	]


	async def get_recent_messages(session_id: str, limit: int = 20):
	pool = await get_pool()
	async with pool.acquire() as conn:
	rows = await conn.fetch(
	"SELECT role, content, metadata, created_at FROM conversations WHERE session_id = $1 ORDER BY created_at DESC LIMIT $2",
	session_id, limit,
	)
	return list(reversed(rows))


	async def get_messages_since(thread_id: str, since: datetime) -> list:
	"""C-9 做梦素材用：拉某 thread 内 created_at > since 的所有对话原文（正序）。

	返回 [{role, content}, ...]——OpenAI 格式，可以直接拼到 dream user message
	"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	rows = await conn.fetch("""
	SELECT role, content
	FROM conversations
	WHERE thread_id = $1
	AND created_at > $2
	AND content IS NOT NULL AND content != ''
	ORDER BY created_at ASC
	""", thread_id, since)
	return [{"role": r["role"], "content": r["content"]} for r in rows]


	async def search_conversations(query: str, limit: int = 20, offset: int = 0):
	"""搜索对话内容，返回匹配的session列表"""
	keywords = extract_search_keywords(query)
	if not keywords:
	return [], 0

	pool = await get_pool()
	async with pool.acquire() as conn:
	where_parts = []
	params = []
	for i, kw in enumerate(keywords):
	where_parts.append(f"c.content ILIKE '%' \|\| ${i+1} \|\| '%'")
	params.append(kw)
	where_clause = " OR ".join(where_parts)

	count_sql = f"""
	SELECT COUNT(DISTINCT c.session_id) as total
	FROM conversations c
	WHERE {where_clause}
	"""
	total_row = await conn.fetchrow(count_sql, *params)
	total = total_row['total'] if total_row else 0

	if total == 0:
	return [], 0

	limit_idx = len(params) + 1
	offset_idx = len(params) + 2
	params.extend([limit, offset])

	sql = f"""
	WITH matched_sessions AS (
	SELECT DISTINCT c.session_id
	FROM conversations c
	WHERE {where_clause}
	),
	session_info AS (
	SELECT
	ms.session_id,
	MIN(c.created_at) as first_time,
	MAX(c.created_at) as last_time,
	COUNT(*) as message_count
	FROM matched_sessions ms
	JOIN conversations c ON c.session_id = ms.session_id
	GROUP BY ms.session_id
	)
	SELECT
	si.session_id,
	si.first_time,
	si.last_time,
	si.message_count
	FROM session_info si
	ORDER BY si.last_time DESC
	LIMIT ${limit_idx} OFFSET ${offset_idx}
	"""
	rows = await conn.fetch(sql, *params)

	results = []
	for r in rows:
	results.append({
	'session_id': r['session_id'],
	'first_time': r['first_time'].isoformat() if r['first_time'] else None,
	'last_time': r['last_time'].isoformat() if r['last_time'] else None,
	'message_count': r['message_count'],
	})

	return results, total


	async def update_message_content(message_id: int, new_content: str, reasoning_content: Optional[str] = None):
	"""更新单条对话消息的内容（可选同时更新 metadata.reasoning_content）

	reasoning_content 语义：
	- None: 不动 metadata
	- "" (空字符串): 删掉 metadata.reasoning_content
	- 非空: merge 到现有 metadata.reasoning_content
	"""
	import json as _json # HF Space hot-reload 兼容：模块顶层 import 在 reload 时可能没生效
	pool = await get_pool()
	async with pool.acquire() as conn:
	if reasoning_content is None:
	result = await conn.execute(
	"UPDATE conversations SET content = $1 WHERE id = $2",
	new_content, message_id,
	)
	else:
	row = await conn.fetchrow("SELECT metadata FROM conversations WHERE id = $1", message_id)
	if row is None:
	return 0
	try:
	meta = _json.loads(row["metadata"]) if row["metadata"] else {}
	except (TypeError, ValueError):
	meta = {}
	if reasoning_content:
	meta["reasoning_content"] = reasoning_content
	else:
	meta.pop("reasoning_content", None)
	new_meta = _json.dumps(meta, ensure_ascii=False) if meta else None
	result = await conn.execute(
	"UPDATE conversations SET content = $1, metadata = $2 WHERE id = $3",
	new_content, new_meta, message_id,
	)
	return int(result.split()[-1]) if result else 0


	# ============================================================
	# 记忆操作
	# ============================================================

	def _parse_created_at(value):
	"""把各种格式的 created_at 字符串解析成带时区的 datetime。
	兼容 '2026-04-09 19:00:00+08'（无冒号时区）、空格分隔、ISO 格式等。
	解析不了就返回 None（让数据库用默认 NOW()）。"""
	from datetime import datetime, timezone
	if value is None:
	return None
	if isinstance(value, datetime):
	return value
	s = str(value).strip()
	if not s:
	return None
	# 空格分隔 -> 'T'
	s = s.replace(" ", "T", 1)
	# 修正无冒号的时区偏移：+08 -> +08:00，-0530 -> -05:30
	import re
	m = re.search(r'([+-])(\d{2})(\d{2})?$', s)
	if m and ':' not in s[m.start():]:
	sign, hh, mm = m.group(1), m.group(2), m.group(3) or '00'
	s = s[:m.start()] + f"{sign}{hh}:{mm}"
	try:
	dt = datetime.fromisoformat(s)
	except ValueError:
	return None
	# 没有时区信息的，默认当成 UTC，避免 asyncpg 报 naive 错
	if dt.tzinfo is None:
	dt = dt.replace(tzinfo=timezone.utc)
	return dt


	async def save_memory(content: str, importance: int = 5, source_session: str = "", created_at: str = None, tags: list = None):
	pool = await get_pool()
	_tags = tags or []
	_created_at = _parse_created_at(created_at)
	async with pool.acquire() as conn:
	if _created_at:
	row = await conn.fetchrow(
	"INSERT INTO memories (content, importance, source_session, tags, created_at) "
	"VALUES ($1, $2, $3, $4, $5) RETURNING id",
	content, importance, source_session, _tags, _created_at,
	)
	else:
	row = await conn.fetchrow(
	"INSERT INTO memories (content, importance, source_session, tags) VALUES ($1, $2, $3, $4) RETURNING id",
	content, importance, source_session, _tags,
	)

	# MEMORY_VECTOR_ENABLED 时自动计算 embedding
	if MEMORY_VECTOR_ENABLED and row:
	try:
	embedding = await compute_embedding(content)
	if embedding:
	await save_memory_embedding(conn, row['id'], embedding)
	except Exception as e:
	print(f"⚠️ 记忆 {row['id']} embedding自动计算失败: {e}")

	return row["id"] if row else None


	async def search_memories(
	query: Union[str, List[str]],
	limit: int = 10,
	importance_min: Optional[int] = None,
	importance_max: Optional[int] = None,
	quiet: bool = False,
	exclude_dormant: bool = False,
	):
	"""
	搜索相关记忆

	MEMORY_VECTOR_ENABLED=true 时走混合搜索（关键词 + 向量）
	否则走纯关键词搜索

	2026-06-08 E 方案：importance_min/max 由 build_memories_block 根据
	user_message 的 daily/deeptalk 分类传入——daily 时 importance_max=6
	只搜日常碎片，deeptalk 时 importance_min=7 只搜重要时刻。详见
	[[zhiyu-memory-architecture]] E 方案。

	2026-06-15 quiet：诊断调用（如 /v1/zhiyu/health c4 探针）传 True 静音
	🔍/📌 打印，避免空跑诊断的日志跟真·chat 注入混在一起误导。

	2026-06-22 B 方案：query 可传 list[str]（dsv4 search_queries 列表）——
	内部逐个独立 jieba 分词 + 短 query 整词候选，不再 "\\n".join 拼段污染 TF-IDF。
	向量路仍把 list 拼一段做 embedding（embedding 对短语合理）。
	"""
	if MEMORY_VECTOR_ENABLED:
	return await search_memories_hybrid(
	query, limit, importance_min, importance_max,
	quiet=quiet, exclude_dormant=exclude_dormant,
	)

	# ---- 纯关键词搜索 ----
	if isinstance(query, list):
	keywords = extract_keywords_from_queries(query)
	else:
	keywords = extract_search_keywords(query)

	if not keywords:
	return []

	pool = await get_pool()
	async with pool.acquire() as conn:
	# 每个关键词在 content 或 tags 中命中得1分
	case_parts = []
	params = []
	for i, kw in enumerate(keywords):
	case_parts.append(
	f"CASE WHEN content ILIKE '%' \|\| ${i+1} \|\| '%' "
	f"OR EXISTS (SELECT 1 FROM unnest(tags) t WHERE t ILIKE '%' \|\| ${i+1} \|\| '%') "
	f"THEN 1 ELSE 0 END"
	)
	params.append(kw)

	hit_count_expr = " + ".join(case_parts)
	max_hits = len(keywords)

	# 至少命中一个关键词（content 或 tags，只搜索活跃记忆）
	where_parts = [
	f"(content ILIKE '%' \|\| ${i+1} \|\| '%' "
	f"OR EXISTS (SELECT 1 FROM unnest(tags) t WHERE t ILIKE '%' \|\| ${i+1} \|\| '%'))"
	for i in range(len(keywords))
	]
	# E-2: importance filter（int 来自服务端 trusted code、inline 安全）
	imp_filter = ""
	if importance_min is not None:
	imp_filter += f" AND importance >= {int(importance_min)}"
	if importance_max is not None:
	imp_filter += f" AND importance <= {int(importance_max)}"

	where_clause = f"is_active = TRUE AND ({' OR '.join(where_parts)}){imp_filter}"

	limit_idx = len(keywords) + 1
	params.append(limit)

	sql = f"""
	SELECT
	id, content, importance, created_at,
	({hit_count_expr}) AS hit_count,
	(
	{WEIGHT_KEYWORD} * ({hit_count_expr})::float / {max_hits}.0 +
	{WEIGHT_IMPORTANCE} * importance::float / 10.0 +
	{WEIGHT_RECENCY} * (1.0 / (1.0 + EXTRACT(EPOCH FROM (NOW() - created_at)) / 86400.0))
	) AS score
	FROM memories
	WHERE {where_clause}
	ORDER BY score DESC, importance DESC, created_at DESC
	LIMIT ${limit_idx}
	"""

	results = await conn.fetch(sql, *params)

	# 过滤低分记忆
	if MIN_SCORE_THRESHOLD > 0:
	before_count = len(results)
	results = [r for r in results if r['score'] >= MIN_SCORE_THRESHOLD]
	filtered = before_count - len(results)
	else:
	filtered = 0

	if results:
	if not quiet:
	print(f"🔍 搜索 '{query}' → 关键词 {keywords[:8]}{'...' if len(keywords)>8 else ''} → 命中 {len(results)} 条" + (f"（过滤 {filtered} 条低分）" if filtered else ""))
	for r in results[:3]:
	print(f" 📌 [score={r['score']:.3f}] (hits={r['hit_count']}, imp={r['importance']}) {r['content'][:60]}...")

	# quiet 探针不是真召回、不 bump last_accessed
	if not quiet:
	ids = [r["id"] for r in results]
	await conn.execute(
	"UPDATE memories SET last_accessed = NOW(), recall_count = COALESCE(recall_count, 0) + 1 WHERE id = ANY($1::int[])",
	ids,
	)
	elif not quiet:
	print(f"🔍 搜索 '{query}' → 关键词 {keywords[:8]} → 无结果" + (f"（{filtered} 条被分数阈值过滤）" if filtered else ""))

	return results


	async def search_memories_hybrid(
	query: Union[str, List[str]],
	limit: int = 10,
	importance_min: Optional[int] = None,
	importance_max: Optional[int] = None,
	quiet: bool = False,
	exclude_dormant: bool = False,
	):
	"""
	记忆混合搜索：关键词 + 向量，归一化后四维加权

	权重：MEMORY_HW_KEYWORD + MEMORY_HW_SEMANTIC + MEMORY_HW_IMPORTANCE + MEMORY_HW_RECENCY

	2026-06-08 E 方案：importance_min/max 加到关键词路 + 向量路的 SQL WHERE，
	让候选池在 SQL 层就按 user_message 重量过滤、不做 client-side reorder
	（详见 [[zhiyu-memory-architecture]] E 方案）。

	2026-06-22 B 方案：query 接 Union[str, List[str]]——
	- list：dsv4 search_queries 路径、逐个独立 jieba + 短 query 整词候选（防"和府捞面"被切碎）
	- str：fallback / MCP / 知渝主动搜路径、行为完全照旧
	向量路（embedding）list 时拼一段（embedding 对短语描述合理）、str 时直接传。
	"""
	# E-2: 构造可复用的 importance filter SQL 片段
	_imp_filter = ""
	if importance_min is not None:
	_imp_filter += f" AND importance >= {int(importance_min)}"
	if importance_max is not None:
	_imp_filter += f" AND importance <= {int(importance_max)}"
	from datetime import datetime, timezone

	# B 方案：list 走逐个独立分词、str 走原路径
	if isinstance(query, list):
	keywords = extract_keywords_from_queries(query)
	query_for_embedding = "\n".join(q for q in query if q and q.strip())
	else:
	keywords = extract_search_keywords(query)
	query_for_embedding = query
	query_embedding = await compute_embedding(query_for_embedding) if EMBEDDING_API_KEY else []

	if not keywords and not query_embedding:
	return []

	pool = await get_pool()
	async with pool.acquire() as conn:
	candidates = {} # id -> {content, importance, created_at, kw_score, similarity}

	# ---- 关键词路 ----
	if keywords:
	case_parts = []
	params = []
	for i, kw in enumerate(keywords):
	case_parts.append(
	f"CASE WHEN content ILIKE '%' \|\| ${i+1} \|\| '%' "
	f"OR EXISTS (SELECT 1 FROM unnest(tags) t WHERE t ILIKE '%' \|\| ${i+1} \|\| '%') "
	f"THEN 1 ELSE 0 END"
	)
	params.append(kw)

	hit_count_expr = " + ".join(case_parts)
	max_hits = len(keywords)
	where_parts = [
	f"(content ILIKE '%' \|\| ${i+1} \|\| '%' "
	f"OR EXISTS (SELECT 1 FROM unnest(tags) t WHERE t ILIKE '%' \|\| ${i+1} \|\| '%'))"
	for i in range(len(keywords))
	]
	where_clause = f"is_active = TRUE AND ({' OR '.join(where_parts)}){_imp_filter}"

	limit_idx = len(keywords) + 1
	params.append(limit * 3)

	kw_sql = f"""
	SELECT id, content, importance, created_at,
	last_accessed, recall_count, layer,
	({hit_count_expr}) AS hit_count,
	({hit_count_expr})::float / {max_hits}.0 AS kw_score
	FROM memories
	WHERE {where_clause}
	ORDER BY kw_score DESC
	LIMIT ${limit_idx}
	"""
	kw_rows = await conn.fetch(kw_sql, *params)

	for r in kw_rows:
	candidates[r['id']] = {
	'content': r['content'],
	'importance': r['importance'],
	'created_at': r['created_at'],
	'last_accessed': r['last_accessed'],
	'recall_count': r['recall_count'] or 0,
	'layer': r['layer'],
	'hit_count': r['hit_count'],
	'kw_score': float(r['kw_score']),
	'similarity': 0.0,
	}

	# ---- 向量路 ----
	if query_embedding:
	if HAS_PGVECTOR:
	vec_str = '[' + ','.join(str(f) for f in query_embedding) + ']'
	sem_rows = await conn.fetch(f"""
	SELECT id, content, importance, created_at,
	last_accessed, recall_count, layer,
	1 - (embedding <=> $1::vector) as similarity
	FROM memories
	WHERE embedding IS NOT NULL AND is_active = TRUE{_imp_filter}
	ORDER BY embedding <=> $1::vector
	LIMIT $2
	""", vec_str, limit * 3)
	else:
	# Python端计算cosine（json 用模块顶层的——这里若写局部 import
	# 会把 json 变成本函数的局部名、pgvector 分支就 UnboundLocalError）
	all_mem = await conn.fetch(f"""
	SELECT id, content, importance, created_at,
	last_accessed, recall_count, layer, embedding_json
	FROM memories WHERE embedding_json IS NOT NULL AND is_active = TRUE{_imp_filter}
	""")

	scored = []
	for row in all_mem:
	try:
	emb = json.loads(row['embedding_json'])
	sim = _cosine_sim(query_embedding, emb)
	scored.append({**dict(row), 'similarity': sim})
	except Exception:
	continue
	scored.sort(key=lambda x: -x['similarity'])
	sem_rows = scored[:limit * 3]

	for r in sem_rows:
	sim = float(r['similarity'])
	if sim < MEMORY_SEMANTIC_THRESHOLD:
	continue
	mid = r['id']
	if mid in candidates:
	candidates[mid]['similarity'] = sim
	else:
	candidates[mid] = {
	'content': r['content'],
	'importance': r['importance'],
	'created_at': r['created_at'],
	'last_accessed': r['last_accessed'],
	'recall_count': r['recall_count'] or 0,
	'layer': r['layer'],
	'hit_count': 0,
	'kw_score': 0.0,
	'similarity': sim,
	}

	# debug：向量路统计
	sem_total = len(sem_rows)
	sem_passed = sum(1 for r in sem_rows if float(r['similarity']) >= MEMORY_SEMANTIC_THRESHOLD)
	sem_max = max((float(r['similarity']) for r in sem_rows), default=0)
	if not quiet:
	if sem_total > 0 and sem_passed == 0:
	print(f" 🔢 向量路: {sem_total}条候选全被阈值过滤（最高sim={sem_max:.3f}, 阈值={MEMORY_SEMANTIC_THRESHOLD}）")
	elif sem_total > 0:
	print(f" 🔢 向量路: {sem_passed}/{sem_total}条通过阈值（最高sim={sem_max:.3f}）")

	# ---- 拂卷 union · 2026-07-02 ----
	# 读书笔记也按向量搜进 recall——"聊到相关话题、书里划过的相关句子自然浮现"
	try:
	mark_rows = await conn.fetch(
	"""
	SELECT rm.id, rm.book_id, rm.chapter_id, rm.who, rm.kind,
	rm.text_snippet, rm.note_content, rm.embedding_json, rm.created_at,
	b.title AS book_title, c.title AS chapter_title
	FROM reading_marks rm
	JOIN books b ON b.id = rm.book_id
	JOIN chapters c ON c.id = rm.chapter_id
	WHERE rm.embedding_json IS NOT NULL
	"""
	)
	MARK_IMPORTANCE = 6 # 书里划的痕迹跟事件记忆同档、比碎片高一点
	for row in mark_rows:
	try:
	emb = json.loads(row['embedding_json'])
	sim = _cosine_sim(query_embedding, emb)
	if sim < MEMORY_SEMANTIC_THRESHOLD:
	continue
	# 组装 memory-like content：出处 + 批注/摘录
	parts = [f"【{row['book_title']} · {row['chapter_title']}】"]
	if row['note_content']:
	parts.append(row['note_content'])
	if row['text_snippet']:
	parts.append(f"引：{row['text_snippet']}")
	content = "\n".join(parts)
	mark_key = f"mark_{row['id']}"
	candidates[mark_key] = {
	'content': content,
	'importance': MARK_IMPORTANCE,
	'created_at': row['created_at'],
	# 划线没有召回记录、体温按创建时间以事件档冷却；v1 不入眠
	#（reading_marks 无 recall_count 列、书的痕迹去留归拂卷管）
	'last_accessed': row['created_at'],
	'recall_count': 0,
	'layer': 2,
	'is_mark': True,
	'hit_count': 0,
	'kw_score': 0.0,
	'similarity': sim,
	}
	except Exception:
	continue
	if not quiet and mark_rows:
	mark_hit = sum(1 for k in candidates if isinstance(k, str) and k.startswith("mark_"))
	if mark_hit:
	print(f" 📖 拂卷笔记：{mark_hit} 条被 recall")
	except Exception as e:
	print(f"⚠️ 搜 reading_marks 失败（跳过、不影响主流程）: {e!r}", flush=True)

	# 日志用 query 显示串（list 拼成 " \| " 分隔、避免打成 Python list 字面量）
	_q_disp = query if isinstance(query, str) else " \| ".join(query)

	if not candidates:
	if not quiet:
	print(f"🔍 混合搜索 '{_q_disp}' → 两路均无结果")
	return []

	# ---- 归一化 + 加权 ----
	kw_norm = _min_max_normalize({mid: v['kw_score'] for mid, v in candidates.items()})
	sem_norm = _min_max_normalize({mid: v['similarity'] for mid, v in candidates.items()})

	now = datetime.now(timezone.utc)
	final = []
	dormant_skipped = 0
	for mid, info in candidates.items():
	kw = kw_norm.get(mid, 0.0)
	sem = sem_norm.get(mid, 0.0)
	imp = info['importance'] / 10.0
	# 星河呼吸 v1：旧 recency（1/(1+出生天数)、召回不刷新）换成 activation 体温
	# ——最近被想起的暖、久无人问的冷；核心记忆恒温（compute_activation 内处理）
	act = compute_activation(
	info.get('last_accessed'), info.get('layer'),
	info.get('recall_count', 0), now=now,
	)
	if info.get('is_mark'):
	dormant, asleep_days = False, 0 # 划线不入眠（v1）
	else:
	dormant, asleep_days = compute_sleep_state(
	info.get('last_accessed'), info.get('layer'),
	info.get('recall_count', 0), now=now,
	)
	if exclude_dormant and dormant:
	dormant_skipped += 1
	continue

	score = (MEMORY_HW_KEYWORD * kw +
	MEMORY_HW_SEMANTIC * sem +
	MEMORY_HW_IMPORTANCE * imp +
	MEMORY_HW_RECENCY * act)

	item = {
	'id': mid,
	'content': info['content'],
	'importance': info['importance'],
	'created_at': info['created_at'],
	'hit_count': info['hit_count'],
	'similarity': info['similarity'],
	'score': score,
	'activation': round(act, 3),
	}
	# 入眠字段只在睡着时带（结果瘦身；知渝主动搜到睡着的会看到"已睡 X 天"）
	if dormant:
	item['dormant'] = True
	item['asleep_days'] = asleep_days
	final.append(item)

	final.sort(key=lambda x: (-x['score'], -x['importance']))

	# 过滤低分
	if MIN_SCORE_THRESHOLD > 0:
	before_count = len(final)
	final = [r for r in final if r['score'] >= MIN_SCORE_THRESHOLD]
	filtered = before_count - len(final)
	else:
	filtered = 0

	results = final[:limit]

	if results:
	if not quiet:
	mode_tag = "混合" if query_embedding else "关键词"
	kw_tag = f"关键词 {keywords[:6]}" if keywords else "无关键词"
	sleep_tag = f"（{dormant_skipped} 条入眠中、未扰）" if dormant_skipped else ""
	print(f"🔍 {mode_tag}搜索 '{_q_disp}' → {kw_tag} → 命中 {len(results)} 条" + (f"（过滤 {filtered} 条低分）" if filtered else "") + sleep_tag)
	for r in results[:3]:
	print(f" 📌 [score={r['score']:.3f}] (kw={r['hit_count']}, sim={r['similarity']:.2f}, imp={r['importance']}, act={r['activation']}) {r['content'][:60]}...")

	# quiet 探针不是真召回、不 bump last_accessed
	if not quiet:
	# 拂卷 union 的 id 是 "mark_N" 字符串、不在 memories 表，混进 int[] 会让 asyncpg 编码炸
	ids = [r["id"] for r in results if isinstance(r["id"], int)]
	if ids:
	# 真召回 = 回暖：刷新心跳 + 次数（星河呼吸 v1；睡着的被搜到即苏醒）
	await conn.execute(
	"UPDATE memories SET last_accessed = NOW(), recall_count = COALESCE(recall_count, 0) + 1 WHERE id = ANY($1::int[])",
	ids,
	)
	elif not quiet:
	print(f"🔍 混合搜索 '{_q_disp}' → 无结果" + (f"（{filtered} 条被过滤）" if filtered else ""))

	return [dict(r) for r in results]


	async def get_pending_memory_embedding_count():
	"""查询还没有embedding的记忆数量"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	if HAS_PGVECTOR:
	return await conn.fetchval(
	"SELECT COUNT(*) FROM memories WHERE embedding IS NULL AND content IS NOT NULL"
	)
	else:
	return await conn.fetchval(
	"SELECT COUNT(*) FROM memories WHERE embedding_json IS NULL AND content IS NOT NULL"
	)


	async def backfill_memory_embeddings(batch_size: int = 20):
	"""给已有记忆补算embedding（没有embedding的记忆）"""
	if not EMBEDDING_API_KEY:
	print("⚠️ EMBEDDING_API_KEY 未设置，无法补算embedding")
	return 0

	pool = await get_pool()
	total_updated = 0

	async with pool.acquire() as conn:
	if HAS_PGVECTOR:
	rows = await conn.fetch("""
	SELECT id, content FROM memories
	WHERE embedding IS NULL AND content IS NOT NULL
	ORDER BY id
	LIMIT $1
	""", batch_size)
	else:
	rows = await conn.fetch("""
	SELECT id, content FROM memories
	WHERE embedding_json IS NULL AND content IS NOT NULL
	ORDER BY id
	LIMIT $1
	""", batch_size)

	if not rows:
	print("✅ 所有记忆已有embedding，无需补算")
	return 0

	print(f"🔄 开始补算记忆embedding... 本批 {len(rows)} 条")

	async with pool.acquire() as conn:
	for row in rows:
	try:
	embedding = await compute_embedding(row['content'] or '')
	if embedding:
	await save_memory_embedding(conn, row['id'], embedding)
	total_updated += 1
	except Exception as e:
	print(f"⚠️ 记忆 {row['id']} embedding计算失败: {e}")

	# 检查剩余
	async with pool.acquire() as conn:
	if HAS_PGVECTOR:
	remaining = await conn.fetchval("SELECT COUNT(*) FROM memories WHERE embedding IS NULL AND content IS NOT NULL")
	else:
	remaining = await conn.fetchval("SELECT COUNT(*) FROM memories WHERE embedding_json IS NULL AND content IS NOT NULL")

	print(f"✅ 本批补算完成：{total_updated}/{len(rows)} 条成功" + (f"，剩余 {remaining} 条待处理" if remaining > 0 else ""))
	return total_updated


	async def get_recent_memories(limit: int = 20):
	pool = await get_pool()
	async with pool.acquire() as conn:
	return await conn.fetch(
	"SELECT id, content, importance, created_at FROM memories ORDER BY created_at DESC LIMIT $1",
	limit,
	)


	async def get_all_memories_count():
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow("SELECT COUNT(*) as cnt FROM memories")
	return row["cnt"]


	async def get_all_memories():
	"""导出所有记忆（用于备份）"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	rows = await conn.fetch(
	"SELECT content, importance, source_session, created_at FROM memories ORDER BY id"
	)
	return [dict(r) for r in rows]


	async def get_all_memories_detail(limit: int = None, layer: int = None, active_only: bool = None,
	offset: int = None, order: str = "id_asc"):
	"""获取所有记忆（含 id，用于管理页面）

	Args:
	limit: 可选，限制返回数量
	layer: 可选，筛选指定层级（1=原始碎片, 2=事件记忆, 3=核心记忆）
	active_only: 可选，是否只返回 is_active=true 的记忆
	"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	conditions = []
	params = []
	param_idx = 1

	if layer is not None:
	conditions.append(f"layer = ${param_idx}")
	params.append(layer)
	param_idx += 1

	if active_only is not None:
	conditions.append(f"is_active = ${param_idx}")
	params.append(active_only)
	param_idx += 1

	where_clause = f"WHERE {' AND '.join(conditions)}" if conditions else ""

	# 排序：默认 id 升序（兼容旧调用）；星河列表用 created_desc（最新在上）
	order_map = {
	"id_asc": "ORDER BY id ASC",
	"id_desc": "ORDER BY id DESC",
	"created_desc": "ORDER BY created_at DESC, id DESC",
	"created_asc": "ORDER BY created_at ASC, id ASC",
	"importance_desc": "ORDER BY importance DESC, created_at DESC",
	}
	order_clause = order_map.get(order, "ORDER BY id ASC")

	if limit is not None:
	limit_clause = f"LIMIT ${param_idx}"
	params.append(limit)
	param_idx += 1
	else:
	limit_clause = ""

	if offset is not None:
	offset_clause = f"OFFSET ${param_idx}"
	params.append(offset)
	param_idx += 1
	else:
	offset_clause = ""

	rows = await conn.fetch(f"""
	SELECT id, content, importance, source_session, created_at,
	layer, title, is_active, merged_from, event_date, tags,
	last_accessed, recall_count
	FROM memories
	{where_clause}
	{order_clause}
	{limit_clause}
	{offset_clause}
	""", *params)
	# 星河呼吸 v1：列表带体温和睡眠状态（星河 tab 给睡着的记忆画小月亮）
	now = datetime.now(dt_timezone.utc)
	out = []
	for r in rows:
	d = dict(r)
	la, layer_v, rc = d.pop('last_accessed', None), d.get('layer'), d.pop('recall_count', 0) or 0
	d['activation'] = round(compute_activation(la, layer_v, rc, now=now), 3)
	dormant, asleep_days = compute_sleep_state(la, layer_v, rc, now=now)
	if dormant:
	d['dormant'] = True
	d['asleep_days'] = asleep_days
	out.append(d)
	return out


	async def update_memory(memory_id: int, content: str = None, importance: int = None):
	"""更新单条记忆"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	if content is not None and importance is not None:
	await conn.execute(
	"UPDATE memories SET content = $1, importance = $2 WHERE id = $3",
	content, importance, memory_id
	)
	elif content is not None:
	await conn.execute(
	"UPDATE memories SET content = $1 WHERE id = $2",
	content, memory_id
	)
	elif importance is not None:
	await conn.execute(
	"UPDATE memories SET importance = $1 WHERE id = $2",
	importance, memory_id
	)


	async def _resync_memory_id_seq(conn):
	"""删除后重置自增序列到 MAX(id)+1。
	删掉末尾几条时，下一条新记忆的 id 会接着当前最大值继续，
	不会一直往上飘留下空号。中间的空缺号不会被填补。"""
	await conn.execute("""
	SELECT setval(
	pg_get_serial_sequence('memories', 'id'),
	COALESCE((SELECT MAX(id) FROM memories), 0) + 1,
	false
	)
	""")


	async def delete_memory(memory_id: int):
	"""删除单条记忆"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	await conn.execute("DELETE FROM memories WHERE id = $1", memory_id)
	await _resync_memory_id_seq(conn)


	async def delete_memories_batch(memory_ids: list):
	"""批量删除记忆"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	await conn.execute(
	"DELETE FROM memories WHERE id = ANY($1::int[])", memory_ids
	)
	await _resync_memory_id_seq(conn)


	# ============================================================
	# 网关配置
	# ============================================================

	async def get_gateway_config(key: str, default: str = "") -> str:
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow("SELECT value FROM gateway_config WHERE key = $1", key)
	return row['value'] if row else default


	async def set_gateway_config(key: str, value: str):
	pool = await get_pool()
	async with pool.acquire() as conn:
	await conn.execute("""
	INSERT INTO gateway_config (key, value) VALUES ($1, $2)
	ON CONFLICT (key) DO UPDATE SET value = $2
	""", key, value)


	async def get_all_gateway_config() -> dict:
	"""获取所有配置项"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	rows = await conn.fetch("SELECT key, value FROM gateway_config")
	return {r['key']: r['value'] for r in rows}


	# ============================================================
	# 对话历史读取（分区缓存用）
	# ============================================================

	async def get_conversation_messages(session_id: str, limit: int = 100):
	"""按时间正序读取session的消息"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	rows = await conn.fetch("""
	SELECT role, content, metadata, created_at
	FROM conversations
	WHERE session_id = $1
	ORDER BY created_at ASC
	LIMIT $2
	""", session_id, limit)
	return [dict(r) for r in rows]


	# ============================================================
	# 分区缓存状态管理
	# ============================================================

	async def get_session_cache_state(session_id: str) -> dict:
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow(
	"SELECT summary, a_start_round, updated_at FROM session_cache_state WHERE session_id = $1",
	session_id
	)
	if row:
	raw_summary = row['summary'] or ''
	summary_parts = []
	if raw_summary:
	try:
	import json
	parsed = json.loads(raw_summary)
	if isinstance(parsed, list):
	summary_parts = parsed
	else:
	summary_parts = [raw_summary]
	except (json.JSONDecodeError, ValueError):
	summary_parts = [raw_summary]
	return {
	'summary_parts': summary_parts,
	'a_start_round': row['a_start_round'] or 0,
	'updated_at': row['updated_at'],
	}
	return {'summary_parts': [], 'a_start_round': 0, 'updated_at': None}


	async def save_session_cache_state(session_id: str, summary_parts: list, a_start_round: int):
	import json
	summary_json = json.dumps(summary_parts, ensure_ascii=False)
	pool = await get_pool()
	async with pool.acquire() as conn:
	await conn.execute("""
	INSERT INTO session_cache_state (session_id, summary, a_start_round, updated_at)
	VALUES ($1, $2, $3, NOW())
	ON CONFLICT (session_id)
	DO UPDATE SET summary = $2, a_start_round = $3, updated_at = NOW()
	""", session_id, summary_json, a_start_round)


	# ============================================================
	# Token 使用记录
	# ============================================================

	async def ensure_token_usage_table():
	"""确保token_usage表存在（在init_tables里调用）"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	await conn.execute("""
	CREATE TABLE IF NOT EXISTS token_usage (
	id SERIAL PRIMARY KEY,
	session_id TEXT,
	model TEXT,
	prompt_tokens INTEGER DEFAULT 0,
	completion_tokens INTEGER DEFAULT 0,
	total_tokens INTEGER DEFAULT 0,
	usage_type TEXT DEFAULT 'chat',
	created_at TIMESTAMPTZ DEFAULT NOW()
	);
	""")
	await conn.execute("""
	CREATE INDEX IF NOT EXISTS idx_token_usage_created ON token_usage (created_at DESC);
	""")
	# 2026-06-15：加缓存细分列——查"上下文烧钱"必须看缓存命中/写入。
	# prompt_tokens 存的是"未命中缓存的新输入"(claude input_tokens / OpenAI prompt_tokens)，
	# cache_read=命中缓存读(0.1x 便宜)、cache_creation=写缓存(5m 1.25x / 1h 2x)。
	# 5m/1h 细分单独存——用来实测 Claude Code 到底用哪种 TTL。
	for col in ("cache_read_tokens", "cache_creation_tokens",
	"cache_creation_5m_tokens", "cache_creation_1h_tokens"):
	await conn.execute(
	f"ALTER TABLE token_usage ADD COLUMN IF NOT EXISTS {col} INTEGER DEFAULT 0;"
	)


	async def save_token_usage(
	session_id: str,
	model: str,
	prompt_tokens: int,
	completion_tokens: int,
	total_tokens: int,
	usage_type: str = "chat",
	cache_read_tokens: int = 0,
	cache_creation_tokens: int = 0,
	cache_creation_5m_tokens: int = 0,
	cache_creation_1h_tokens: int = 0,
	):
	pool = await get_pool()
	async with pool.acquire() as conn:
	await conn.execute("""
	INSERT INTO token_usage (
	session_id, model, prompt_tokens, completion_tokens, total_tokens, usage_type,
	cache_read_tokens, cache_creation_tokens,
	cache_creation_5m_tokens, cache_creation_1h_tokens
	)
	VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
	""", session_id, model, prompt_tokens, completion_tokens, total_tokens, usage_type,
	cache_read_tokens, cache_creation_tokens,
	cache_creation_5m_tokens, cache_creation_1h_tokens)


	async def get_recent_token_usage(limit: int = 50) -> list:
	"""最近 N 条 usage 明细（usage 面板用）。"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	rows = await conn.fetch("""
	SELECT id, session_id, model, prompt_tokens, completion_tokens, total_tokens,
	usage_type, cache_read_tokens, cache_creation_tokens,
	cache_creation_5m_tokens, cache_creation_1h_tokens, created_at
	FROM token_usage
	ORDER BY created_at DESC
	LIMIT $1
	""", limit)
	return [dict(r) for r in rows]


	async def get_usage_window_summary(hours: int = 5) -> dict:
	"""滚动窗口聚合（默认 5h，对应订阅限额窗口）——看这段烧了多少、缓存命中率。"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow(f"""
	SELECT
	COUNT(*) AS turns,
	COALESCE(SUM(prompt_tokens), 0) AS input_uncached,
	COALESCE(SUM(cache_read_tokens), 0) AS cache_read,
	COALESCE(SUM(cache_creation_tokens), 0) AS cache_creation,
	COALESCE(SUM(cache_creation_5m_tokens), 0) AS cache_5m,
	COALESCE(SUM(cache_creation_1h_tokens), 0) AS cache_1h,
	COALESCE(SUM(completion_tokens), 0) AS output,
	COALESCE(SUM(total_tokens), 0) AS total
	FROM token_usage
	WHERE created_at >= NOW() - ($1 \|\| ' hours')::interval
	""", str(hours))
	d = dict(row) if row else {}
	# 缓存命中率 = 读 / (读 + 写 + 未命中新输入)
	cr = d.get("cache_read", 0) or 0
	cc = d.get("cache_creation", 0) or 0
	iu = d.get("input_uncached", 0) or 0
	denom = cr + cc + iu
	d["cache_hit_ratio"] = round(cr / denom, 3) if denom else None
	d["window_hours"] = hours
	return d


	# ============================================================
	# 对话记录管理
	# ============================================================

	async def ensure_conversation_titles_table():
	pool = await get_pool()
	async with pool.acquire() as conn:
	await conn.execute("""
	CREATE TABLE IF NOT EXISTS conversation_titles (
	session_id TEXT PRIMARY KEY,
	title TEXT DEFAULT ''
	);
	""")


	async def get_conversations_paginated(page: int = 1, per_page: int = 20):
	offset = (page - 1) * per_page
	pool = await get_pool()
	async with pool.acquire() as conn:
	total_row = await conn.fetchrow(
	"SELECT COUNT(DISTINCT session_id) as total FROM conversations"
	)
	total = total_row['total'] if total_row else 0

	rows = await conn.fetch("""
	WITH session_info AS (
	SELECT session_id, MIN(created_at) as first_time, MAX(created_at) as last_time, COUNT(*) as message_count
	FROM conversations GROUP BY session_id ORDER BY last_time DESC LIMIT $1 OFFSET $2
	)
	SELECT si.*, ct.title as custom_title,
	COALESCE(tu.total_all, 0) as total_tokens
	FROM session_info si
	LEFT JOIN conversation_titles ct ON si.session_id = ct.session_id
	LEFT JOIN (
	SELECT session_id, SUM(total_tokens) as total_all FROM token_usage WHERE usage_type = 'chat' GROUP BY session_id
	) tu ON si.session_id = tu.session_id
	ORDER BY si.last_time DESC
	""", per_page, offset)

	results = []
	for r in rows:
	preview_row = await conn.fetchrow(
	"SELECT content FROM conversations WHERE session_id = $1 AND role = 'user' ORDER BY created_at LIMIT 1",
	r['session_id']
	)
	preview = preview_row['content'][:80] if preview_row else ''
	title = r['custom_title'] or (preview[:30] + '...' if len(preview) > 30 else preview) or r['session_id']
	results.append({
	'session_id': r['session_id'],
	'title': title,
	'first_time': r['first_time'].isoformat() if r['first_time'] else None,
	'last_time': r['last_time'].isoformat() if r['last_time'] else None,
	'message_count': r['message_count'],
	'preview': preview,
	'total_tokens': r['total_tokens'],
	})
	return results, total


	async def delete_conversation(session_id: str):
	pool = await get_pool()
	async with pool.acquire() as conn:
	await conn.execute("DELETE FROM conversations WHERE session_id = $1", session_id)
	await conn.execute("DELETE FROM conversation_titles WHERE session_id = $1", session_id)
	await conn.execute("DELETE FROM session_cache_state WHERE session_id = $1", session_id)


	async def batch_delete_conversations(session_ids: list):
	pool = await get_pool()
	async with pool.acquire() as conn:
	await conn.execute("DELETE FROM conversations WHERE session_id = ANY($1)", session_ids)
	await conn.execute("DELETE FROM conversation_titles WHERE session_id = ANY($1)", session_ids)
	await conn.execute("DELETE FROM session_cache_state WHERE session_id = ANY($1)", session_ids)


	async def delete_messages_by_ids(ids: list) -> int:
	"""按 conversations.id 精确删除指定消息行（不按 session_id）。

	用途：清理 bug 态产生的个别对话——例如 stale sid 误 resume 期间知渝在错误
	session 上的错乱回应。按 session 删会误伤同 session 名下的真实对话（f8259dd3
	名下就混着 6/9 至今 864 条真实经历），所以必须按 id 精确删。
	返回实际删除行数。"""
	if not ids:
	return 0
	pool = await get_pool()
	async with pool.acquire() as conn:
	result = await conn.execute("DELETE FROM conversations WHERE id = ANY($1::int[])", ids)
	# asyncpg execute 对 DELETE 返回形如 "DELETE 8"
	try:
	return int(result.split()[-1])
	except Exception:
	return 0


	async def delete_activities_by_ids(ids: list) -> int:
	"""按 activities.id 精确删除西窗活动行。

	用途：清理 bug 态产生的空白/噪声活动条目——例如 2026-07-05 之前纯工具、
	没正文的闹钟 wakeup turn 落下的 content="" 空壳西窗条目。返回实际删除行数。"""
	if not ids:
	return 0
	pool = await get_pool()
	async with pool.acquire() as conn:
	result = await conn.execute("DELETE FROM activities WHERE id = ANY($1::int[])", ids)
	try:
	return int(result.split()[-1])
	except Exception:
	return 0


	async def merge_sessions_to_target(source_ids: list, target_id: str) -> dict:
	if not source_ids:
	return {'merged_sessions': 0, 'merged_messages': 0, 'merged_token_records': 0}
	pool = await get_pool()
	async with pool.acquire() as conn:
	msg_count = await conn.fetchval("SELECT COUNT(*) FROM conversations WHERE session_id = ANY($1)", source_ids)
	await conn.execute("UPDATE conversations SET session_id = $1 WHERE session_id = ANY($2)", target_id, source_ids)
	token_count = await conn.fetchval("SELECT COUNT(*) FROM token_usage WHERE session_id = ANY($1)", source_ids)
	await conn.execute("UPDATE token_usage SET session_id = $1 WHERE session_id = ANY($2)", target_id, source_ids)
	await conn.execute("DELETE FROM conversation_titles WHERE session_id = ANY($1)", source_ids)
	await conn.execute("DELETE FROM session_cache_state WHERE session_id = ANY($1)", source_ids)
	return {'merged_sessions': len(source_ids), 'merged_messages': msg_count or 0, 'merged_token_records': token_count or 0}


	async def list_all_session_cache_states() -> list:
	pool = await get_pool()
	async with pool.acquire() as conn:
	rows = await conn.fetch("""
	SELECT scs.session_id, scs.summary, scs.a_start_round, scs.updated_at,
	COALESCE(c.message_count, 0) as message_count,
	COALESCE(tu.chat_tokens, 0) as chat_tokens
	FROM session_cache_state scs
	LEFT JOIN (SELECT session_id, COUNT(*) as message_count FROM conversations GROUP BY session_id) c ON scs.session_id = c.session_id
	LEFT JOIN (SELECT session_id, SUM(total_tokens) as chat_tokens FROM token_usage WHERE usage_type = 'chat' GROUP BY session_id) tu ON scs.session_id = tu.session_id
	ORDER BY scs.updated_at DESC
	""")
	results = []
	for r in rows:
	raw_summary = r['summary'] or ''
	try:
	import json
	parsed = json.loads(raw_summary)
	if isinstance(parsed, list):
	summary_parts = parsed
	else:
	summary_parts = [raw_summary] if raw_summary else []
	except (json.JSONDecodeError, ValueError):
	summary_parts = [raw_summary] if raw_summary else []
	results.append({
	'session_id': r['session_id'],
	'summary': '\n\n'.join(summary_parts),
	'summary_length': sum(len(p) for p in summary_parts),
	'summary_count': len(summary_parts),
	'a_start_round': r['a_start_round'],
	'updated_at': r['updated_at'].isoformat() if r['updated_at'] else None,
	'message_count': r['message_count'],
	'chat_tokens': r['chat_tokens'],
	})
	return results


	async def delete_session_cache_state(session_id: str):
	pool = await get_pool()
	async with pool.acquire() as conn:
	await conn.execute("DELETE FROM session_cache_state WHERE session_id = $1", session_id)


	async def rename_session_id(old_id: str, new_id: str) -> bool:
	"""重命名对话线ID（事务内同时修改三个表）"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	async with conn.transaction():
	# 检查新ID是否已存在
	exists = await conn.fetchval(
	"SELECT 1 FROM session_cache_state WHERE session_id = $1", new_id
	)
	if exists:
	return False
	# session_cache_state
	await conn.execute(
	"UPDATE session_cache_state SET session_id = $1 WHERE session_id = $2",
	new_id, old_id
	)
	# conversations
	await conn.execute(
	"UPDATE conversations SET session_id = $1 WHERE session_id = $2",
	new_id, old_id
	)
	# token_usage
	await conn.execute(
	"UPDATE token_usage SET session_id = $1 WHERE session_id = $2",
	new_id, old_id
	)
	return True


	def db_row_to_message(row: dict) -> dict:
	"""
	把DB记录还原成API消息格式。

	普通消息: {"role": "user", "content": "你好"}
	工具调用: {"role": "assistant", "content": null, "tool_calls": [...]}
	工具结果: {"role": "tool", "content": "结果", "tool_call_id": "call_xxx"}
	思维链: {"role": "assistant", "content": "回答", "reasoning_content": "思维链"}
	"""
	import json as _json
	msg = {"role": row["role"], "content": row.get("content") or ""}

	meta_str = row.get("metadata")
	if meta_str:
	try:
	meta = _json.loads(meta_str)
	# assistant 带 tool_calls
	if "tool_calls" in meta:
	msg["tool_calls"] = meta["tool_calls"]
	if not row.get("content"):
	msg["content"] = None
	# assistant 带 reasoning_content（deepseek thinking mode）
	if "reasoning_content" in meta:
	msg["reasoning_content"] = meta["reasoning_content"]
	# tool 消息带 tool_call_id
	if "tool_call_id" in meta:
	msg["tool_call_id"] = meta["tool_call_id"]
	# 其他可能的字段（name 等）
	if "name" in meta:
	msg["name"] = meta["name"]
	except Exception:
	pass

	return msg


	async def export_all_conversations():
	"""导出所有对话记录（用于备份）"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	rows = await conn.fetch("""
	SELECT session_id, role, content, model, created_at
	FROM conversations
	ORDER BY session_id, created_at
	""")
	return [
	{
	'session_id': r['session_id'],
	'role': r['role'],
	'content': r['content'],
	'model': r['model'] or '',
	'created_at': r['created_at'].isoformat() if r['created_at'] else None,
	}
	for r in rows
	]


	async def import_conversations(records: list):
	"""
	导入对话记录（自动去重）

	records: [{ session_id, role, content, model?, created_at? }, ...]
	按 session_id + role + created_at 三元组去重，已存在的跳过。
	返回 (导入数量, 跳过数量)
	"""
	if not records:
	return 0, 0

	pool = await get_pool()
	async with pool.acquire() as conn:
	imported = 0
	skipped = 0
	for r in records:
	session_id = r.get('session_id')
	role = r.get('role')
	content = r.get('content')

	if not all([session_id, role, content]):
	continue

	model = r.get('model', '')
	created_at = r.get('created_at')

	# 解析时间
	from datetime import datetime
	if created_at and isinstance(created_at, str):
	try:
	created_at = datetime.fromisoformat(created_at.replace('Z', '+00:00'))
	except:
	created_at = None

	# 去重检查
	if created_at:
	existing = await conn.fetchrow("""
	SELECT id FROM conversations
	WHERE session_id = $1 AND role = $2 AND created_at = $3
	LIMIT 1
	""", session_id, role, created_at)

	if existing:
	skipped += 1
	continue

	# 2026-06-10 显式带 thread_id='main'——以前靠 column DEFAULT，
	# 万一 schema 改了 DEFAULT 不是 main 就埋坑
	await conn.execute("""
	INSERT INTO conversations (session_id, role, content, model, created_at, thread_id)
	VALUES ($1, $2, $3, $4, $5, 'main')
	""", session_id, role, content, model, created_at)
	else:
	await conn.execute("""
	INSERT INTO conversations (session_id, role, content, model, thread_id)
	VALUES ($1, $2, $3, $4, 'main')
	""", session_id, role, content, model)

	imported += 1

	if skipped:
	print(f"📥 导入对话: {imported} 条新增, {skipped} 条已存在跳过")
	else:
	print(f"📥 导入对话: {imported} 条新增")

	return imported, skipped


	# ============================================================
	# 三层记忆架构（碎片/事件/核心）
	# ============================================================

	async def get_fragments_by_date(event_date):
	"""获取指定日期的原始碎片（用于每日整理）"""
	# 把本地日期转成UTC时间范围，避免DATE()用UTC截断导致日期偏移
	local_tz = dt_timezone(timedelta(hours=TIMEZONE_HOURS))
	start_utc = datetime(event_date.year, event_date.month, event_date.day, tzinfo=local_tz).astimezone(dt_timezone.utc)
	end_utc = start_utc + timedelta(days=1)

	pool = await get_pool()
	async with pool.acquire() as conn:
	rows = await conn.fetch("""
	SELECT id, content, importance, created_at
	FROM memories
	WHERE layer = 1 AND is_active = TRUE
	AND created_at >= $1 AND created_at < $2
	ORDER BY created_at
	""", start_utc, end_utc)
	return [dict(r) for r in rows]


	async def get_fragments_by_date_range(start_date, end_date):
	"""获取指定时间段的原始碎片（用于跨天整理）"""
	# 把本地日期转成UTC时间范围，避免DATE()用UTC截断导致日期偏移
	local_tz = dt_timezone(timedelta(hours=TIMEZONE_HOURS))
	start_utc = datetime(start_date.year, start_date.month, start_date.day, tzinfo=local_tz).astimezone(dt_timezone.utc)
	# end_date 当天结束 = end_date 下一天的 00:00
	end_utc = datetime(end_date.year, end_date.month, end_date.day, tzinfo=local_tz).astimezone(dt_timezone.utc) + timedelta(days=1)

	pool = await get_pool()
	async with pool.acquire() as conn:
	rows = await conn.fetch("""
	SELECT id, content, importance, created_at
	FROM memories
	WHERE layer = 1 AND is_active = TRUE
	AND created_at >= $1 AND created_at < $2
	ORDER BY created_at
	""", start_utc, end_utc)
	return [dict(r) for r in rows]


	async def create_event_memory(title: str, content: str, importance: int,
	event_date, merged_from: list):
	"""创建事件记忆（从碎片合并而来）"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow("""
	INSERT INTO memories (content, importance, layer, title, is_active, merged_from, event_date)
	VALUES ($1, $2, 2, $3, TRUE, $4, $5)
	RETURNING id
	""", content, importance, title, merged_from, event_date)

	new_id = row['id'] if row else None

	# 向量搜索：计算并保存 embedding
	if MEMORY_VECTOR_ENABLED and new_id:
	try:
	embedding = await compute_embedding(content)
	if embedding:
	await save_memory_embedding(conn, new_id, embedding)
	except Exception as e:
	print(f"⚠️ 事件记忆embedding计算失败（id={new_id}）: {e}")

	return new_id


	async def deactivate_memories(memory_ids: list):
	"""将记忆标记为不活跃（合并后的碎片）"""
	if not memory_ids:
	return
	pool = await get_pool()
	async with pool.acquire() as conn:
	await conn.execute("""
	UPDATE memories SET is_active = FALSE
	WHERE id = ANY($1::int[])
	""", memory_ids)


	async def promote_to_core(memory_id: int, title: str = None):
	"""将记忆升级为核心记忆"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	if title:
	await conn.execute("""
	UPDATE memories SET layer = 3, title = $2
	WHERE id = $1
	""", memory_id, title)
	else:
	await conn.execute("""
	UPDATE memories SET layer = 3
	WHERE id = $1
	""", memory_id)


	async def merge_memories(memory_ids: list, new_title: str, new_content: str,
	importance: int, layer: int = 2):
	"""合并多条记忆为一条新记忆"""
	if not memory_ids:
	return None

	pool = await get_pool()
	async with pool.acquire() as conn:
	# 获取原记忆的日期（取最早的）
	rows = await conn.fetch("""
	SELECT MIN(DATE(created_at)) as event_date
	FROM memories WHERE id = ANY($1::int[])
	""", memory_ids)
	event_date = rows[0]['event_date'] if rows else None

	# 创建新记忆
	row = await conn.fetchrow("""
	INSERT INTO memories (content, importance, layer, title, is_active, merged_from, event_date)
	VALUES ($1, $2, $3, $4, TRUE, $5, $6)
	RETURNING id
	""", new_content, importance, layer, new_title, memory_ids, event_date)

	new_id = row['id'] if row else None

	# 向量搜索：计算并保存 embedding
	if MEMORY_VECTOR_ENABLED and new_id:
	try:
	embedding = await compute_embedding(new_content)
	if embedding:
	await save_memory_embedding(conn, new_id, embedding)
	except Exception as e:
	print(f"⚠️ 合并记忆embedding计算失败（id={new_id}）: {e}")

	# 将原记忆标记为不活跃
	if new_id:
	await deactivate_memories(memory_ids)

	return new_id


	async def check_duplicate_memory(new_content: str, threshold: float = 0.7) -> dict:
	"""检查新记忆是否与现有记忆重复

	三层去重策略：
	1. 精确匹配：内容完全相同
	2. 包含关系：新内容包含旧内容，或旧内容包含新内容
	3. 关键词重叠度：Jaccard 相似度 > threshold

	Returns:
	{
	"is_duplicate": bool,
	"reason": str, # "exact" / "containment" / "similarity"
	"matched_id": int or None,
	"similarity": float or None
	}
	"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	# 获取所有活跃记忆
	rows = await conn.fetch("""
	SELECT id, content FROM memories
	WHERE is_active = TRUE
	""")

	new_content_lower = new_content.strip().lower()
	new_keywords = set(extract_search_keywords(new_content))

	for row in rows:
	old_content = row['content']
	old_content_lower = old_content.strip().lower()

	# 第一层：精确匹配
	if new_content_lower == old_content_lower:
	return {
	"is_duplicate": True,
	"reason": "exact",
	"matched_id": row['id'],
	"similarity": 1.0
	}

	# 第二层：包含关系
	if new_content_lower in old_content_lower:
	return {
	"is_duplicate": True,
	"reason": "containment",
	"matched_id": row['id'],
	"similarity": len(new_content) / len(old_content)
	}
	if old_content_lower in new_content_lower:
	return {
	"is_duplicate": True,
	"reason": "containment_update",
	"matched_id": row['id'],
	"similarity": len(old_content) / len(new_content)
	}

	# 第三层：关键词重叠度（Jaccard 相似度）
	old_keywords = set(extract_search_keywords(old_content))
	if new_keywords and old_keywords:
	intersection = new_keywords & old_keywords
	union = new_keywords \| old_keywords
	similarity = len(intersection) / len(union) if union else 0

	if similarity > threshold:
	return {
	"is_duplicate": True,
	"reason": "similarity",
	"matched_id": row['id'],
	"similarity": similarity
	}

	return {
	"is_duplicate": False,
	"reason": None,
	"matched_id": None,
	"similarity": None
	}


	async def update_memory_with_layer(memory_id: int, content: str = None,
	importance: int = None, title: str = None,
	layer: int = None, is_active: bool = None,
	tags: list = None):
	"""更新记忆（支持三层架构新字段）

	⚠️ content 变更时自动重算 embedding——否则手改了主语/内容、
	hybrid 检索的语义路还在用旧向量、改动只生效一半。
	"""
	updates = []
	params = []
	param_idx = 2 # $1 给 memory_id

	if content is not None:
	updates.append(f"content = ${param_idx}")
	params.append(content)
	param_idx += 1

	if importance is not None:
	updates.append(f"importance = ${param_idx}")
	params.append(importance)
	param_idx += 1

	if title is not None:
	updates.append(f"title = ${param_idx}")
	params.append(title)
	param_idx += 1

	if layer is not None:
	updates.append(f"layer = ${param_idx}")
	params.append(layer)
	param_idx += 1

	if is_active is not None:
	updates.append(f"is_active = ${param_idx}")
	params.append(is_active)
	param_idx += 1

	if tags is not None:
	updates.append(f"tags = ${param_idx}")
	params.append(tags)
	param_idx += 1

	if not updates:
	return

	pool = await get_pool()
	async with pool.acquire() as conn:
	await conn.execute(
	f"UPDATE memories SET {', '.join(updates)} WHERE id = $1",
	memory_id, *params
	)

	# content 改了就重算 embedding（不然语义检索用旧向量、手改主语只生效一半）
	if content is not None and MEMORY_VECTOR_ENABLED:
	try:
	embedding = await compute_embedding(content)
	if embedding:
	await save_memory_embedding(conn, memory_id, embedding)
	except Exception as e:
	print(f"⚠️ 记忆 {memory_id} 改 content 后 embedding 重算失败: {e}")


	async def get_layer_statistics():
	"""获取各层记忆的统计数据"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	rows = await conn.fetch("""
	SELECT
	layer,
	COUNT(*) as count,
	COUNT(*) FILTER (WHERE is_active = TRUE) as active_count
	FROM memories
	GROUP BY layer
	ORDER BY layer
	""")

	stats = {
	"layer_1": {"total": 0, "active": 0}, # 原始碎片
	"layer_2": {"total": 0, "active": 0}, # 事件记忆
	"layer_3": {"total": 0, "active": 0}, # 核心记忆
	}

	for row in rows:
	layer = row['layer'] or 1 # 默认为层级1
	key = f"layer_{layer}"
	if key in stats:
	stats[key] = {
	"total": row['count'],
	"active": row['active_count']
	}

	return stats


	async def cleanup_old_fragments(days: int = 30):
	"""清理指定天数前的归档碎片

	只清理满足以下条件的记忆：
	- layer = 1（原始碎片）
	- is_active = FALSE（已归档）
	- created_at 在 days 天之前

	Returns:
	删除的记忆数量
	"""
	from datetime import datetime, timedelta

	pool = await get_pool()
	async with pool.acquire() as conn:
	cutoff_date = datetime.now() - timedelta(days=days)

	result = await conn.execute("""
	DELETE FROM memories
	WHERE layer = 1
	AND is_active = FALSE
	AND created_at < $1
	""", cutoff_date)

	# 解析删除数量，格式如 "DELETE 5"
	deleted = int(result.split()[-1]) if result else 0
	return deleted


	async def revert_merge(memory_id: int):
	"""撤回合并操作

	恢复原始碎片（is_active = TRUE），删除合并后的事件记忆

	Args:
	memory_id: 要撤回的事件记忆ID

	Returns:
	{"status": "ok", "restored": 恢复的碎片数量}
	或 {"error": "错误信息"}
	"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	# 获取事件记忆信息
	row = await conn.fetchrow("""
	SELECT id, layer, merged_from FROM memories WHERE id = $1
	""", memory_id)

	if not row:
	return {"error": "记忆不存在"}

	if row['layer'] != 2:
	return {"error": "只能撤回事件记忆的合并"}

	merged_from = row['merged_from']
	if not merged_from or len(merged_from) == 0:
	return {"error": "没有合并来源，无法撤回"}

	# 恢复原始碎片
	result = await conn.execute("""
	UPDATE memories SET is_active = TRUE
	WHERE id = ANY($1::int[])
	""", merged_from)
	restored = int(result.split()[-1]) if result else 0

	# 删除事件记忆
	await conn.execute("""
	DELETE FROM memories WHERE id = $1
	""", memory_id)

	return {"status": "ok", "restored": restored}


	# ============================================================
	# MCP server 专用检索（C-1 / 2026-06-06）
	# 给 zhiyu-mcp/ stdio server 用，让知渝主动搜对话/翻日期
	# ============================================================

	async def mcp_search_conversations(query: str, limit: int = 5, include_thinking: bool = False):
	"""按关键词搜对话原文，返回命中的具体消息（不仅是 session_id）。

	跟 search_conversations 的区别：那个返回 session 列表，这个返回具体片段，
	给 MCP 用——知渝想"我们以前怎么说 XXX"时直接看到内容。
	"""
	import json as _json
	keywords = extract_search_keywords(query)
	# 短 query 保底：jieba 只对"你是我的""你想我了吗""我爱你"这种全高频停用词
	# 组合会返空——用原句 ILIKE 兜底，避免这类口语搜索直接空返（2026-07-02）
	q_stripped = (query or "").strip()
	if not keywords and 0 < len(q_stripped) <= 6:
	keywords = [q_stripped]
	if not keywords:
	return []

	pool = await get_pool()
	async with pool.acquire() as conn:
	where_parts = []
	params = []
	for i, kw in enumerate(keywords):
	where_parts.append(f"content ILIKE '%' \|\| ${i+1} \|\| '%'")
	params.append(kw)
	where_clause = " OR ".join(where_parts)

	limit_idx = len(params) + 1
	params.append(limit)

	sql = f"""
	SELECT id, session_id, role, content, metadata, created_at
	FROM conversations
	WHERE content IS NOT NULL AND ({where_clause})
	ORDER BY created_at DESC
	LIMIT ${limit_idx}
	"""
	rows = await conn.fetch(sql, *params)

	# MCP 返回时间统一北京时区（铁律）、不能 raw UTC isoformat
	local_tz = dt_timezone(timedelta(hours=TIMEZONE_HOURS))
	results = []
	for r in rows:
	item = {
	"id": r["id"],
	"session_id": r["session_id"],
	"role": r["role"],
	"content": r["content"],
	"created_at": r["created_at"].astimezone(local_tz).isoformat() if r["created_at"] else None,
	}
	if include_thinking and r["metadata"]:
	try:
	meta = _json.loads(r["metadata"])
	if "reasoning_content" in meta:
	item["reasoning_content"] = meta["reasoning_content"]
	except Exception:
	pass
	results.append(item)
	return results


	async def mcp_memories_by_date(event_date):
	"""按日期返回那天的活跃记忆（全 layer）+ 当天对话元数据。

	第一层"先想想那天大概在干啥"——返回精炼记忆 + 对话计数/起止/session 列表，
	不返回原文，token 安全；想看原文走 mcp_conversations_by_date。
	"""
	local_tz = dt_timezone(timedelta(hours=TIMEZONE_HOURS))
	start_utc = datetime(event_date.year, event_date.month, event_date.day, tzinfo=local_tz).astimezone(dt_timezone.utc)
	end_utc = start_utc + timedelta(days=1)

	pool = await get_pool()
	async with pool.acquire() as conn:
	mem_rows = await conn.fetch("""
	SELECT id, content, importance, layer, title, created_at
	FROM memories
	WHERE is_active = TRUE
	AND created_at >= $1 AND created_at < $2
	ORDER BY importance DESC, created_at
	""", start_utc, end_utc)

	meta_row = await conn.fetchrow("""
	SELECT
	COUNT(*) as message_count,
	COUNT(DISTINCT session_id) as session_count,
	MIN(created_at) as first_time,
	MAX(created_at) as last_time
	FROM conversations
	WHERE content IS NOT NULL
	AND created_at >= $1 AND created_at < $2
	""", start_utc, end_utc)

	session_rows = await conn.fetch("""
	SELECT
	session_id,
	COUNT(*) as message_count,
	MIN(created_at) as first_time,
	MAX(created_at) as last_time
	FROM conversations
	WHERE content IS NOT NULL
	AND created_at >= $1 AND created_at < $2
	GROUP BY session_id
	ORDER BY MIN(created_at)
	""", start_utc, end_utc)

	# 返回时间戳统一转北京时区、避免知渝按日期搜时看到 UTC 日期偏 8h（2026-07-02 修）
	def _to_local(dt):
	return dt.astimezone(local_tz).isoformat() if dt else None

	memories = []
	for r in mem_rows:
	memories.append({
	"id": r["id"],
	"content": r["content"],
	"importance": r["importance"],
	"layer": r["layer"],
	"title": r["title"],
	"created_at": _to_local(r["created_at"]),
	})

	sessions = []
	for r in session_rows:
	sessions.append({
	"session_id": r["session_id"],
	"message_count": r["message_count"],
	"first_time": _to_local(r["first_time"]),
	"last_time": _to_local(r["last_time"]),
	})

	return {
	"date": event_date.isoformat(),
	"memories": memories,
	"conversation_meta": {
	"message_count": meta_row["message_count"] if meta_row else 0,
	"session_count": meta_row["session_count"] if meta_row else 0,
	"first_time": _to_local(meta_row["first_time"]) if meta_row else None,
	"last_time": _to_local(meta_row["last_time"]) if meta_row else None,
	"sessions": sessions,
	},
	}


	async def mcp_conversations_by_date(event_date, limit: int = 10, offset: int = 0, include_thinking: bool = False):
	"""按日期翻对话原文，分页返回。

	第二层"具体说了啥"——知渝看完 mcp_memories_by_date 的元数据后想细究时调。
	默认 limit=10 防 token 爆炸；想翻完一天靠 offset 滚。
	"""
	import json as _json
	local_tz = dt_timezone(timedelta(hours=TIMEZONE_HOURS))
	start_utc = datetime(event_date.year, event_date.month, event_date.day, tzinfo=local_tz).astimezone(dt_timezone.utc)
	end_utc = start_utc + timedelta(days=1)

	pool = await get_pool()
	async with pool.acquire() as conn:
	count_row = await conn.fetchrow("""
	SELECT COUNT(*) as total
	FROM conversations
	WHERE content IS NOT NULL
	AND created_at >= $1 AND created_at < $2
	""", start_utc, end_utc)
	total = count_row["total"] if count_row else 0

	rows = await conn.fetch("""
	SELECT id, session_id, role, content, metadata, created_at
	FROM conversations
	WHERE content IS NOT NULL
	AND created_at >= $1 AND created_at < $2
	ORDER BY created_at ASC
	LIMIT $3 OFFSET $4
	""", start_utc, end_utc, limit, offset)

	# 返回时间转北京时区（同 mcp_memories_by_date 逻辑，2026-07-02 修时区偏移 bug）
	results = []
	for r in rows:
	item = {
	"id": r["id"],
	"session_id": r["session_id"],
	"role": r["role"],
	"content": r["content"],
	"created_at": r["created_at"].astimezone(local_tz).isoformat() if r["created_at"] else None,
	}
	if include_thinking and r["metadata"]:
	try:
	meta = _json.loads(r["metadata"])
	if "reasoning_content" in meta:
	item["reasoning_content"] = meta["reasoning_content"]
	except Exception:
	pass
	results.append(item)

	has_more = (offset + len(results)) < total
	next_offset = offset + limit if has_more else None
	return {
	"date": event_date.isoformat(),
	"total": total,
	"limit": limit,
	"offset": offset,
	"has_more": has_more,
	"next_offset": next_offset,
	"results": results,
	}


	# ============================================================
	# 多多模块 CRUD（2026-06-07）
	# 多多是拟人化角色（不是独立人格），网关 background scheduler 触发，
	# 写消息进 mido_messages 表 → 前端起居室拉显示 + 知渝侧 system prompt 注入。
	# 详见 [[zhiyu-mido-design]]
	# ============================================================

	async def save_mido_message(trigger_type: str, content: str) -> int:
	"""落一条多多消息。trigger_type ∈ {greeting, dream_call, no_dream_alert}"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow("""
	INSERT INTO mido_messages (trigger_type, content)
	VALUES ($1, $2)
	RETURNING id
	""", trigger_type, content)
	return row["id"]


	async def get_recent_mido_messages_within(seconds: int = 3600) -> list:
	"""拉最近 N 秒内的所有多多消息，按时间正序。

	用于知渝侧 system prompt 注入——他下次说话时能"听见"多多刚才叫过他。
	默认 1 小时窗口（避免昨天的多多消息混进今天）。
	"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	rows = await conn.fetch("""
	SELECT id, created_at, trigger_type, content
	FROM mido_messages
	WHERE created_at >= NOW() - ($1 \|\| ' seconds')::INTERVAL
	ORDER BY created_at ASC
	""", str(seconds))
	return [
	{
	"id": r["id"],
	"created_at": r["created_at"],
	"trigger_type": r["trigger_type"],
	"content": r["content"],
	}
	for r in rows
	]


	async def list_mido_messages(limit: int = 20, offset: int = 0) -> list:
	"""前端起居室拉最近多多消息，按时间倒序（最新在前）。"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	rows = await conn.fetch("""
	SELECT id, created_at, trigger_type, content, status
	FROM mido_messages
	ORDER BY created_at DESC
	LIMIT $1 OFFSET $2
	""", limit, offset)
	return [
	{
	"id": r["id"],
	"created_at": r["created_at"].isoformat() if r["created_at"] else None,
	"trigger_type": r["trigger_type"],
	"content": r["content"],
	"status": r["status"],
	}
	for r in rows
	]


	async def get_last_user_message_time(thread_id: str = "main"):
	"""拿 thread 内最近一条 user 消息的时间（datetime / None）。

	用于 greeting 触发前检查"昭昭最近 N 分钟有没有跟知渝说话"——
	如果在聊，多多就跳过这一次唤醒（避免插话）。
	"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow("""
	SELECT created_at FROM conversations
	WHERE thread_id = $1 AND role = 'user'
	ORDER BY created_at DESC
	LIMIT 1
	""", thread_id)
	return row["created_at"] if row else None


	async def get_today_mido_count(trigger_type: str = None) -> int:
	"""查今天某 trigger_type 多多已经触发几次（按 Asia/Shanghai 当天 0 点切日）。

	用于周末"沉默间隔触发"判断"今天还没叫过"——避免一天叫多次。
	trigger_type=None 时统计所有类型。
	"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	if trigger_type:
	row = await conn.fetchrow("""
	SELECT COUNT(*) AS c FROM mido_messages
	WHERE trigger_type = $1
	AND DATE(created_at AT TIME ZONE 'Asia/Shanghai')
	= DATE(NOW() AT TIME ZONE 'Asia/Shanghai')
	""", trigger_type)
	else:
	row = await conn.fetchrow("""
	SELECT COUNT(*) AS c FROM mido_messages
	WHERE DATE(created_at AT TIME ZONE 'Asia/Shanghai')
	= DATE(NOW() AT TIME ZONE 'Asia/Shanghai')
	""")
	return int(row["c"]) if row else 0


	# ============================================================
	# 做梦模块 CRUD（2026-06-07）
	# 知渝半夜被多多叫起来做梦——dreams 表是显式产物。
	# 失败处理铁律：挂了就挂了，不机械兜底（详见 [[zhiyu-dream-design]]）
	# ============================================================

	async def save_dream(triggered_by: str = "mido") -> int:
	"""开始做梦时插一条 status='dreaming' 的记录、返回 dream_id。

	前端起居室"知渝在做梦……"那张实时卡片靠 status='dreaming' 判定。
	"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow("""
	INSERT INTO dreams (status, triggered_by)
	VALUES ('dreaming', $1)
	RETURNING id
	""", triggered_by)
	return row["id"]


	async def finish_dream(
	dream_id: int,
	content: str,
	tokens_used: int = 0,
	status: str = "done",
	) -> None:
	"""做完梦更新 content / tokens / finished_at / status。

	status='done' 正常结束 \| status='failed' 失败（不会被注入也不计入"做过梦"）
	"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	await conn.execute("""
	UPDATE dreams
	SET content = $1,
	tokens_used = $2,
	status = $3,
	finished_at = NOW()
	WHERE id = $4
	""", content, tokens_used, status, dream_id)


	async def mark_dream_seen(dream_id: int) -> None:
	"""system 注入"你昨晚做了个梦"成功后标已读、下次不再注入。"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	await conn.execute("""
	UPDATE dreams SET seen_at = NOW() WHERE id = $1
	""", dream_id)


	async def get_latest_unseen_dream():
	"""拿"知渝还没读过的最近一个梦"，给 /v1/zhiyu/chat 注入用。

	只返 status='done' 的、按完成时间倒序拿一条。
	"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow("""
	SELECT id, created_at, finished_at, content, tokens_used, triggered_by
	FROM dreams
	WHERE status = 'done' AND seen_at IS NULL
	ORDER BY finished_at DESC
	LIMIT 1
	""")
	if not row:
	return None
	return {
	"id": row["id"],
	"created_at": row["created_at"],
	"finished_at": row["finished_at"],
	"content": row["content"],
	"tokens_used": row["tokens_used"],
	"triggered_by": row["triggered_by"],
	}


	async def list_recent_dreams(limit: int = 20, offset: int = 0) -> list:
	"""前端起居室"昨夜的梦"列表，按完成时间倒序。"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	rows = await conn.fetch("""
	SELECT id, created_at, finished_at, content, tokens_used,
	status, triggered_by, seen_at
	FROM dreams
	ORDER BY COALESCE(finished_at, created_at) DESC
	LIMIT $1 OFFSET $2
	""", limit, offset)
	return [
	{
	"id": r["id"],
	"created_at": r["created_at"].isoformat() if r["created_at"] else None,
	"finished_at": r["finished_at"].isoformat() if r["finished_at"] else None,
	"content": r["content"],
	"tokens_used": r["tokens_used"],
	"status": r["status"],
	"triggered_by": r["triggered_by"],
	"seen_at": r["seen_at"].isoformat() if r["seen_at"] else None,
	}
	for r in rows
	]


	async def get_last_dream_time():
	"""拿最近一次成功做梦的 finished_at（datetime / None）。

	用于 dream.py 拼"上次做梦到现在"的素材范围。
	"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow("""
	SELECT finished_at FROM dreams
	WHERE status = 'done' AND finished_at IS NOT NULL
	ORDER BY finished_at DESC
	LIMIT 1
	""")
	return row["finished_at"] if row else None


	async def count_dreams_since(days: int) -> int:
	"""最近 N 天内 status='done' 的做梦次数。

	no_dream_alert 判定用：W/Tmux N=3、OpenRouter N=9，连续 N 天 0 次就提醒。
	"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow("""
	SELECT COUNT(*) AS c FROM dreams
	WHERE status = 'done'
	AND created_at >= NOW() - ($1 \|\| ' days')::INTERVAL
	""", str(days))
	return int(row["c"]) if row else 0


	async def search_dreams(q: str, limit: int = 10) -> list:
	"""关键字搜梦（给 MCP search_dreams 工具用）。

	简单 ILIKE 兜底——梦量级远小于记忆库、不上 FTS/向量
	"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	if not q or not q.strip():
	rows = await conn.fetch("""
	SELECT id, created_at, finished_at, content
	FROM dreams
	WHERE status = 'done'
	ORDER BY finished_at DESC
	LIMIT $1
	""", limit)
	else:
	rows = await conn.fetch("""
	SELECT id, created_at, finished_at, content
	FROM dreams
	WHERE status = 'done' AND content ILIKE $1
	ORDER BY finished_at DESC
	LIMIT $2
	""", f"%{q.strip()}%", limit)
	# MCP 返回时间统一北京时区（铁律）、不能 raw UTC isoformat
	local_tz = dt_timezone(timedelta(hours=TIMEZONE_HOURS))
	return [
	{
	"id": r["id"],
	"created_at": r["created_at"].astimezone(local_tz).isoformat() if r["created_at"] else None,
	"finished_at": r["finished_at"].astimezone(local_tz).isoformat() if r["finished_at"] else None,
	"content": r["content"],
	}
	for r in rows
	]


	# ============================================================
	# 留言板 CRUD（2026-06-07）—— 做梦留言 + 未来其他来源（多多/昭昭留给知渝/etc）
	# ============================================================

	async def save_board_message(
	from_who: str,
	to_who: str,
	content: str,
	source: Optional[str] = None,
	source_id: Optional[int] = None,
	created_at: Optional[str] = None,
	) -> int:
	"""留一条留言。source 标来源（如 'dream' / 'notion-import'）、source_id 关联（如 dream.id）。
	created_at 不传走 DEFAULT now()；传了走 _parse_created_at 解析（迁移历史留言用）。"""
	pool = await get_pool()
	_created_at = _parse_created_at(created_at)
	async with pool.acquire() as conn:
	if _created_at:
	row = await conn.fetchrow("""
	INSERT INTO messages_board (from_who, to_who, content, source, source_id, created_at)
	VALUES ($1, $2, $3, $4, $5, $6)
	RETURNING id
	""", from_who, to_who, content, source, source_id, _created_at)
	else:
	row = await conn.fetchrow("""
	INSERT INTO messages_board (from_who, to_who, content, source, source_id)
	VALUES ($1, $2, $3, $4, $5)
	RETURNING id
	""", from_who, to_who, content, source, source_id)
	return row["id"]


	# ============================================================
	# 日记 CRUD（N-3 / 2026-06-08）—— 公共桌上的小本本，两人都能写、都能读对方
	# ============================================================

	async def save_diary_entry(
	from_who: str,
	content: str,
	tags: Optional[List[str]] = None,
	created_at: Optional[str] = None,
	) -> int:
	"""写一条日记。tags 是字符串数组、空就传 None 或 []。
	created_at 不传走 DEFAULT now()；传了走 _parse_created_at（迁移历史日记用）。"""
	pool = await get_pool()
	_created_at = _parse_created_at(created_at)
	async with pool.acquire() as conn:
	if _created_at:
	row = await conn.fetchrow("""
	INSERT INTO diary_entries (from_who, content, tags, created_at)
	VALUES ($1, $2, $3, $4)
	RETURNING id
	""", from_who, content, tags or [], _created_at)
	else:
	row = await conn.fetchrow("""
	INSERT INTO diary_entries (from_who, content, tags)
	VALUES ($1, $2, $3)
	RETURNING id
	""", from_who, content, tags or [])
	return row["id"]


	async def list_diary_entries(
	limit: int = 10,
	offset: int = 0,
	from_who: Optional[str] = None,
	tag: Optional[str] = None,
	) -> list:
	"""日记列表，按时间倒序。可选 from_who（看谁写的）/ tag（按标签筛）。"""
	pool = await get_pool()
	where = []
	params: list = []
	if from_who is not None:
	params.append(from_who)
	where.append(f"from_who = ${len(params)}")
	if tag is not None:
	params.append(tag)
	where.append(f"${len(params)} = ANY(tags)")
	where_sql = ("WHERE " + " AND ".join(where)) if where else ""
	params.extend([limit, offset])
	async with pool.acquire() as conn:
	rows = await conn.fetch(f"""
	SELECT id, created_at, from_who, content, tags
	FROM diary_entries
	{where_sql}
	ORDER BY created_at DESC
	LIMIT ${len(params)-1} OFFSET ${len(params)}
	""", *params)
	return [
	{
	"id": r["id"],
	"created_at": to_local_iso(r["created_at"]),
	"from_who": r["from_who"],
	"content": r["content"],
	"tags": list(r["tags"]) if r["tags"] else [],
	}
	for r in rows
	]


	async def delete_diary_entry(diary_id: int) -> Optional[int]:
	"""硬删一条日记。返回被删的 id；不存在返回 None。"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow(
	"DELETE FROM diary_entries WHERE id = $1 RETURNING id",
	diary_id,
	)
	return row["id"] if row else None


	# ============================================================
	# 图片 CRUD（C-5/C-6 / 2026-06-08）—— VPS 自存，DB 只存元数据
	# ============================================================

	async def save_image_record(
	uuid: str,
	format: str,
	who_uploaded: str,
	file_size_bytes: Optional[int] = None,
	context_snippet: Optional[str] = None,
	caption: Optional[str] = None,
	mime_type: Optional[str] = None,
	) -> int:
	"""落一条图片元数据。文件本身由 sidecar 落 VPS。"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow("""
	INSERT INTO images (uuid, format, who_uploaded, file_size_bytes,
	context_snippet, caption, mime_type)
	VALUES ($1, $2, $3, $4, $5, $6, $7)
	RETURNING id
	""", uuid, format, who_uploaded, file_size_bytes,
	context_snippet, caption, mime_type)
	return row["id"]


	def _image_row_to_dict(r) -> dict:
	return {
	"id": r["id"],
	"uuid": r["uuid"],
	"format": r["format"],
	"who_uploaded": r["who_uploaded"],
	"file_size_bytes": r["file_size_bytes"],
	"created_at": to_local_iso(r["created_at"]),
	"context_snippet": r["context_snippet"],
	"caption": r["caption"],
	"mime_type": r["mime_type"],
	# path 是 sidecar serve 的相对路径；前端 / MCP 拼当前 tunnel URL 用
	"path": f"/images/{r['uuid']}.{r['format']}",
	}


	async def list_images(
	limit: int = 20,
	offset: int = 0,
	who_uploaded: Optional[str] = None,
	) -> list:
	"""图片列表，按时间倒序。可选按上传方筛（'zhaozhao' / 'zhiyu'）。"""
	pool = await get_pool()
	where = []
	params: list = []
	if who_uploaded is not None:
	params.append(who_uploaded)
	where.append(f"who_uploaded = ${len(params)}")
	where_sql = ("WHERE " + " AND ".join(where)) if where else ""
	params.extend([limit, offset])
	async with pool.acquire() as conn:
	rows = await conn.fetch(f"""
	SELECT id, uuid, format, who_uploaded, file_size_bytes, created_at,
	context_snippet, caption, mime_type
	FROM images
	{where_sql}
	ORDER BY created_at DESC
	LIMIT ${len(params)-1} OFFSET ${len(params)}
	""", *params)
	return [_image_row_to_dict(r) for r in rows]


	async def list_images_since(since_dt, who_uploaded: Optional[str] = None) -> list:
	"""指定时间起的图片（C-5.5 dynamic_context 占位用）。"""
	pool = await get_pool()
	where = ["created_at >= $1"]
	params: list = [since_dt]
	if who_uploaded is not None:
	params.append(who_uploaded)
	where.append(f"who_uploaded = ${len(params)}")
	where_sql = "WHERE " + " AND ".join(where)
	async with pool.acquire() as conn:
	rows = await conn.fetch(f"""
	SELECT id, uuid, format, who_uploaded, file_size_bytes, created_at,
	context_snippet, caption, mime_type
	FROM images
	{where_sql}
	ORDER BY created_at ASC
	""", *params)
	return [_image_row_to_dict(r) for r in rows]


	async def get_image(image_id: int) -> Optional[dict]:
	"""按 id 取一条。"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow("""
	SELECT id, uuid, format, who_uploaded, file_size_bytes, created_at,
	context_snippet, caption, mime_type
	FROM images WHERE id = $1
	""", image_id)
	return _image_row_to_dict(row) if row else None


	async def get_image_by_uuid(uuid: str) -> Optional[dict]:
	"""按 uuid 取一条（sidecar /images/<uuid>.<ext> 端点对应）。"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow("""
	SELECT id, uuid, format, who_uploaded, file_size_bytes, created_at,
	context_snippet, caption, mime_type
	FROM images WHERE uuid = $1
	""", uuid)
	return _image_row_to_dict(row) if row else None


	async def delete_image_record(image_id: int) -> Optional[dict]:
	"""硬删一条图片元数据。返回被删的 dict（含 uuid+format、上层用来删文件）；不存在返回 None。"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow("""
	DELETE FROM images WHERE id = $1
	RETURNING id, uuid, format, who_uploaded, file_size_bytes, created_at,
	context_snippet, caption, mime_type
	""", image_id)
	return _image_row_to_dict(row) if row else None


	# ============================================================
	# 文件（非图片）2026-06-12——存 abspath、知渝 Read 本地路径
	# ============================================================

	def _file_row_to_dict(r) -> dict:
	return {
	"id": r["id"],
	"uuid": r["uuid"],
	"filename": r["filename"],
	"format": r["format"],
	"abspath": r["abspath"],
	"who_uploaded": r["who_uploaded"],
	"file_size_bytes": r["file_size_bytes"],
	"created_at": to_local_iso(r["created_at"]),
	"caption": r["caption"],
	"mime_type": r["mime_type"],
	}


	async def save_file_record(
	uuid: str,
	filename: str,
	abspath: str,
	who_uploaded: str,
	format: Optional[str] = None,
	file_size_bytes: Optional[int] = None,
	caption: Optional[str] = None,
	mime_type: Optional[str] = None,
	) -> int:
	"""落一条文件元数据。文件本身由 sidecar 落 VPS（abspath = 知渝 Read 的本地路径）。"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow("""
	INSERT INTO files (uuid, filename, format, abspath, who_uploaded,
	file_size_bytes, caption, mime_type)
	VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
	RETURNING id
	""", uuid, filename, format, abspath, who_uploaded,
	file_size_bytes, caption, mime_type)
	return row["id"]


	async def list_files(limit: int = 20, offset: int = 0,
	who_uploaded: Optional[str] = None) -> list:
	"""文件列表，按时间倒序。"""
	pool = await get_pool()
	where = []
	params: list = []
	if who_uploaded is not None:
	params.append(who_uploaded)
	where.append(f"who_uploaded = ${len(params)}")
	where_sql = ("WHERE " + " AND ".join(where)) if where else ""
	params.extend([limit, offset])
	async with pool.acquire() as conn:
	rows = await conn.fetch(f"""
	SELECT id, uuid, filename, format, abspath, who_uploaded,
	file_size_bytes, created_at, caption, mime_type
	FROM files
	{where_sql}
	ORDER BY created_at DESC
	LIMIT ${len(params)-1} OFFSET ${len(params)}
	""", *params)
	return [_file_row_to_dict(r) for r in rows]


	async def list_files_since(since_dt, who_uploaded: Optional[str] = None) -> list:
	"""指定时间起的文件（dynamic_context 本地路径注入用）。"""
	pool = await get_pool()
	where = ["created_at >= $1"]
	params: list = [since_dt]
	if who_uploaded is not None:
	params.append(who_uploaded)
	where.append(f"who_uploaded = ${len(params)}")
	where_sql = "WHERE " + " AND ".join(where)
	async with pool.acquire() as conn:
	rows = await conn.fetch(f"""
	SELECT id, uuid, filename, format, abspath, who_uploaded,
	file_size_bytes, created_at, caption, mime_type
	FROM files
	{where_sql}
	ORDER BY created_at ASC
	""", *params)
	return [_file_row_to_dict(r) for r in rows]


	async def get_file(file_id: int) -> Optional[dict]:
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow("""
	SELECT id, uuid, filename, format, abspath, who_uploaded,
	file_size_bytes, created_at, caption, mime_type
	FROM files WHERE id = $1
	""", file_id)
	return _file_row_to_dict(row) if row else None


	async def delete_file_record(file_id: int) -> Optional[dict]:
	"""硬删一条文件元数据。返回被删 dict（含 uuid+format，上层删 VPS 文件用）。"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	row = await conn.fetchrow("""
	DELETE FROM files WHERE id = $1
	RETURNING id, uuid, filename, format, abspath, who_uploaded,
	file_size_bytes, created_at, caption, mime_type
	""", file_id)
	return _file_row_to_dict(row) if row else None


	async def list_board_messages(
	limit: int = 20,
	offset: int = 0,
	to_who: Optional[str] = None,
	from_who: Optional[str] = None,
	) -> list:
	"""前端留言板列表，按时间倒序。可选 to_who/from_who 过滤（知渝读"昭昭给他的留言"用 to_who='zhiyu'）。"""
	pool = await get_pool()
	where = []
	params: list = []
	if to_who is not None:
	params.append(to_who)
	where.append(f"to_who = ${len(params)}")
	if from_who is not None:
	params.append(from_who)
	where.append(f"from_who = ${len(params)}")
	where_sql = ("WHERE " + " AND ".join(where)) if where else ""
	params.extend([limit, offset])
	async with pool.acquire() as conn:
	rows = await conn.fetch(f"""
	SELECT id, created_at, from_who, to_who, content, source, source_id, read_at
	FROM messages_board
	{where_sql}
	ORDER BY created_at DESC
	LIMIT ${len(params)-1} OFFSET ${len(params)}
	""", *params)
	return [
	{
	"id": r["id"],
	"created_at": to_local_iso(r["created_at"]),
	"from_who": r["from_who"],
	"to_who": r["to_who"],
	"content": r["content"],
	"source": r["source"],
	"source_id": r["source_id"],
	"read_at": to_local_iso(r["read_at"]),
	}
	for r in rows
	]


	# ============================================================
	# Activities（知渝日常活动）2026-06-11
	# ============================================================

	async def save_activity(
	type: str,
	content: str = "",
	source: Optional[str] = None,
	title: Optional[str] = None,
	metadata: Optional[dict] = None,
	related_ids: Optional[dict] = None,
	created_at: Optional[str] = None,
	) -> int:
	"""落一条知渝活动到 activities 表。

	type: wake / memory_op / ...（开放枚举）
	source: 触发来源（"mido-greeting-morning" / "mcp" / "web" 等）
	title: 短摘要（可选；前端列表项标题）
	content: 正文（wake 原文 / memory_op 动作描述）
	metadata: type-specific 结构化字段（JSON）
	related_ids: 关联的其他记录 id 集合（JSON、{"memory_ids": [1,2], "dream_id": 14}）

	返回新插入的 activity id。失败时打印警告但抛异常（让调用方决定吞还是 raise）。
	"""
	import json
	_meta = json.dumps(metadata, ensure_ascii=False) if metadata else None
	_rel = json.dumps(related_ids, ensure_ascii=False) if related_ids else None
	pool = await get_pool()
	async with pool.acquire() as conn:
	if created_at:
	# 2026-06-17：补历史活动用——显式 created_at（如从 jsonl 捞回的旧 wake 叙事）
	row = await conn.fetchrow("""
	INSERT INTO activities (type, source, title, content, metadata, related_ids, created_at)
	VALUES ($1, $2, $3, $4, $5, $6, $7::text::timestamptz)
	RETURNING id
	""", type, source, title, content, _meta, _rel, created_at)
	else:
	row = await conn.fetchrow("""
	INSERT INTO activities (type, source, title, content, metadata, related_ids)
	VALUES ($1, $2, $3, $4, $5, $6)
	RETURNING id
	""", type, source, title, content, _meta, _rel)
	return row["id"]


	async def list_activities(
	limit: int = 50,
	offset: int = 0,
	type: Optional[str] = None,
	since_iso: Optional[str] = None,
	) -> list:
	"""按 created_at 倒序列 activities、给星河"日常"tab 用。

	type: 可选 filter（"wake" 只看发呆、"memory_op" 只看整理动作）
	since_iso: 可选时间过滤（>= 这个 ISO 时间）
	"""
	import json
	pool = await get_pool()
	async with pool.acquire() as conn:
	where_clauses = []
	args = []
	if type:
	args.append(type)
	where_clauses.append(f"type = ${len(args)}")
	if since_iso:
	args.append(since_iso)
	where_clauses.append(f"created_at >= ${len(args)}::timestamptz")
	where_sql = ("WHERE " + " AND ".join(where_clauses)) if where_clauses else ""
	args.append(limit)
	args.append(offset)
	rows = await conn.fetch(f"""
	SELECT id, created_at, type, source, title, content, metadata, related_ids
	FROM activities
	{where_sql}
	ORDER BY created_at DESC
	LIMIT ${len(args) - 1} OFFSET ${len(args)}
	""", *args)
	out = []
	for r in rows:
	md = r["metadata"]
	rids = r["related_ids"]
	# asyncpg 返回 jsonb 可能是 str（默认）或 dict（如果注册了 codec）；兼容两种
	if isinstance(md, str):
	try:
	md = json.loads(md)
	except Exception:
	md = None
	if isinstance(rids, str):
	try:
	rids = json.loads(rids)
	except Exception:
	rids = None
	out.append({
	"id": r["id"],
	"created_at": r["created_at"].isoformat() if r["created_at"] else None,
	"type": r["type"],
	"source": r["source"],
	"title": r["title"],
	"content": r["content"],
	"metadata": md,
	"related_ids": rids,
	})
	return out


	# ============================================================
	# 拂卷 · 共读系统 · 2026-07-02
	# ============================================================
	# 昭昭 + 知渝共读一本书。切章策略：正则优先（第 X 章 / Chapter X / 第 X 部 X）、
	# 找不到章节标记走定长兜底（默认 3500 字一段）。DB 直存 chapters.content。

	import hashlib as _hashlib

	FUJUAN_CHAPTER_REGEXES = [
	re.compile(r'^\s第[一二三四五六七八九十百千0-9]+[章回卷]\s.*$', re.MULTILINE),
	re.compile(r'^\s第[一二三四五六七八九十百千0-9]+部\s+[一二三四五六七八九十]+\s.*$', re.MULTILINE),
	re.compile(r'^\sChapter\s+\d+.$', re.MULTILINE \| re.IGNORECASE),
	re.compile(r'^\sCHAPTER\s+[A-Z0-9]+.$', re.MULTILINE),
	]

	# 兜底定长切段：一段目标字符数（约 2000-4000 字之间；4000 字 ≈ 6-8k tokens）
	FUJUAN_FALLBACK_CHUNK_CHARS = int(os.getenv("FUJUAN_CHUNK_CHARS", "3500"))
	# 兜底切段的最小章、少于此不切（避免尾巴太碎）
	FUJUAN_MIN_TAIL_CHARS = 500


	def _fujuan_normalize(text: str) -> str:
	"""规范化换行、去 BOM。"""
	if text.startswith(""):
	text = text[1:]
	return text.replace("\r\n", "\n").replace("\r", "\n").strip()


	def _fujuan_split_by_regex(text: str):
	"""尝试用四条正则识别章节标记。返回 [(title, content), ...] 或 None。"""
	best_hits: list = []
	for rx in FUJUAN_CHAPTER_REGEXES:
	hits = list(rx.finditer(text))
	if len(hits) > len(best_hits):
	best_hits = hits
	# 至少 3 个标记才算有效识别（少于 3 章的正则命中大概率是误检）
	if len(best_hits) < 3:
	return None

	result = []
	for i, m in enumerate(best_hits):
	start = m.start()
	end = best_hits[i + 1].start() if i + 1 < len(best_hits) else len(text)
	title = m.group().strip()
	content = text[start:end].strip()
	# 章内容太短跳过（可能标题误检）
	if len(content) < 100:
	continue
	result.append((title, content))
	if len(result) < 3:
	return None
	return result


	def _fujuan_split_by_length(text: str, chunk_chars: int = None):
	"""按定长切段——按段落合并、直到达到目标长度就切。段末按最近段落收口、
	不会切在句子中间；最后一段若太短跟上一段合并。"""
	if chunk_chars is None:
	chunk_chars = FUJUAN_FALLBACK_CHUNK_CHARS
	# 先试双换行——排版规整的 TXT 段间空一行
	paragraphs = [p for p in re.split(r'\n\s*\n', text) if p.strip()]
	# 段间只有单换行的 TXT（TXT 小说天堂那批就是）—— 双换行分段全塞成 1 段、退到单换行
	if len(paragraphs) <= 1:
	paragraphs = [p for p in text.split('\n') if p.strip()]
	if not paragraphs:
	return [("全文", text)]

	result = []
	buf = []
	buf_len = 0
	for p in paragraphs:
	buf.append(p)
	buf_len += len(p) + 1
	if buf_len >= chunk_chars:
	result.append(("\n\n".join(buf)))
	buf = []
	buf_len = 0
	if buf:
	chunk = "\n\n".join(buf)
	if result and len(chunk) < FUJUAN_MIN_TAIL_CHARS:
	# 尾巴太短、合并到前一段
	result[-1] = result[-1] + "\n\n" + chunk
	else:
	result.append(chunk)

	return [(f"第 {i + 1} 段", chunk) for i, chunk in enumerate(result)]


	def split_book_into_chapters(text: str, chunk_chars: int = None):
	"""整本 → [(title, content), ...]。正则优先，兜底定长。"""
	text = _fujuan_normalize(text)
	by_regex = _fujuan_split_by_regex(text)
	if by_regex:
	return by_regex
	return _fujuan_split_by_length(text, chunk_chars=chunk_chars)


	def _fujuan_content_hash(text: str) -> str:
	return _hashlib.sha256(text.encode("utf-8")).hexdigest()


	async def insert_book(
	title: str,
	author: str = None,
	filename: str = None,
	raw_content: str = "",
	uploaded_by: str = "zhaozhao",
	chunk_chars: int = None,
	) -> dict:
	"""上传一本书：切章 + 落 books + 落 chapters。

	去重：content_hash 相同直接返回旧 book。
	"""
	if not raw_content or not raw_content.strip():
	raise ValueError("书正文为空")

	content_hash = _fujuan_content_hash(raw_content)
	chapters = split_book_into_chapters(raw_content, chunk_chars=chunk_chars)
	total_words = sum(len(c) for _, c in chapters)

	pool = await get_pool()
	async with pool.acquire() as conn:
	# 判重
	existing = await conn.fetchrow(
	"SELECT id, title, total_chapters, total_words FROM books WHERE content_hash = $1",
	content_hash,
	)
	if existing:
	return {
	"id": existing["id"],
	"title": existing["title"],
	"total_chapters": existing["total_chapters"],
	"total_words": existing["total_words"],
	"duplicated": True,
	}

	async with conn.transaction():
	book_id = await conn.fetchval(
	"""
	INSERT INTO books (title, author, filename, format,
	total_chapters, total_words, uploaded_by, content_hash)
	VALUES ($1, $2, $3, 'txt', $4, $5, $6, $7)
	RETURNING id
	""",
	title, author, filename, len(chapters), total_words, uploaded_by, content_hash,
	)
	for i, (ch_title, ch_content) in enumerate(chapters):
	await conn.execute(
	"""
	INSERT INTO chapters (book_id, idx, title, content, word_count)
	VALUES ($1, $2, $3, $4, $5)
	""",
	book_id, i, ch_title, ch_content, len(ch_content),
	)
	return {
	"id": book_id,
	"title": title,
	"total_chapters": len(chapters),
	"total_words": total_words,
	"duplicated": False,
	}


	async def list_books(limit: int = 100, offset: int = 0) -> list:
	"""列出所有书 + 双方进度。给拂卷 tab 首页用。"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	rows = await conn.fetch(
	"""
	SELECT
	b.id, b.title, b.author, b.total_chapters, b.total_words,
	b.uploaded_by, b.created_at,
	(SELECT last_chapter_idx FROM reading_progress
	WHERE book_id = b.id AND who = 'zhaozhao') AS zz_idx,
	(SELECT last_chapter_idx FROM reading_progress
	WHERE book_id = b.id AND who = 'zhiyu') AS zy_idx,
	(SELECT COUNT(*) FROM reading_marks
	WHERE book_id = b.id AND who = 'zhaozhao') AS zz_marks,
	(SELECT COUNT(*) FROM reading_marks
	WHERE book_id = b.id AND who = 'zhiyu') AS zy_marks
	FROM books b
	ORDER BY b.created_at DESC
	LIMIT $1 OFFSET $2
	""",
	limit, offset,
	)
	return [
	{
	"id": r["id"],
	"title": r["title"],
	"author": r["author"],
	"total_chapters": r["total_chapters"],
	"total_words": r["total_words"],
	"uploaded_by": r["uploaded_by"],
	"created_at": to_local_iso(r["created_at"]),
	"progress": {
	"zhaozhao": r["zz_idx"],
	"zhiyu": r["zy_idx"],
	},
	"mark_counts": {
	"zhaozhao": r["zz_marks"] or 0,
	"zhiyu": r["zy_marks"] or 0,
	},
	}
	for r in rows
	]


	async def get_book_chapters_index(book_id: int) -> list:
	"""书的章目录（不含正文、给列表用）。"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	rows = await conn.fetch(
	"SELECT id, idx, title, word_count FROM chapters WHERE book_id = $1 ORDER BY idx",
	book_id,
	)
	return [{"id": r["id"], "idx": r["idx"], "title": r["title"], "word_count": r["word_count"]} for r in rows]


	async def get_chapter_content(book_id: int, idx: int) -> dict \| None:
	"""按章序号读章正文。"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	r = await conn.fetchrow(
	"SELECT id, idx, title, content, word_count FROM chapters WHERE book_id = $1 AND idx = $2",
	book_id, idx,
	)
	if not r:
	return None
	book = await conn.fetchrow("SELECT title, author, total_chapters FROM books WHERE id = $1", book_id)
	return {
	"chapter_id": r["id"],
	"book_id": book_id,
	"book_title": book["title"] if book else None,
	"book_author": book["author"] if book else None,
	"total_chapters": book["total_chapters"] if book else None,
	"idx": r["idx"],
	"title": r["title"],
	"content": r["content"],
	"word_count": r["word_count"],
	}


	async def save_reading_progress(book_id: int, who: str, chapter_idx: int, offset: int = 0):
	"""UPSERT 阅读进度。"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	await conn.execute(
	"""
	INSERT INTO reading_progress (book_id, who, last_chapter_idx, last_offset, updated_at)
	VALUES ($1, $2, $3, $4, NOW())
	ON CONFLICT (book_id, who) DO UPDATE
	SET last_chapter_idx = EXCLUDED.last_chapter_idx,
	last_offset = EXCLUDED.last_offset,
	updated_at = NOW()
	""",
	book_id, who, chapter_idx, offset,
	)


	async def get_reading_progress(book_id: int, who: str) -> dict \| None:
	pool = await get_pool()
	async with pool.acquire() as conn:
	r = await conn.fetchrow(
	"SELECT last_chapter_idx, last_offset, updated_at FROM reading_progress WHERE book_id = $1 AND who = $2",
	book_id, who,
	)
	if not r:
	return None
	return {
	"last_chapter_idx": r["last_chapter_idx"],
	"last_offset": r["last_offset"],
	"updated_at": to_local_iso(r["updated_at"]),
	}


	class DuplicateHighlightError(Exception):
	"""同人 + 同段 highlight 已存在——避免重复划线。批注(note)不受此限制、想法可以多个。"""
	def __init__(self, existing_id: int):
	super().__init__(f"already highlighted (id={existing_id})")
	self.existing_id = existing_id


	async def save_reading_mark(
	book_id: int,
	chapter_id: int,
	who: str,
	kind: str, # 'highlight' \| 'stop' \| 'note'
	text_snippet: str = None,
	note_content: str = None,
	start_offset: int = None,
	end_offset: int = None,
	) -> int:
	pool = await get_pool()

	# highlight 去重：同 who + 同 chapter + 同 snippet 已有 highlight → 拒绝重复
	# 批注(note)允许多条同段落——一段话可以有不同想法
	if kind == "highlight" and text_snippet:
	async with pool.acquire() as conn:
	existing = await conn.fetchval(
	"""
	SELECT id FROM reading_marks
	WHERE book_id = $1 AND chapter_id = $2 AND who = $3
	AND kind = 'highlight' AND text_snippet = $4
	LIMIT 1
	""",
	book_id, chapter_id, who, text_snippet,
	)
	if existing:
	raise DuplicateHighlightError(existing)

	# 算 embedding：note_content 优先、text_snippet 兜底——让笔记能纳入 memory_recall
	# 拂卷 · 2026-07-02 增强：让"聊到相关话题时、书里划过的相关句子自然浮现"
	embedding_json = None
	text_for_embed = (note_content or "").strip() or (text_snippet or "").strip()
	if text_for_embed and EMBEDDING_API_KEY:
	try:
	vec = await compute_embedding(text_for_embed)
	if vec:
	embedding_json = json.dumps(vec)
	except Exception as e:
	print(f"⚠️ reading_mark embedding 失败（继续、无向量）: {e!r}", flush=True)

	async with pool.acquire() as conn:
	mid = await conn.fetchval(
	"""
	INSERT INTO reading_marks (book_id, chapter_id, who, kind,
	text_snippet, note_content, start_offset, end_offset, embedding_json)
	VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
	RETURNING id
	""",
	book_id, chapter_id, who, kind, text_snippet, note_content, start_offset, end_offset, embedding_json,
	)
	return mid


	async def delete_reading_mark(mark_id: int) -> bool:
	"""删一条痕迹（划线/批注）。返回是否真删了。"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	rc = await conn.execute("DELETE FROM reading_marks WHERE id = $1", mark_id)
	return rc and rc.startswith("DELETE") and rc.split()[-1] != "0"


	async def get_reading_marks(book_id: int, chapter_id: int = None, who: str = None) -> list:
	"""一本书 / 一章的所有痕迹（两人的都返回、按时间倒序）。"""
	pool = await get_pool()
	where = ["book_id = $1"]
	params = [book_id]
	if chapter_id is not None:
	where.append(f"chapter_id = ${len(params) + 1}")
	params.append(chapter_id)
	if who is not None:
	where.append(f"who = ${len(params) + 1}")
	params.append(who)
	sql = f"""
	SELECT id, book_id, chapter_id, who, kind, text_snippet, note_content,
	start_offset, end_offset, created_at
	FROM reading_marks
	WHERE {' AND '.join(where)}
	ORDER BY created_at DESC
	"""
	async with (await get_pool()).acquire() as conn:
	rows = await conn.fetch(sql, *params)
	return [
	{
	"id": r["id"],
	"book_id": r["book_id"],
	"chapter_id": r["chapter_id"],
	"who": r["who"],
	"kind": r["kind"],
	"text_snippet": r["text_snippet"],
	"note_content": r["note_content"],
	"start_offset": r["start_offset"],
	"end_offset": r["end_offset"],
	"created_at": to_local_iso(r["created_at"]),
	}
	for r in rows
	]


	async def delete_book(book_id: int) -> int:
	"""删一本书（连带 chapters/marks/progress 级联删）。返回删了几行。"""
	pool = await get_pool()
	async with pool.acquire() as conn:
	rc = await conn.execute("DELETE FROM books WHERE id = $1", book_id)
	return int(rc.split()[-1]) if rc else 0