feat: 事件时间轴缩略帧 + 人物管理头像 + 人物合并硬规则校验

## 新架构：Oracle 集中计算 + NAS 代理展示

### Oracle 端 (fam-edge)
- 新增 frame_service: ffmpeg 视频抽帧 + VLM 人物定位裁剪头像（磁盘缓存）
- 新增 /api/oracle/frame: 按 video_id+ts 抽帧返回 jpeg（带 token）
- 新增 /api/oracle/avatar: 按 label 生成人物头像（VLM 定位人物 + 兜底整帧居中）
- 新增 person_identifier: 人物身份识别模块
- Gemini 适配器支持 flash/flash-lite 双模型切换，429 自动降级
- frame_service VLM 全模型 429 时进入 10 分钟熔断，避免每次请求白打配额
- 兜底头像不落缓存，配额恢复后自动重试 VLM 精确定位

### 人物合并硬规则校验（框架级修复）
- person_service: LLM 合并结果落库前加硬冲突检测
  - 性别冲突 → 绝不合并
  - 年龄档跨未成年/成年 → 绝不合并（防止把爷爷/宝宝并进同一人）
- oracle_db: upsert_person 入口剥离括号后缀（人物A（别名：人物B） → 人物A），消灭垃圾人物行
- 修复 set_canonical 丢弃 source 参数的 bug（旧代码硬编码 'manual' 导致错误合并被永久固化）
- get_events_for_label: 只提取该身份组的特征文本，头像定位更精准

### NAS 端 (fam-core)
- 新增 img_proxy: /api/proxy/frame 和 /api/proxy/avatar 代理 Oracle 图片
- app.py 注册 img_bp 蓝图
- oracle_sync / db_layer / member_manager 同步人物表

### UI 端 (fam-ui)
- 事件时间轴: 每条事件卡片加时间点缩略帧
- 人物管理: 每人卡片加头像（150x150 圆角）
- parse_persons: 剥离括号备注，与 Oracle 归一化一致
- 新增 EventItem 组件、Timeline 页改造
- Chat / ServiceStatus 页相应调整

### 数据库
- scripts/ddl.sql: 同步表结构更新
- Oracle people 表: features_json / display_uid / source 字段完善

fam-edge/tests/test_frame_service.py

@@ -1,3 +1,6 @@
+import os
+
+from fam_edge import frame_service
 from fam_edge.frame_service import _bbox_to_pixels
 
 
@@ -20,3 +23,52 @@ def test_bbox_to_pixels_full_frame():
 def test_bbox_to_pixels_zero_area():
     x1, y1, x2, y2 = _bbox_to_pixels([500, 500, 500, 500], 400, 300)
     assert (x1, y1) == (x2, y2)
+
+
+class _FakeRow(dict):
+    """支持 row['key'] 访问的假 sqlite3.Row。"""
+    def __getitem__(self, k):
+        return dict.get(self, k)
+
+
+class _FakeDb:
+    def __init__(self, local_path, event_start_time):
+        self._row = _FakeRow(local_path=local_path, event_start_time=event_start_time)
+
+    def get_video_by_id(self, video_id):
+        return self._row
+
+
+def test_extract_frame_cache_key_includes_width(tmp_path, monkeypatch):
+    """核心诉求: 同一 (video_id, ts) 不同调用方要不同分辨率（时间轴缩略图/头像/
+    人物识别裁人脸），缓存 key 不带 width 会导致后来的高分辨率请求悄悄拿到早先
+    缓存的低分辨率帧——这里验证两次不同 width 请求各自落到独立的缓存文件。"""
+    monkeypatch.setattr(frame_service, "CACHE_DIR", str(tmp_path))
+    video_path = tmp_path / "fake_video.mp4"
+    video_path.write_bytes(b"not a real video, ffmpeg call is mocked")
+    db = _FakeDb(str(video_path), "2026-08-22 10:00:00")
+
+    written_widths = []
+
+    def fake_run_ffmpeg(args, timeout=60):
+        # 把请求的 -vf scale=WIDTH:-2 记下来，往输出路径写点假数据模拟成功
+        out_path = args[-1]
+        vf = next((a for a in args if a.startswith('scale=')), '')
+        written_widths.append(vf)
+        with open(out_path, 'wb') as f:
+            f.write(b'\xff\xd8fakejpeg')
+        return True
+
+    monkeypatch.setattr(frame_service, "_run_ffmpeg", fake_run_ffmpeg)
+
+    data_small = frame_service.extract_frame(db, 42, "2026-08-22 10:00:05", width=400)
+    data_large = frame_service.extract_frame(db, 42, "2026-08-22 10:00:05", width=2880)
+
+    assert data_small is not None and data_large is not None
+    cache_files = sorted(os.listdir(tmp_path))
+    frame_caches = [f for f in cache_files if f.startswith('frame_42_5_')]
+    assert len(frame_caches) == 2, f"expected 2 distinct cache files, got {frame_caches}"
+    assert 'frame_42_5_400.jpg' in frame_caches
+    assert 'frame_42_5_2880.jpg' in frame_caches
+    # 两次都真的各自调用了 ffmpeg（第二次没有因为撞到第一次的缓存而被跳过）
+    assert len(written_widths) == 2

