修复 History Import 失败分块在 Resume 时被跳过

src/import_memory.py

@@ -810,6 +810,7 @@ def __init__(self, state_dir: str):
             "memories_skipped": 0,
             "memories_raw": 0,
             "chunks_failed": 0,
+            "failed_chunk_indices": [],
             "errors": [],
             "status": "idle",  # idle | running | paused | completed | error
             "job_id": "",
@@ -859,6 +860,7 @@ def reset(
             "memories_skipped": 0,
             "memories_raw": 0,
             "chunks_failed": 0,
+            "failed_chunk_indices": [],
             "errors": [],
             "status": "running",
             "job_id": job_id,
@@ -1164,37 +1166,73 @@ async def start(
             self.release_start_reservation(job_id)
 
     async def _process_chunks(self, preserve_raw: bool) -> dict:
-        """从当前进度开始处理分块。"""
+        """从当前进度开始处理分块；续传时先补跑之前失败的分块。"""
         start_idx = self.state.data["processed"]
         await self._prime_exact_content_hashes()
 
-        for i in range(start_idx, len(self._chunks)):
-            if self._paused:
-                self.state.data["status"] = "paused"
-                self.state.save()
-                logger.info(f"Import paused at chunk {i}/{len(self._chunks)}")
-                return self.state.to_dict()
-
-            chunk = self._chunks[i]
+        # processed 表示主游标已经尝试到哪里；失败分块单独持久化，Resume 时
+        # 必须先补跑它们，否则单靠 processed 会从失败位置之后继续，静默漏块。
+        raw_failed_indices = self.state.data.get("failed_chunk_indices", [])
+        failed_indices: list[int] = []
+        for value in raw_failed_indices if isinstance(raw_failed_indices, list) else []:
+            if isinstance(value, int) and 0 <= value < len(self._chunks) and value not in failed_indices:
+                failed_indices.append(value)
+        self.state.data["failed_chunk_indices"] = failed_indices
+
+        async def process_chunk(index: int) -> bool:
+            chunk = self._chunks[index]
             try:
-                chunk_ok = await self._process_single_chunk(chunk, preserve_raw)
+                return await self._process_single_chunk(chunk, preserve_raw)
             except Exception as e:
-                chunk_ok = False
-                err_msg = f"分块 {i} 处理失败（{type(e).__name__}）"
+                err_msg = f"分块 {index} 处理失败（{type(e).__name__}）"
                 logger.warning(
                     "Import chunk failed: index=%s err_type=%s detail=hidden",
-                    i,
+                    index,
                     type(e).__name__,
                 )
                 if len(self.state.data["errors"]) < _STATE_ERR_LOG_MAX:
                     self.state.data["errors"].append(err_msg)
+                return False
+
+        # Resume 先重试历史失败分块。成功后解除失败标记；再次失败则保留，
+        # 但不重复增加 chunks_failed。
+        for failed_index in list(failed_indices):
+            if self._paused:
+                self.state.data["status"] = "paused"
+                self.state.save()
+                logger.info(
+                    f"Import paused before retrying failed chunk "
+                    f"{failed_index}/{len(self._chunks)}"
+                )
+                return self.state.to_dict()
+
+            chunk_ok = await process_chunk(failed_index)
+            if chunk_ok:
+                failed_indices.remove(failed_index)
+                self.state.data["failed_chunk_indices"] = failed_indices
+                self.state.data["chunks_failed"] = max(
+                    0, int(self.state.data.get("chunks_failed", 0)) - 1
+                )
+            self.state.save()
+
+        for i in range(start_idx, len(self._chunks)):
+            if self._paused:
+                self.state.data["status"] = "paused"
+                self.state.save()
+                logger.info(f"Import paused at chunk {i}/{len(self._chunks)}")
+                return self.state.to_dict()
+
+            chunk_ok = await process_chunk(i)
             if not chunk_ok:
-                self.state.data["chunks_failed"] = int(
-                    self.state.data.get("chunks_failed", 0)
-                ) + 1
+                if i not in failed_indices:
+                    failed_indices.append(i)
+                    self.state.data["failed_chunk_indices"] = failed_indices
+                    self.state.data["chunks_failed"] = int(
+                        self.state.data.get("chunks_failed", 0)
+                    ) + 1
 
             self.state.data["processed"] = i + 1
-            # 每处理一个分块就保存进度
+            # 每处理一个分块就保存进度；失败索引与主游标一起原子持久化。
             self.state.save()
 
         failed_chunks = int(self.state.data.get("chunks_failed", 0))

tests/test_import_memory_regressions.py

@@ -1,4 +1,4 @@
-"""import_memory.py 回归测试 —— 找茬会话（2026-07-15）发现的四个问题。
+"""import_memory.py 回归测试 —— 覆盖历史导入的断点续传、分块与持久化边界。
 
 1. preserve_raw 断点续传重复导入：进度只在整个 chunk 处理完才落盘，崩溃
    重启后同一个 chunk 会被重新提取一遍；preserve_raw 分支原来完全跳过去重。
@@ -7,6 +7,7 @@
 3. 单次提取正文被固定按 12000 字符截断，远小于 chunk 自身 ~10000 token
    的目标预算（对英文/混合内容而言），且不留任何痕迹。
 4. ImportState.save() 无 fsync/长路径前缀（见 utils.atomic_write_text 用法）。
+5. #137：失败 chunk 推进 processed 后，Resume 不能静默跳过未完成工作。
 """
 import hashlib
 import json
@@ -381,6 +382,124 @@ async def process(_chunk, _preserve_raw):
     assert result["chunks_failed"] == outcomes.count(False)
 
 
+@pytest.mark.asyncio
+async def test_resume_retries_failed_chunk_before_advancing(tmp_path, monkeypatch):
+    engine = ImportEngine(
+        {"buckets_dir": str(tmp_path), "human": "用户"},
+        FakeBucketManager(),
+        FakeDehydrator(),
+    )
+    engine._chunks = [
+        {"content": "chunk-0"},
+        {"content": "chunk-1"},
+    ]
+    engine.state.reset("history.txt", "hash", 2, job_id="job")
+    calls: list[str] = []
+    chunk0_attempts = 0
+
+    async def process(chunk, _preserve_raw):
+        nonlocal chunk0_attempts
+        name = chunk["content"]
+        calls.append(name)
+        if name == "chunk-0":
+            chunk0_attempts += 1
+            if chunk0_attempts == 1:
+                # 模拟 chunk 0 失败期间用户点击 Pause：本轮会在进入 chunk 1 前暂停。
+                engine._paused = True
+                return False
+        return True
+
+    monkeypatch.setattr(engine, "_process_single_chunk", process)
+
+    first = await engine._process_chunks(False)
+
+    assert first["status"] == "paused"
+    assert first["processed"] == 1
+    assert first["chunks_failed"] == 1
+    assert first["failed_chunk_indices"] == [0]
+    assert calls == ["chunk-0"]
+
+    # Resume：必须先补跑失败的 chunk 0，再从主游标继续 chunk 1。
+    engine._paused = False
+    resumed = await engine._process_chunks(False)
+
+    assert calls == ["chunk-0", "chunk-0", "chunk-1"]
+    assert resumed["processed"] == 2
+    assert resumed["chunks_failed"] == 0
+    assert resumed["failed_chunk_indices"] == []
+    assert resumed["status"] == "completed"
+
+
+@pytest.mark.asyncio
+async def test_failed_chunk_retry_does_not_double_count_failure(tmp_path, monkeypatch):
+    engine = ImportEngine(
+        {"buckets_dir": str(tmp_path), "human": "用户"},
+        FakeBucketManager(),
+        FakeDehydrator(),
+    )
+    engine._chunks = [
+        {"content": "chunk-0"},
+        {"content": "chunk-1"},
+    ]
+    engine.state.reset("history.txt", "hash", 2, job_id="job")
+    engine.state.data.update({
+        "processed": 1,
+        "chunks_failed": 1,
+        "failed_chunk_indices": [0],
+        "status": "paused",
+    })
+
+    async def process(chunk, _preserve_raw):
+        return chunk["content"] != "chunk-0"
+
+    monkeypatch.setattr(engine, "_process_single_chunk", process)
+
+    result = await engine._process_chunks(False)
+
+    assert result["processed"] == 2
+    assert result["chunks_failed"] == 1
+    assert result["failed_chunk_indices"] == [0]
+    assert result["status"] == "partial"
+
+
+@pytest.mark.asyncio
+async def test_resume_keeps_failure_state_consistent_when_later_chunk_fails(
+    tmp_path, monkeypatch
+):
+    engine = ImportEngine(
+        {"buckets_dir": str(tmp_path), "human": "用户"},
+        FakeBucketManager(),
+        FakeDehydrator(),
+    )
+    engine._chunks = [
+        {"content": "chunk-0"},
+        {"content": "chunk-1"},
+        {"content": "chunk-2"},
+    ]
+    engine.state.reset("history.txt", "hash", 3, job_id="job")
+    engine.state.data.update({
+        "processed": 1,
+        "chunks_failed": 1,
+        "failed_chunk_indices": [0],
+        "status": "paused",
+    })
+    calls: list[str] = []
+
+    async def process(chunk, _preserve_raw):
+        calls.append(chunk["content"])
+        return chunk["content"] != "chunk-1"
+
+    monkeypatch.setattr(engine, "_process_single_chunk", process)
+
+    result = await engine._process_chunks(False)
+
+    assert calls == ["chunk-0", "chunk-1", "chunk-2"]
+    assert result["processed"] == 3
+    assert result["chunks_failed"] == 1
+    assert result["failed_chunk_indices"] == [1]
+    assert result["status"] == "partial"
+
+
 @pytest.mark.asyncio
 async def test_import_preflight_failure_replaces_stale_status(tmp_path):
     dehydrator = FakeDehydrator()

update_manifest.json

@@ -77,8 +77,8 @@
     },
     {
       "path": "src/import_memory.py",
-      "sha256": "9f89c8ef32b09f0d6ee3274bba1984b8d383ee2ae338ccd4e56dd04171ab5f55",
-      "size": 60702
+      "sha256": "693b7a33db1e2d575d18a2fd1032f5dd5408637e9d958bedf8782adeb57a3956",
+      "size": 62700
     },
     {
       "path": "src/migrate_engine.py",