feat: 数据归档脚本（导出/清理/VACUUM/dry-run）

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-10-14 09:23:42 +08:00
parent ac9a4da51d
commit 3a3292c239
2 changed files with 702 additions and 0 deletions
@@ -0,0 +1,261 @@
+"""数据归档脚本(Phase 4 Task 3): 租户库老会话 导出→清理→VACUUM。
+
+conversations.created_at 早于 N 天(默认 365)的会话整链归档: 五表行先按
+会话分组导出 JSON 到 data/archives/<tenant>/<日期>-<N>d.json, 落盘成功后
+才删库(倒 FK: lead_push_logs → evaluations → customer_profiles →
+messages → conversations, 单事务), 删了行才 VACUUM。导出/删除集同源于
+同一截止串, 无漂移。
+
+纪律:
+- 门禁同 import_legacy: 租户须已在 platform.db bootstrap, 否则报错提示
+  先跑 bootstrap, 绝不摸租户库;
+- 截止时刻 Python 侧算好 ISO 串再入 SQL(不依赖 SQLite datetime() 解析):
+  ORM 行存 'YYYY-MM-DD HH:MM:SS.ffffff'(无 tz 后缀), 老库导入行可能带
+  '+00:00' 后缀——同族字符串字典序比较对两者皆单调(差异位在 '.'/'+',
+  仅整秒边界有微差, 天级截止无实际影响);
+- 导出先落盘后删除: 文件写失败 → 数据一字节不动(多一份孤儿导出无害,
+  反之是不可逆丢失); 目录懒建(无可归档/失败时不产生任何文件);
+- --dry-run 只列不动: 库零变化、不导出、不 VACUUM;
+- archives 目录 = tenants 目录的兄弟 data/archives(同 registry
+  uploads_root 的派生纪律: 生产 <repo>/tenants → <repo>/data/archives,
+  测试注入 tmp tenants → tmp 内隔离; data/ 在 .gitignore 内天然覆盖)。
+
+用法(Windows 须 PYTHONUTF8=1, 输出含中文):
+  PYTHONUTF8=1 python scripts/archive.py --tenant <slug> \
+      [--days 365] [--dry-run] [--platform-db <路径>] [--tenants-dir <目录>]
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import sqlite3
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+# 仓库根入 sys.path(照 import_legacy.py 锚定纪律: 脚本可从任意 CWD 跑)
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+import aiosqlite
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+
+# 导出分组顺序(父先子后, 与 import_legacy 的 TABLE_ORDER 一致)
+TABLE_ORDER = (
+    "conversations", "messages", "customer_profiles", "evaluations",
+    "lead_push_logs",
+)
+# 删除序 = 倒 FK(子先父后; SQLite 默认不开 FK 强制, 显式按序删)
+DELETE_ORDER = tuple(reversed(TABLE_ORDER))
+CHILD_TABLES = TABLE_ORDER[1:]            # 以 conversation_id 关联的子表
+
+# 各表确定性排序(导出顺序稳定; 注意 customer_profiles 无 created_at
+# 只有 updated_at, evaluations 是 evaluated_at——五表时间列不齐名)
+TABLE_ORDER_BY = {
+    "conversations": "created_at, id",
+    "messages": "created_at, id",
+    "customer_profiles": "updated_at, id",
+    "evaluations": "evaluated_at, id",
+    "lead_push_logs": "created_at, id",
+}
+
+
+def _tenant_bootstrapped(platform_db: Path, slug: str) -> None:
+    """门禁(同 import_legacy): 租户须已在 platform.db 注册;
+    只读打开(先看后动, 门禁绝不写 platform.db)。"""
+    if not platform_db.is_file():
+        raise SystemExit(
+            f"platform.db 不存在: {platform_db}\n"
+            f"租户须先 bootstrap: 先跑 scripts/bootstrap_platform.py")
+    try:
+        con = sqlite3.connect(
+            platform_db.resolve().as_uri() + "?mode=ro", uri=True)
+        try:
+            row = con.execute(
+                "SELECT 1 FROM tenants WHERE slug = ?", (slug,)).fetchone()
+        finally:
+            con.close()
+    except sqlite3.Error as exc:
+        raise SystemExit(
+            f"platform.db 读取失败({exc}): 先跑 scripts/bootstrap_platform.py"
+        ) from exc
+    if row is None:
+        raise SystemExit(
+            f"租户 {slug} 未在 platform.db 注册: "
+            f"先跑 scripts/bootstrap_platform.py 导入该租户档案")
+
+
+def _cutoff(days: int) -> str:
+    """截止 ISO 串(Python 侧算, 见模块 docstring 的时间形态论证)。"""
+    cut = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=days)
+    return cut.strftime("%Y-%m-%d %H:%M:%S.%f")
+
+
+async def _columns(conn, table: str) -> list[str]:
+    cur = await conn.execute(f"PRAGMA table_info({table})")
+    cols = [row[1] for row in await cur.fetchall()]
+    if not cols:
+        raise SystemExit(
+            f"租户库缺表: {table}(schema 未就位, 先跑应用或 import_legacy)")
+    return cols
+
+
+async def _scalar(conn, sql: str, params=()):
+    cur = await conn.execute(sql, params)
+    row = await cur.fetchone()
+    return row[0]
+
+
+async def _old_conversations(conn, cutoff: str) -> list:
+    """待归档会话 (id, session_id, created_at), 时间序。"""
+    cur = await conn.execute(
+        "SELECT id, session_id, created_at FROM conversations "
+        "WHERE created_at < ? ORDER BY created_at, id", (cutoff,))
+    return list(await cur.fetchall())
+
+
+async def _export_document(conn, cutoff: str) -> dict:
+    """按会话分组的五表完整行(conversations 键=会话行; 子表键=会话 id →
+    行列表; 无该子表行的会话无键)。SELECT * 全列, 值原样读出(JSON 列即
+    库内存储文本, 字节级保真)。"""
+    conn.row_factory = aiosqlite.Row
+    doc: dict = {t: {} for t in TABLE_ORDER}
+    for table in TABLE_ORDER:
+        where_id = table == "conversations"
+        sql = (f"SELECT * FROM {table} WHERE "
+               f"{'id' if where_id else 'conversation_id'} IN "
+               f"(SELECT id FROM conversations WHERE created_at < ?) "
+               f"ORDER BY {TABLE_ORDER_BY[table]}")
+        cur = await conn.execute(sql, (cutoff,))
+        for row in await cur.fetchall():
+            fields = dict(row)
+            if where_id:
+                doc[table][row["id"]] = fields
+            else:
+                doc[table].setdefault(row["conversation_id"],
+                                      []).append(fields)
+    return doc
+
+
+def _write_export(path: Path, doc: dict) -> None:
+    """先落盘后删库的'落盘'端: 目录懒建 + 原子替换(临时文件 + replace,
+    中断不留半份导出)。"""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_suffix(".json.tmp")
+    tmp.write_text(
+        json.dumps(doc, ensure_ascii=False, indent=2), encoding="utf-8")
+    tmp.replace(path)
+
+
+async def run(args) -> None:
+    days: int = args.days
+    if days < 0:
+        raise SystemExit(f"--days 不得为负: {days}")
+    platform_db = Path(args.platform_db)
+    _tenant_bootstrapped(platform_db, args.tenant)      # 门禁先于一切
+
+    tenants_dir = Path(args.tenants_dir)
+    tenant_db = tenants_dir / f"{args.tenant}.db"
+    if not tenant_db.is_file():
+        raise SystemExit(f"租户库不存在: {tenant_db}")
+    archives_dir = tenants_dir.parent / "data" / "archives"
+    cutoff = _cutoff(days)
+    size_before = tenant_db.stat().st_size
+
+    async with aiosqlite.connect(str(tenant_db), timeout=30) as conn:
+        for table in TABLE_ORDER:                       # 缺表显性报错
+            await _columns(conn, table)
+        old = await _old_conversations(conn, cutoff)
+        old_ids = [row[0] for row in old]
+
+        print(f"租户库: {tenant_db}")
+        print(f"归档策略: created_at < {cutoff}({days} 天前)")
+
+        if args.dry_run:
+            print("模式: dry-run(只列不动, 库零变化)")
+            print(f"待归档会话: {len(old_ids)} 条")
+            for cid, session_id, created_at in old:
+                print(f"  {cid} session={session_id} "
+                      f"created_at={created_at}")
+            if old_ids:
+                print("将删除的各表行数(未动):")
+                for table in ("conversations", *CHILD_TABLES):
+                    n = len(old_ids) if table == "conversations" else \
+                        await _scalar(conn, (
+                            f"SELECT COUNT(*) FROM {table} "
+                            "WHERE conversation_id IN (SELECT id FROM "
+                            "conversations WHERE created_at < ?)"), (cutoff,))
+                    print(f"  {table}: {n}")
+            print("dry-run 结论: 零改动(未导出/未删除/未 VACUUM)")
+            return
+
+        export_path = archives_dir / args.tenant / (
+            f"{datetime.now(timezone.utc).strftime('%Y%m%d')}-{days}d.json")
+        deleted: dict[str, int] = {}
+        if not old_ids:
+            print("归档会话: 0 条")
+            print("结论: 无可归档会话, 零改动(未导出/未删除/未 VACUUM)")
+            print(f"VACUUM: 未执行(无删除行)")
+            print(f"库文件: {size_before} → {tenant_db.stat().st_size} 字节")
+            return
+
+        # ---- 导出(先落盘, 失败即中止, 数据不动) ---------------------------
+        doc = {"tenant": args.tenant,
+               "generated_at": datetime.now(timezone.utc).isoformat(),
+               "days": days, "cutoff": cutoff,
+               "conversation_count": len(old_ids)}
+        doc.update(await _export_document(conn, cutoff))
+        _write_export(export_path, doc)
+
+        # ---- 清库(倒 FK, 单事务; 子查询同源截止串) ------------------------
+        for table in DELETE_ORDER:
+            col = "id" if table == "conversations" else "conversation_id"
+            cur = await conn.execute(
+                f"DELETE FROM {table} WHERE {col} IN "
+                f"(SELECT id FROM conversations WHERE created_at < ?)",
+                (cutoff,))
+            deleted[table] = cur.rowcount
+        await conn.commit()
+
+        # ---- VACUUM(删了行才跑; 已 commit, 事务外) ------------------------
+        pages_before = await _scalar(conn, "PRAGMA page_count")
+        await conn.execute("VACUUM")
+        pages_after = await _scalar(conn, "PRAGMA page_count")
+
+        # ---- 报告 ----------------------------------------------------------
+        print(f"归档会话: {len(old_ids)} 条; 各表删除行数(倒 FK 序):")
+        for table in DELETE_ORDER:
+            print(f"  {table}: {deleted[table]}")
+        print(f"导出: {export_path}")
+        size_after = tenant_db.stat().st_size
+        print(f"库文件: {size_before} → {size_after} 字节")
+        print(f"VACUUM: 已执行(page_count {pages_before} → {pages_after})")
+        print("归档结论: 导出已落盘, 删除已提交, VACUUM 完成")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description=__doc__,
+        formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("--tenant", required=True,
+                        help="租户 slug(须已 bootstrap 注册)")
+    parser.add_argument("--days", type=int, default=365,
+                        help="归档阈值: created_at 早于 N 天前的会话"
+                             "(默认 365)")
+    parser.add_argument("--dry-run", action="store_true",
+                        help="只列不动: 不导出/不删除/不 VACUUM, 库零变化")
+    parser.add_argument("--platform-db",
+                        default=str(REPO_ROOT / "data" / "platform.db"),
+                        help="platform.db 路径(默认 <repo>/data/, 门禁查"
+                             "租户是否已 bootstrap)")
+    parser.add_argument("--tenants-dir",
+                        default=str(REPO_ROOT / "tenants"),
+                        help="租户库目录(默认 <repo>/tenants; 导出落其"
+                             "兄弟目录 data/archives/<tenant>/)")
+    args = parser.parse_args()
+    asyncio.run(run(args))
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,441 @@
+"""scripts/archive.py(Phase 4 Task 3): 租户库老会话归档(导出→清理→VACUUM)。
+
+契约(计划原文):
+1. >N 天会话(含子表行)→ 导出 JSON 到 data/archives/<tenant>/ → 清库 →
+   五表清空(旧会话相关); <N 天不动;
+2. --dry-run 只列不动: 库零变化、不导出、不 VACUUM;
+3. 清理后 VACUUM → page_count 减少(PRAGMA page_count 断言, 比文件大小稳);
+4. 导出 JSON 含被删行的完整字段(五表按会话分组);
+5. 门禁同 import_legacy: 租户须已 bootstrap(platform.db 有 slug)。
+
+测试方式(照 test_import_legacy.py): import + sys.argv 注入, 非 subprocess;
+夹具全在 tmp(archives 目录取 tenants 目录兄弟 data/ 派生, 同 registry
+uploads_root 的布局纪律, tmp 内自动隔离)。时间形态双覆盖: ORM 行存
+'...SS.ffffff'(无 tz 后缀), 另造 '+00:00' 后缀的老库行(导入保真形态),
+字典序截止串比较两者皆单调。
+"""
+import hashlib
+import importlib.util
+import json
+import sqlite3
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from app.engine.models import (
+    Base,
+    Conversation,
+    ConversationSource,
+    ConversationStatus,
+    CustomerProfile,
+    Evaluation,
+    IntentLevel,
+    LeadPushLog,
+    Message,
+    MessageRole,
+)
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+SEED_DIR = REPO_ROOT / "seed_profiles"
+FIVE_TABLES = (
+    "conversations", "messages", "customer_profiles", "evaluations",
+    "lead_push_logs",
+)
+# 子表 → conversation_id 列(清库/保留断言遍历用)
+CONV_ID_COL = {
+    "conversations": "id", "messages": "conversation_id",
+    "customer_profiles": "conversation_id", "evaluations": "conversation_id",
+    "lead_push_logs": "conversation_id",
+}
+
+
+def _load(name: str):
+    spec = importlib.util.spec_from_file_location(
+        name, REPO_ROOT / "scripts" / f"{name}.py")
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+# 模块级加载一次(脚本 import 期只做 sys.path 锚定, 无副作用)
+archive = _load("archive")
+bootstrap_platform = _load("bootstrap_platform")
+
+
+def run_archive(monkeypatch, *argv):
+    monkeypatch.setattr(sys, "argv", ["archive.py", *argv])
+    archive.main()
+
+
+def run_bootstrap(monkeypatch, db: Path, *profiles: str):
+    monkeypatch.setattr(
+        sys, "argv",
+        ["bootstrap_platform.py", "--db", str(db), "--seed", str(SEED_DIR),
+         "--profiles", *profiles])
+    bootstrap_platform.main()
+
+
+# ---- 夹具/辅助 ---------------------------------------------------------------
+
+def _ro(db) -> sqlite3.Connection:
+    return sqlite3.connect(f"file:{db}?mode=ro", uri=True)
+
+
+def _count(db, table) -> int:
+    con = _ro(db)
+    try:
+        return con.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
+    finally:
+        con.close()
+
+
+def _ids(db, table, col="id") -> set:
+    con = _ro(db)
+    try:
+        return {r[0] for r in con.execute(f"SELECT {col} FROM {table}")}
+    finally:
+        con.close()
+
+
+def _full_rows(db, table) -> list[dict]:
+    """全列行(PRAGMA 列序), 导出完整性对照的采集端。"""
+    con = _ro(db)
+    con.row_factory = sqlite3.Row
+    try:
+        return [dict(r) for r in con.execute(
+            f"SELECT * FROM {table} ORDER BY id")]
+    finally:
+        con.close()
+
+
+def _page_count(db) -> int:
+    con = _ro(db)
+    try:
+        return con.execute("PRAGMA page_count").fetchone()[0]
+    finally:
+        con.close()
+
+
+def _sha256(path) -> str:
+    return hashlib.sha256(Path(path).read_bytes()).hexdigest()
+
+
+# ---- 数据形态 ----------------------------------------------------------------
+
+OLD1 = "11111111-1111-1111-1111-111111111111"   # 2020-01-01(ORM 形态)
+OLD2 = "22222222-2222-2222-2222-222222222222"   # 2020-06-15(+00:00 后缀)
+MID = "44444444-4444-4444-4444-444444444444"    # now-40d(365 天内, 30 天外)
+REC1 = "33333333-3333-3333-3333-333333333333"   # now-1d
+E1 = "eeeeeeee-0000-0000-0000-000000000001"
+E2 = "eeeeeeee-0000-0000-0000-000000000002"
+E3 = "eeeeeeee-0000-0000-0000-000000000003"
+M1 = "aaaa0000-0000-0000-0000-000000000001"
+M2 = "aaaa0000-0000-0000-0000-000000000002"
+M3 = "aaaa0000-0000-0000-0000-000000000003"
+M4 = "aaaa0000-0000-0000-0000-000000000004"
+P1 = "bbbb0000-0000-0000-0000-000000000001"
+P2 = "bbbb0000-0000-0000-0000-000000000002"
+L1 = "dddd0000-0000-0000-0000-000000000001"
+L2 = "dddd0000-0000-0000-0000-000000000002"
+L3 = "dddd0000-0000-0000-0000-000000000003"
+L4 = "dddd0000-0000-0000-0000-000000000004"
+
+NOW = datetime.now(timezone.utc)
+
+
+def _seed(db: Path) -> None:
+    """4 会话: OLD1(全子表)+ OLD2(无 profile, +00:00 时间形态)+ MID(裸,
+    40 天)+ REC1(now-1d, 全子表)。"""
+    engine = create_engine(f"sqlite:///{db}")
+    Base.metadata.create_all(engine)
+    with Session(engine) as s:
+        s.add(Conversation(id=OLD1, session_id="sess-old1",
+                           status=ConversationStatus.completed,
+                           source=ConversationSource.widget,
+                           client_info={"ip": "1.2.3.4"},
+                           created_at=datetime(2020, 1, 1, 10, tzinfo=timezone.utc),
+                           updated_at=datetime(2020, 1, 1, 10, 5,
+                                               tzinfo=timezone.utc)))
+        s.add(Conversation(id=MID, session_id="sess-mid",
+                           status=ConversationStatus.active,
+                           source=ConversationSource.standalone,
+                           created_at=NOW - timedelta(days=40)))
+        s.add(Conversation(id=REC1, session_id="sess-rec1",
+                           status=ConversationStatus.active,
+                           source=ConversationSource.widget,
+                           created_at=NOW - timedelta(days=1)))
+        s.add_all([
+            Message(id=M1, conversation_id=OLD1, role=MessageRole.user,
+                    content="old1 问价",
+                    created_at=datetime(2020, 1, 1, 10, 1, tzinfo=timezone.utc)),
+            Message(id=M2, conversation_id=OLD1, role=MessageRole.assistant,
+                    content="old1 答",
+                    created_at=datetime(2020, 1, 1, 10, 2, tzinfo=timezone.utc)),
+            Message(id=M3, conversation_id=OLD2, role=MessageRole.user,
+                    content="old2 留了电话",
+                    created_at=datetime(2020, 6, 15, 10, 1, tzinfo=timezone.utc)),
+            Message(id=M4, conversation_id=REC1, role=MessageRole.user,
+                    content="recent 消息", created_at=NOW - timedelta(days=1)),
+        ])
+        s.add(CustomerProfile(id=P1, conversation_id=OLD1,
+                              raw_info={"name": "张三"}, summary="老客",
+                              updated_at=datetime(2020, 1, 1, 10, 3,
+                                                  tzinfo=timezone.utc)))
+        s.add(CustomerProfile(id=P2, conversation_id=REC1, raw_info={},
+                              summary="新客", updated_at=NOW))
+        s.add_all([
+            Evaluation(id=E1, conversation_id=OLD1,
+                       intent_level=IntentLevel.interested, reasoning="明确",
+                       contact_info={"phone": "13800000000"}, email_sent=True,
+                       correction_count=1,
+                       evaluated_at=datetime(2020, 1, 1, 10, 4,
+                                             tzinfo=timezone.utc)),
+            Evaluation(id=E2, conversation_id=OLD2,
+                       intent_level=IntentLevel.pending, reasoning="",
+                       contact_info={},
+                       evaluated_at=datetime(2020, 6, 15, 10, 2,
+                                             tzinfo=timezone.utc)),
+            Evaluation(id=E3, conversation_id=REC1,
+                       intent_level=IntentLevel.interested, reasoning="热",
+                       contact_info={"email": "a@b.c"}, email_sent=True),
+        ])
+        s.add_all([
+            LeadPushLog(id=L1, evaluation_id=E1, conversation_id=OLD1,
+                        push_type="lead", status="success", attempts=1,
+                        response_status=200, outcome="accepted",
+                        created_at=datetime(2020, 1, 1, 10, 5,
+                                            tzinfo=timezone.utc)),
+            LeadPushLog(id=L2, evaluation_id=E1, conversation_id=OLD1,
+                        push_type="correction", status="failed", attempts=3,
+                        error_message="HTTP 500",
+                        created_at=datetime(2020, 1, 1, 10, 6,
+                                            tzinfo=timezone.utc)),
+            LeadPushLog(id=L3, evaluation_id=E2, conversation_id=OLD2,
+                        push_type="returning", status="pending", attempts=0),
+            LeadPushLog(id=L4, evaluation_id=E3, conversation_id=REC1,
+                        push_type="lead", status="success", attempts=1,
+                        response_status=200, outcome="accepted"),
+        ])
+        s.commit()
+    engine.dispose()
+
+    # OLD2: 裸 SQL 插入, created_at 带 +00:00 后缀(老库导入保真的时间形态)
+    con = sqlite3.connect(db)
+    con.execute(
+        "INSERT INTO conversations VALUES "
+        f"('{OLD2}', 'sess-old2', NULL, 'completed', 'widget', NULL, "
+        "'2020-06-15 10:00:00+00:00', '2020-06-15 10:00:00+00:00')")
+    con.commit()
+    con.close()
+
+
+# 归档后各表期望保留(默认 365 天: OLD1/OLD2 走, MID/REC1 留)——键取
+# CONV_ID_COL 列的取值集: conversations 看会话 id, 子表看 conversation_id
+KEPT = {
+    "conversations": {MID, REC1}, "messages": {REC1},
+    "customer_profiles": {REC1}, "evaluations": {REC1},
+    "lead_push_logs": {REC1},
+}
+
+
+class Env:
+    """tmp 演练环境: platform.db(bootstrap 过 jiancai)+ tenants 目录;
+    archives 目录取 tenants 兄弟 data/archives(tmp 内隔离)。"""
+
+    def __init__(self, tmp_path: Path, monkeypatch):
+        self.tmp_path = tmp_path
+        self.platform_db = tmp_path / "platform.db"
+        self.tenants_dir = tmp_path / "tenants"
+        self.tenants_dir.mkdir(parents=True)   # sqlite 不建父目录, 种子先建
+        run_bootstrap(monkeypatch, self.platform_db, "jiancai")
+
+    def archive_args(self, *extra: str, tenant: str = "jiancai"):
+        return ("--tenant", tenant,
+                "--platform-db", str(self.platform_db),
+                "--tenants-dir", str(self.tenants_dir), *extra)
+
+    @property
+    def tenant_db(self) -> Path:
+        return self.tenants_dir / "jiancai.db"
+
+    @property
+    def archives_dir(self) -> Path:
+        return self.tmp_path / "data" / "archives"
+
+    def export_path(self, days: int) -> Path:
+        date = datetime.now(timezone.utc).strftime("%Y%m%d")
+        return self.archives_dir / "jiancai" / f"{date}-{days}d.json"
+
+
+# ---- 1. 归档: 旧会话五表清空, 近期不动 ------------------------------------------
+
+def test_archive_old_conversations(tmp_path, monkeypatch, capsys):
+    env = Env(tmp_path, monkeypatch)
+    _seed(env.tenant_db)
+
+    run_archive(monkeypatch, *env.archive_args())     # 默认 --days 365
+    out = capsys.readouterr().out
+
+    assert "归档会话: 2 条" in out
+    for table in FIVE_TABLES:                 # 五表: 旧的清空, 近期原样
+        assert _ids(env.tenant_db, table, CONV_ID_COL[table]) == KEPT[table], \
+            table
+    assert {OLD1, OLD2} & _ids(env.tenant_db, "conversations") == set()
+    assert env.export_path(365).is_file()     # 导出落 data/archives/<tenant>/
+    assert "导出: " in out
+
+
+# ---- 2. dry-run: 只列不动 ------------------------------------------------------
+
+def test_dry_run_lists_without_touching(tmp_path, monkeypatch, capsys):
+    env = Env(tmp_path, monkeypatch)
+    _seed(env.tenant_db)
+    snapshot = {t: _full_rows(env.tenant_db, t) for t in FIVE_TABLES}
+
+    run_archive(monkeypatch, *env.archive_args("--days", "30", "--dry-run"))
+    out = capsys.readouterr().out
+
+    assert "dry-run" in out
+    assert "待归档会话: 3 条" in out           # OLD1/OLD2/MID(40d > 30d)
+    for cid in (OLD1, OLD2, MID):
+        assert cid in out, "dry-run 应列出待归档会话 id"
+    assert {t: _full_rows(env.tenant_db, t) for t in FIVE_TABLES} == snapshot
+    assert not env.archives_dir.exists()      # 不导出 → 目录懒建不发生
+    assert "VACUUM: 已执行" not in out
+    assert "零改动" in out
+
+
+# ---- 3. VACUUM 收缩 ------------------------------------------------------------
+
+def test_vacuum_shrinks(tmp_path, monkeypatch, capsys):
+    env = Env(tmp_path, monkeypatch)
+    engine = create_engine(f"sqlite:///{env.tenant_db}")
+    Base.metadata.create_all(engine)
+    with Session(engine) as s:                # 150 会话×4 消息, 均为 2020 年
+        s.add_all([
+            Conversation(
+                id=f"cccc0000-0000-0000-0000-{i:012d}",
+                session_id=f"sess-{i}", status=ConversationStatus.completed,
+                source=ConversationSource.widget,
+                created_at=datetime(2020, 1, 1, tzinfo=timezone.utc))
+            for i in range(150)])
+        s.add_all([
+            Message(id=f"aaaa0000-0000-0000-{c:04d}-{m:08d}",
+                    conversation_id=f"cccc0000-0000-0000-0000-{c:012d}",
+                    role=MessageRole.user, content="价" * 600,
+                    created_at=datetime(2020, 1, 1, 10, 0, tzinfo=timezone.utc))
+            for c in range(150) for m in range(4)])
+        s.commit()
+    engine.dispose()
+
+    pages_before = _page_count(env.tenant_db)
+    size_before = env.tenant_db.stat().st_size
+
+    run_archive(monkeypatch, *env.archive_args())
+    out = capsys.readouterr().out
+
+    assert "归档会话: 150 条" in out
+    assert "VACUUM: 已执行" in out
+    assert _count(env.tenant_db, "conversations") == 0
+    assert _page_count(env.tenant_db) < pages_before   # 页数收缩
+    assert env.tenant_db.stat().st_size < size_before  # 文件也随之变小
+
+
+# ---- 4. 导出完整性: 被删行完整字段, 五表按会话分组 -------------------------------
+
+def test_export_complete(tmp_path, monkeypatch, capsys):
+    env = Env(tmp_path, monkeypatch)
+    _seed(env.tenant_db)
+    before = {t: _full_rows(env.tenant_db, t) for t in FIVE_TABLES}
+
+    run_archive(monkeypatch, *env.archive_args())
+    capsys.readouterr()
+    path = env.export_path(365)
+
+    doc = json.loads(path.read_text(encoding="utf-8"))
+    assert doc["tenant"] == "jiancai"
+    assert doc["days"] == 365
+    assert doc["cutoff"]
+
+    def rows(table, **flt):
+        return sorted((r for r in before[table] if
+                       all(r[k] == v for k, v in flt.items())),
+                      key=lambda r: r["id"])
+
+    by_id = lambda table, i: next(r for r in before[table] if r["id"] == i)
+
+    # 完整字段(全列 dict 逐字相等, 含 JSON 列原文/NULL/整型布尔)
+    assert doc["conversations"][OLD1] == by_id("conversations", OLD1)
+    assert doc["conversations"][OLD2]["created_at"] == \
+        "2020-06-15 10:00:00+00:00"        # +00:00 形态原样保真
+    assert sorted(doc["messages"][OLD1], key=lambda r: r["id"]) == \
+        rows("messages", conversation_id=OLD1)
+    assert doc["messages"][OLD2] == rows("messages", conversation_id=OLD2)
+    assert doc["customer_profiles"][OLD1] == \
+        rows("customer_profiles", conversation_id=OLD1)
+    assert OLD2 not in doc["customer_profiles"]   # 无 profile 的会话无键
+    assert doc["evaluations"][OLD2] == \
+        rows("evaluations", conversation_id=OLD2)
+    assert sorted(doc["lead_push_logs"][OLD1], key=lambda r: r["id"]) == \
+        rows("lead_push_logs", conversation_id=OLD1)
+    # 近期会话不出现在任何分组
+    for table in FIVE_TABLES:
+        assert REC1 not in doc[table], table
+        assert MID not in doc[table], table
+
+
+# ---- 门禁与空跑 -----------------------------------------------------------------
+
+def test_unbootstrapped_tenant_refused(tmp_path, monkeypatch):
+    db = tmp_path / "tenants" / "jiancai.db"
+    db.parent.mkdir(parents=True)
+    db.write_bytes(b"")
+
+    # platform.db 不存在
+    with pytest.raises(SystemExit, match="bootstrap"):
+        run_archive(monkeypatch, "--tenant", "jiancai",
+                    "--platform-db", str(tmp_path / "nope" / "platform.db"),
+                    "--tenants-dir", str(tmp_path / "tenants"))
+
+    # platform.db 存在但只注册了 auto
+    platform_db = tmp_path / "platform.db"
+    run_bootstrap(monkeypatch, platform_db, "auto")
+    with pytest.raises(SystemExit, match="bootstrap"):
+        run_archive(monkeypatch, "--tenant", "jiancai",
+                    "--platform-db", str(platform_db),
+                    "--tenants-dir", str(tmp_path / "tenants"))
+
+
+def test_tenant_db_missing_refused(tmp_path, monkeypatch):
+    env = Env(tmp_path, monkeypatch)
+    assert not env.tenant_db.exists()
+    with pytest.raises(SystemExit, match="租户库不存在"):
+        run_archive(monkeypatch, *env.archive_args())
+    assert not env.archives_dir.exists()
+
+
+def test_no_old_conversations_noop(tmp_path, monkeypatch, capsys):
+    env = Env(tmp_path, monkeypatch)
+    engine = create_engine(f"sqlite:///{env.tenant_db}")
+    Base.metadata.create_all(engine)
+    with Session(engine) as s:                # 只有 1 天内的新会话
+        s.add(Conversation(id=REC1, session_id="sess-rec1",
+                           status=ConversationStatus.active,
+                           source=ConversationSource.widget,
+                           created_at=NOW - timedelta(days=1)))
+        s.commit()
+    engine.dispose()
+    sha = _sha256(env.tenant_db)
+
+    run_archive(monkeypatch, *env.archive_args())
+    out = capsys.readouterr().out
+
+    assert "归档会话: 0 条" in out
+    assert "VACUUM: 未执行" in out
+    assert not env.export_path(365).exists()
+    assert not env.archives_dir.exists()
+    assert _sha256(env.tenant_db) == sha      # 库文件字节级零变化