fam-edge/tests/test_gemini_adapter.py

@@ -108,3 +108,17 @@ def test_rotated_keys_single_key_never_errors():
     a = GeminiAdapter(_cfg())
     for _ in range(3):
         assert a._rotated_keys() == [(0, "key-primary")]
+
+
+def test_chat_timeout_defaults_short_not_shared_with_video_timeout():
+    """核心诉求: 问答是交互场景，不能沿用视频分析的 600s 超时——否则一个卡住
+    的 key/模型会让用户在聊天界面一直等，这正是"一直卡着"这个 bug 的根因。"""
+    a = GeminiAdapter(_cfg(timeout=600))
+    assert a.timeout == 600
+    assert a.chat_timeout == 20
+    assert a.chat_timeout != a.timeout
+
+
+def test_chat_timeout_configurable():
+    a = GeminiAdapter(_cfg(chat_timeout=8))
+    assert a.chat_timeout == 8

fam-edge/tests/test_oracle_db.py

@@ -181,3 +181,133 @@ def test_has_motion_in_range_local_filters_by_camera_id(tmp_path):
         [{"event_id": 1, "camera_id": 99, "event_type": 10, "start_time": 2500, "duration": 5}])
     assert db.has_motion_in_range_local(2000, 3000, camera_id=2) is False
     assert db.has_motion_in_range_local(2000, 3000, camera_id=99) is True
+
+
+# ----------------------------------------------------------------------
+# 人物对应关系表（video_id, raw_uid) -> canonical_name
+# ----------------------------------------------------------------------
+
+def _seed_video_with_events(db, filename="motion_1_1000.mp4"):
+    vid = db.ensure_video(filename, f"/tmp/{filename}", event_start_time="2026-08-22 10:00:00")
+    events = [
+        {"timestamp": "10:00:01", "description": "在客厅走动", "people": ["人物A"],
+         "person_appearances": [{"uid": "人物A", "features": {"gender": "男"}, "action": "走动"}]},
+        {"timestamp": "10:00:05", "description": "坐下", "people": ["人物A", "人物B"],
+         "person_appearances": [
+             {"uid": "人物A", "features": {"gender": "男"}, "action": "坐下"},
+             {"uid": "人物B", "features": {"gender": "女"}, "action": "站立"}]},
+    ]
+    db.mark_video_processed(vid, "摘要", events, ["人物A", "人物B"], "gemini")
+    return vid
+
+
+def test_set_identity_mapping_inserts_new_row(tmp_path):
+    db = _db(tmp_path)
+    assert db.set_identity_mapping(1, "人物A", "爷爷", source="auto_id") is True
+    assert db.get_identity_map_for_video(1) == {"人物A": "爷爷"}
+
+
+def test_set_identity_mapping_updates_existing_non_manual_row(tmp_path):
+    db = _db(tmp_path)
+    db.set_identity_mapping(1, "人物A", "爷爷", source="auto_id")
+    assert db.set_identity_mapping(1, "人物A", "爸爸", source="auto_id") is True
+    assert db.get_identity_map_for_video(1) == {"人物A": "爸爸"}
+
+
+def test_set_identity_mapping_manual_protected_from_auto_overwrite(tmp_path):
+    """核心诉求: 人工纠正过的映射不能被后续自动识别悄悄改回去。"""
+    db = _db(tmp_path)
+    db.set_identity_mapping(1, "人物A", "爸爸", source="manual")
+    changed = db.set_identity_mapping(1, "人物A", "爷爷", source="auto_id")
+    assert changed is False
+    assert db.get_identity_map_for_video(1) == {"人物A": "爸爸"}
+
+
+def test_set_identity_mapping_manual_can_override_manual(tmp_path):
+    db = _db(tmp_path)
+    db.set_identity_mapping(1, "人物A", "爸爸", source="manual")
+    changed = db.set_identity_mapping(1, "人物A", "爷爷", source="manual")
+    assert changed is True
+    assert db.get_identity_map_for_video(1) == {"人物A": "爷爷"}
+
+
+def test_set_identity_mapping_no_change_returns_false(tmp_path):
+    db = _db(tmp_path)
+    db.set_identity_mapping(1, "人物A", "爷爷", source="auto_id")
+    changed = db.set_identity_mapping(1, "人物A", "爷爷", source="auto_id")
+    assert changed is False
+
+
+def test_get_identity_map_for_video_scoped_per_video(tmp_path):
+    """核心诉求: 同一个 raw_uid 字符串在不同视频里可能是不同真人，映射必须按
+    video_id 隔离，不能串。"""
+    db = _db(tmp_path)
+    db.set_identity_mapping(1, "人物A", "爷爷", source="auto_id")
+    db.set_identity_mapping(2, "人物A", "爸爸", source="auto_id")
+    assert db.get_identity_map_for_video(1) == {"人物A": "爷爷"}
+    assert db.get_identity_map_for_video(2) == {"人物A": "爸爸"}
+
+
+def test_rewrite_event_person_names_updates_events_and_video(tmp_path):
+    db = _db(tmp_path)
+    vid = _seed_video_with_events(db)
+    db.rewrite_event_person_names(vid, {"人物A": "爷爷", "人物B": "媳妇"})
+
+    rows = db._conn.execute(
+        "SELECT person_list_json, person_appearances_json FROM events "
+        "WHERE video_id=? ORDER BY id", (vid,)).fetchall()
+    assert json.loads(rows[0]["person_list_json"]) == ["爷爷"]
+    pa0 = json.loads(rows[0]["person_appearances_json"])
+    assert pa0[0]["uid"] == "爷爷"
+    assert json.loads(rows[1]["person_list_json"]) == ["爷爷", "媳妇"]
+    pa1 = json.loads(rows[1]["person_appearances_json"])
+    assert {p["uid"] for p in pa1} == {"爷爷", "媳妇"}
+
+    vrow = db._conn.execute("SELECT people_json FROM videos WHERE id=?", (vid,)).fetchone()
+    assert set(json.loads(vrow["people_json"])) == {"爷爷", "媳妇"}
+
+
+def test_rewrite_event_person_names_noop_on_empty_map(tmp_path):
+    db = _db(tmp_path)
+    vid = _seed_video_with_events(db)
+    before = db._conn.execute(
+        "SELECT person_list_json FROM events WHERE video_id=?", (vid,)).fetchall()
+    db.rewrite_event_person_names(vid, {})
+    after = db._conn.execute(
+        "SELECT person_list_json FROM events WHERE video_id=?", (vid,)).fetchall()
+    assert [r["person_list_json"] for r in before] == [r["person_list_json"] for r in after]
+
+
+def test_correct_video_identity_end_to_end(tmp_path):
+    """核心诉求: 纠错入口应该找到当前展示名对应的映射行，改写映射 + 立即重写
+    展示数据，且标记为 manual（受保护）。"""
+    db = _db(tmp_path)
+    vid = _seed_video_with_events(db)
+    db.set_identity_mapping(vid, "人物A", "爷爷", source="auto_id")
+    db.rewrite_event_person_names(vid, {"人物A": "爷爷"})
+
+    db.correct_video_identity(vid, current_name="爷爷", new_name="爸爸")
+
+    assert db.get_identity_map_for_video(vid) == {"人物A": "爸爸"}
+    rows = db._conn.execute(
+        "SELECT person_list_json FROM events WHERE video_id=? ORDER BY id", (vid,)).fetchall()
+    assert json.loads(rows[0]["person_list_json"]) == ["爸爸"]
+    # manual 之后不能被自动识别覆盖回去
+    changed = db.set_identity_mapping(vid, "人物A", "爷爷", source="auto_id")
+    assert changed is False
+
+
+def test_correct_video_identity_without_prior_mapping_uses_current_name_as_raw_uid(tmp_path):
+    """核心诉求: 老流水线时代产出的数据从没跑过闭集识别，映射表里没有记录——
+    纠错依然要能生效，把 current_name 本身当 raw_uid 存一条新映射。"""
+    db = _db(tmp_path)
+    vid = db.ensure_video("motion_2_2000.mp4", "/tmp/x.mp4", event_start_time="2026-08-22 10:00:00")
+    events = [{"timestamp": "10:00:01", "description": "走动", "people": ["爷爷"],
+              "person_appearances": [{"uid": "爷爷", "features": {"gender": "男"}, "action": "走动"}]}]
+    db.mark_video_processed(vid, "摘要", events, ["爷爷"], "gemini")
+
+    db.correct_video_identity(vid, current_name="爷爷", new_name="爸爸")
+    assert db.get_identity_map_for_video(vid) == {"爷爷": "爸爸"}
+    rows = db._conn.execute(
+        "SELECT person_list_json FROM events WHERE video_id=?", (vid,)).fetchall()
+    assert json.loads(rows[0]["person_list_json"]) == ["爸爸"]

fam-edge/tests/test_person_identifier.py

@@ -0,0 +1,347 @@
+import os
+
+import pytest
+
+from fam_edge.person_identifier import PersonIdentifier
+
+
+def _cfg(ref_dir, **overrides):
+    base = {
+        "enabled": True,
+        "ref_dir": ref_dir,
+        "max_ref_per_person": 6,
+        "min_call_interval_sec": 0,  # 测试不需要真实限速，避免拖慢用例
+        "nvidia": {"api_key": "nvkey", "model_name": "nvidia/test", "timeout": 30,
+                   "max_retries": 2, "retry_backoff_sec": 0.01},
+        "gemini": {"api_key": "gkey1", "model_name": "gemini-flash-lite-latest", "timeout": 30,
+                  "max_retries": 2, "retry_backoff_sec": 0.01},
+    }
+    base.update(overrides)
+    return base
+
+
+def _write_refs(tmp_path, grandpa=2, dad=2):
+    for person, n in (("爷爷", grandpa), ("爸爸", dad)):
+        d = tmp_path / person
+        d.mkdir(parents=True, exist_ok=True)
+        for i in range(n):
+            (d / f"{i:02d}.jpg").write_bytes(b"fakejpegbytes")
+
+
+class _FakeResp:
+    def __init__(self, status_code=200, payload=None):
+        self.status_code = status_code
+        self._payload = payload or {}
+
+    def json(self):
+        return self._payload
+
+
+@pytest.fixture(autouse=True)
+def _no_real_sleep(monkeypatch):
+    """全部用例都不需要真的睡（限速/退避都测计数和结果，不测真实耗时）。"""
+    monkeypatch.setattr("fam_edge.person_identifier.time.sleep", lambda s: None)
+
+
+def test_has_references_false_when_dirs_missing(tmp_path):
+    pi = PersonIdentifier(_cfg(str(tmp_path / "refs")))
+    assert pi.has_references() is False
+
+
+def test_has_references_true_when_both_present(tmp_path):
+    _write_refs(tmp_path, grandpa=3, dad=2)
+    pi = PersonIdentifier(_cfg(str(tmp_path)))
+    assert pi.has_references() is True
+
+
+def test_has_references_false_when_only_one_person_has_refs(tmp_path):
+    (tmp_path / "爷爷").mkdir(parents=True)
+    (tmp_path / "爷爷" / "01.jpg").write_bytes(b"x")
+    pi = PersonIdentifier(_cfg(str(tmp_path)))
+    assert pi.has_references() is False
+
+
+def test_extract_json_person_clean_json():
+    pi = PersonIdentifier(_cfg("/nonexistent"))
+    assert pi._extract_json_person('{"person":"爷爷"}') == '爷爷'
+    assert pi._extract_json_person('{"person":"爸爸"}') == '爸爸'
+
+
+def test_extract_json_person_no_match_returns_none():
+    pi = PersonIdentifier(_cfg("/nonexistent"))
+    assert pi._extract_json_person('不知道是谁') is None
+    assert pi._extract_json_person('') is None
+
+
+def test_extract_json_person_both_mentioned_uses_json_field():
+    """核心诉求: 模型有时会把参考图说明也复述一遍，回复里两个名字都出现——
+    这时候不能瞎猜，要从 JSON 的 person 字段里精确取，取不到就返回 None。"""
+    pi = PersonIdentifier(_cfg("/nonexistent"))
+    text = '参考图1是爷爷，参考图5是爸爸。{"person":"爸爸"}'
+    assert pi._extract_json_person(text) == '爸爸'
+
+
+def test_extract_json_person_both_mentioned_no_json_field_returns_none():
+    pi = PersonIdentifier(_cfg("/nonexistent"))
+    text = '这个人可能是爷爷，也可能是爸爸，不太确定'
+    assert pi._extract_json_person(text) is None
+
+
+def test_classify_returns_none_when_disabled(tmp_path):
+    _write_refs(tmp_path)
+    pi = PersonIdentifier(_cfg(str(tmp_path), enabled=False))
+    assert pi.classify_adult_male(b"crop") is None
+
+
+def test_classify_returns_none_when_no_references(tmp_path):
+    pi = PersonIdentifier(_cfg(str(tmp_path / "empty")))
+    assert pi.classify_adult_male(b"crop") is None
+
+
+def test_classify_falls_back_to_gemini_when_nvidia_unavailable(tmp_path, monkeypatch):
+    """openai SDK 未安装时 NVIDIA 路径应该静默跳过（不报错），落到 Gemini。"""
+    _write_refs(tmp_path)
+    monkeypatch.setattr("fam_edge.person_identifier.OpenAI", None)
+
+    def fake_post(url, json=None, timeout=None):
+        return _FakeResp(200, {
+            "candidates": [{"content": {"parts": [{"text": '{"person":"爸爸"}'}]}}]
+        })
+    monkeypatch.setattr("fam_edge.person_identifier.requests.post", fake_post)
+    pi = PersonIdentifier(_cfg(str(tmp_path)))
+    assert pi.classify_adult_male(b"crop") == '爸爸'
+
+
+def _fake_openai_factory(reply_text=None, exc=None, fail_times=0):
+    """构造一个假 OpenAI 客户端：先失败 fail_times 次再成功，或者一直抛 exc。"""
+    state = {"calls": 0}
+
+    class FakeMessage:
+        content = reply_text
+
+    class FakeChoice:
+        message = FakeMessage()
+
+    class FakeChatResp:
+        choices = [FakeChoice()]
+
+    class FakeCompletions:
+        def create(self, **kwargs):
+            state["calls"] += 1
+            if state["calls"] <= fail_times:
+                raise (exc or RuntimeError("boom"))
+            if exc and fail_times == 0:
+                raise exc
+            return FakeChatResp()
+
+    class FakeChat:
+        completions = FakeCompletions()
+
+    class FakeOpenAI:
+        def __init__(self, base_url=None, api_key=None):
+            pass
+        chat = FakeChat()
+
+    return FakeOpenAI, state
+
+
+def test_classify_nvidia_success_skips_gemini(tmp_path, monkeypatch):
+    _write_refs(tmp_path)
+    FakeOpenAI, state = _fake_openai_factory(reply_text='{"person":"爷爷"}')
+    monkeypatch.setattr("fam_edge.person_identifier.OpenAI", FakeOpenAI)
+
+    gemini_called = {"n": 0}
+    def fake_post(url, json=None, timeout=None):
+        gemini_called["n"] += 1
+        return _FakeResp(200, {})
+    monkeypatch.setattr("fam_edge.person_identifier.requests.post", fake_post)
+
+    pi = PersonIdentifier(_cfg(str(tmp_path)))
+    assert pi.classify_adult_male(b"crop") == '爷爷'
+    assert gemini_called["n"] == 0
+    assert state["calls"] == 1
+
+
+class _FakeHTTPError(Exception):
+    def __init__(self, status_code):
+        self.response = type("R", (), {"status_code": status_code})()
+
+
+def test_nvidia_retries_transient_error_then_succeeds(tmp_path, monkeypatch):
+    """核心诉求: 429/503 这类瞬时故障要退避重试，不是第一次失败就放弃换 provider。"""
+    _write_refs(tmp_path)
+    FakeOpenAI, state = _fake_openai_factory(
+        reply_text='{"person":"爸爸"}', exc=_FakeHTTPError(503), fail_times=1)
+    monkeypatch.setattr("fam_edge.person_identifier.OpenAI", FakeOpenAI)
+    pi = PersonIdentifier(_cfg(str(tmp_path)))
+    assert pi.classify_adult_male(b"crop") == '爸爸'
+    assert state["calls"] == 2  # 第一次 503 失败重试一次后成功
+
+
+def test_nvidia_gives_up_after_max_retries_falls_back_to_gemini(tmp_path, monkeypatch):
+    _write_refs(tmp_path)
+    FakeOpenAI, state = _fake_openai_factory(exc=_FakeHTTPError(503), fail_times=99)
+    monkeypatch.setattr("fam_edge.person_identifier.OpenAI", FakeOpenAI)
+
+    def fake_post(url, json=None, timeout=None):
+        return _FakeResp(200, {
+            "candidates": [{"content": {"parts": [{"text": '{"person":"汤圆"}'}]}}]
+        })
+    # 用一个不属于爷爷/爸爸的返回值只是为了确认真的调用到了 gemini 分支
+    monkeypatch.setattr("fam_edge.person_identifier.requests.post", fake_post)
+
+    cfg = _cfg(str(tmp_path))
+    pi = PersonIdentifier(cfg)
+    pi.classify_adult_male(b"crop")
+    assert state["calls"] == pi.nvidia_max_retries  # 重试到上限就放弃，不会无限重试
+
+
+def test_nvidia_non_retryable_error_gives_up_immediately(tmp_path, monkeypatch):
+    """核心诉求: 400 参数错误这类非瞬时故障，重试没有意义，应该立刻换下一个模型/provider，
+    不要浪费时间重试一个注定失败的请求。"""
+    _write_refs(tmp_path)
+    FakeOpenAI, state = _fake_openai_factory(exc=_FakeHTTPError(400), fail_times=99)
+    monkeypatch.setattr("fam_edge.person_identifier.OpenAI", FakeOpenAI)
+    monkeypatch.setattr("fam_edge.person_identifier.requests.post",
+                        lambda *a, **k: _FakeResp(500, {"error": "down"}))
+    pi = PersonIdentifier(_cfg(str(tmp_path)))
+    pi.classify_adult_male(b"crop")
+    assert state["calls"] == 1  # 400 不重试，一次就放弃这个模型
+
+
+def test_nvidia_falls_through_model_chain(tmp_path, monkeypatch):
+    """核心诉求: 第一个模型重试耗尽后，应该换模型链里的下一个型号再试，而不是
+    直接放弃整个 NVIDIA provider。"""
+    _write_refs(tmp_path)
+    calls = []
+
+    class FakeMessage:
+        def __init__(self, content):
+            self.content = content
+
+    class FakeChoice:
+        def __init__(self, content):
+            self.message = FakeMessage(content)
+
+    class FakeChatResp:
+        def __init__(self, content):
+            self.choices = [FakeChoice(content)]
+
+    class FakeCompletions:
+        def create(self, model, **kwargs):
+            calls.append(model)
+            if model == 'nvidia/model-a':
+                raise _FakeHTTPError(503)
+            return FakeChatResp('{"person":"爷爷"}')
+
+    class FakeChat:
+        completions = FakeCompletions()
+
+    class FakeOpenAI:
+        def __init__(self, base_url=None, api_key=None):
+            pass
+        chat = FakeChat()
+
+    monkeypatch.setattr("fam_edge.person_identifier.OpenAI", FakeOpenAI)
+    cfg = _cfg(str(tmp_path), nvidia={
+        "api_key": "nvkey", "model_name": "nvidia/model-a",
+        "fallback_models": ["nvidia/model-b"], "timeout": 30,
+        "max_retries": 2, "retry_backoff_sec": 0.01,
+    })
+    pi = PersonIdentifier(cfg)
+    assert pi.classify_adult_male(b"crop") == '爷爷'
+    assert calls == ['nvidia/model-a', 'nvidia/model-a', 'nvidia/model-b']
+
+
+def test_gemini_retries_transient_error_on_same_key(tmp_path, monkeypatch):
+    _write_refs(tmp_path)
+    monkeypatch.setattr("fam_edge.person_identifier.OpenAI", None)
+    calls = []
+    def fake_post(url, json=None, timeout=None):
+        calls.append(url.split('key=')[-1])
+        if len(calls) == 1:
+            return _FakeResp(503, {"error": {"code": 503}})
+        return _FakeResp(200, {
+            "candidates": [{"content": {"parts": [{"text": '{"person":"媳妇"}'}]}}]
+        })
+    monkeypatch.setattr("fam_edge.person_identifier.requests.post", fake_post)
+    pi = PersonIdentifier(_cfg(str(tmp_path)))
+    pi.classify_adult_male(b"crop")
+    assert calls == ['gkey1', 'gkey1']  # 同一个 key 重试，不是立刻跳到下一个 key
+
+
+def test_classify_gemini_rotates_across_keys_after_retries_exhausted(tmp_path, monkeypatch):
+    _write_refs(tmp_path)
+    monkeypatch.setattr("fam_edge.person_identifier.OpenAI", None)
+
+    calls = []
+    def fake_post(url, json=None, timeout=None):
+        key = url.split('key=')[-1]
+        calls.append(key)
+        if key == 'gkey1':
+            return _FakeResp(429, {"error": {"code": 429}})
+        return _FakeResp(200, {
+            "candidates": [{"content": {"parts": [{"text": '{"person":"爷爷"}'}]}}]
+        })
+    monkeypatch.setattr("fam_edge.person_identifier.requests.post", fake_post)
+
+    cfg = _cfg(str(tmp_path), gemini={
+        "api_key": "gkey1", "extra_api_keys": ["gkey2"],
+        "model_name": "gemini-flash-lite-latest", "timeout": 30,
+        "max_retries": 2, "retry_backoff_sec": 0.01,
+    })
+    pi = PersonIdentifier(cfg)
+    assert pi.classify_adult_male(b"crop") == '爷爷'
+    assert calls == ['gkey1', 'gkey1', 'gkey2']  # gkey1 重试用尽才换 gkey2
+
+
+def test_classify_both_providers_fail_returns_none(tmp_path, monkeypatch):
+    """核心诉求: NVIDIA 和 Gemini 都失败时绝不能瞎猜，必须返回 None。"""
+    _write_refs(tmp_path)
+    monkeypatch.setattr("fam_edge.person_identifier.OpenAI", None)
+
+    def fake_post(url, json=None, timeout=None):
+        return _FakeResp(500, {"error": "boom"})
+    monkeypatch.setattr("fam_edge.person_identifier.requests.post", fake_post)
+    pi = PersonIdentifier(_cfg(str(tmp_path)))
+    assert pi.classify_adult_male(b"crop") is None
+
+
+def test_env_var_credentials_resolved(tmp_path):
+    os.environ["TEST_NVIDIA_KEY_XYZ"] = "realkey"
+    try:
+        cfg = _cfg(str(tmp_path), nvidia={"api_key": "${TEST_NVIDIA_KEY_XYZ}"})
+        pi = PersonIdentifier(cfg)
+        assert pi.nvidia_api_key == "realkey"
+    finally:
+        del os.environ["TEST_NVIDIA_KEY_XYZ"]
+
+
+def test_max_ref_per_person_limits_loaded_refs(tmp_path):
+    _write_refs(tmp_path, grandpa=10, dad=10)
+    pi = PersonIdentifier(_cfg(str(tmp_path), max_ref_per_person=3))
+    refs = pi._load_refs()
+    assert len(refs['爷爷']) == 3
+    assert len(refs['爸爸']) == 3
+
+
+def test_pace_sleeps_when_called_too_soon(tmp_path, monkeypatch):
+    """核心诉求: 批量回填会短时间内密集调用，min_call_interval_sec 要真的限速，
+    不能形同虚设。"""
+    _write_refs(tmp_path)
+    slept = []
+    monkeypatch.setattr("fam_edge.person_identifier.time.sleep", lambda s: slept.append(s))
+    pi = PersonIdentifier(_cfg(str(tmp_path), min_call_interval_sec=5))
+    pi._last_call_at = __import__("time").time()  # 刚刚调用过
+    pi._pace()
+    assert slept and slept[0] > 0
+
+
+def test_pace_no_sleep_when_interval_already_elapsed(tmp_path, monkeypatch):
+    _write_refs(tmp_path)
+    slept = []
+    monkeypatch.setattr("fam_edge.person_identifier.time.sleep", lambda s: slept.append(s))
+    pi = PersonIdentifier(_cfg(str(tmp_path), min_call_interval_sec=5))
+    pi._last_call_at = 0  # 很久以前
+    pi._pace()
+    assert slept == []

fam-edge/tests/test_qa.py

@@ -0,0 +1,81 @@
+from fam_edge.qa import QAOrchestrator
+
+
+class _FakeAdapter:
+    def __init__(self, provider_name, chunks=None, raises=False):
+        self.provider_name = provider_name
+        self._chunks = chunks or []
+        self._raises = raises
+
+    def chat_stream(self, prompt, max_tokens=512):
+        if self._raises:
+            raise RuntimeError("boom")
+        for c in self._chunks:
+            yield c
+
+    def chat(self, prompt, max_tokens=512):
+        return ''.join(self._chunks) or None
+
+
+def _orchestrator(adapters):
+    qa = QAOrchestrator.__new__(QAOrchestrator)  # 跳过 __init__（不需要真实 config/adapters）
+    qa.adapters = adapters
+    return qa
+
+
+def test_run_qa_stream_first_provider_success():
+    qa = _orchestrator([_FakeAdapter("gemini", chunks=["你", "好"])])
+    events = list(qa.run_qa_stream("hi"))
+    types = [e["type"] for e in events]
+    assert types == ["provider_trying", "chunk", "chunk", "done"]
+    assert events[1]["text"] == "你"
+    assert events[2]["text"] == "好"
+    assert events[-1]["provider"] == "gemini"
+
+
+def test_run_qa_stream_falls_back_when_first_yields_nothing():
+    """核心诉求: 第一个 provider 一个字都没吐出来才允许换下一个——不是失败就切，
+    是"完全没有产出"才切。"""
+    qa = _orchestrator([
+        _FakeAdapter("gemini", chunks=[]),
+        _FakeAdapter("nvidia", chunks=["答案"]),
+    ])
+    events = list(qa.run_qa_stream("hi"))
+    types = [e["type"] for e in events]
+    assert types == ["provider_trying", "provider_failed", "provider_trying", "chunk", "done"]
+    assert events[-1]["provider"] == "nvidia"
+
+
+def test_run_qa_stream_does_not_switch_after_partial_output():
+    """核心诉求: 已经开始吐字之后中途失败，不能悄悄换下一个 provider 接着写
+    （会出现两段风格/内容不连贯的回答拼在一起）——直接结束这次生成。"""
+    class _PartialThenRaise:
+        provider_name = "gemini"
+        def chat_stream(self, prompt, max_tokens=512):
+            yield "先吐"
+            raise RuntimeError("connection reset")
+
+    qa = _orchestrator([_PartialThenRaise(), _FakeAdapter("nvidia", chunks=["不该被用到"])])
+    events = list(qa.run_qa_stream("hi"))
+    types = [e["type"] for e in events]
+    assert types == ["provider_trying", "chunk", "done"]
+    assert events[1]["text"] == "先吐"
+    assert events[-1]["provider"] == "gemini"
+
+
+def test_run_qa_stream_all_providers_fail():
+    qa = _orchestrator([
+        _FakeAdapter("gemini", chunks=[]),
+        _FakeAdapter("nvidia", chunks=[], raises=True),
+    ])
+    events = list(qa.run_qa_stream("hi"))
+    assert events[-1]["type"] == "all_failed"
+    assert "provider_failed" in [e["type"] for e in events]
+
+
+def test_run_qa_stream_exception_treated_as_no_output():
+    qa = _orchestrator([_FakeAdapter("gemini", raises=True), _FakeAdapter("nvidia", chunks=["ok"])])
+    events = list(qa.run_qa_stream("hi"))
+    assert events[0] == {"type": "provider_trying", "provider": "gemini"}
+    assert events[1] == {"type": "provider_failed", "provider": "gemini"}
+    assert events[-1]["provider"] == "nvidia"