feat: classify OCR errors and degraded completions

paperforge/skills/paperforge/scripts/pf_deep.py

@@ -1439,7 +1439,7 @@ def prepare_deep_reading(vault: Path, zotero_key: str, force: bool = False) -> d
         meta = _read_json(meta_path)
         ocr_status = str(meta.get("ocr_status", "pending")).strip().lower()
 
-    if ocr_status != "done":
+    if ocr_status not in ("done", "done_degraded"):
         result["message"] = f"[ERROR] OCR status='{ocr_status}', not 'done'. Wait for OCR or check meta.json."
         return result
 
@@ -1531,7 +1531,7 @@ def scan_deep_reading_queue(vault: Path) -> list[dict]:
     # Sort: OCR completed first, then by domain, then by key
     queue.sort(
         key=lambda row: (
-            0 if row["ocr_status"] == "done" else 1,
+            0 if row["ocr_status"] in ("done", "done_degraded") else 1,
             row["domain"],
             row["zotero_key"],
         )
@@ -1737,8 +1737,8 @@ def main() -> int:
                 serializable.append(r)
             print(json.dumps(serializable, ensure_ascii=False, indent=2))
         else:
-            ready = [row for row in queue if row["ocr_status"] == "done"]
-            blocked = [row for row in queue if row["ocr_status"] != "done"]
+            ready = [row for row in queue if row["ocr_status"] in ("done", "done_degraded")]
+            blocked = [row for row in queue if row["ocr_status"] not in ("done", "done_degraded")]
             print(f"# 待精读队列 ({len(queue)} 篇)")
             print()
             if ready:

paperforge/worker/ocr.py

@@ -59,6 +59,14 @@ from paperforge.worker.ocr_blocks import (
     write_raw_blocks_jsonl,
     write_structured_blocks_jsonl,
 )
+from paperforge.worker.ocr_errors import (
+    OCRAPISchemaError,
+    OCRArtifactIntegrityError,
+    OCRNetworkError,
+    OCRPDFResolveError,
+    OCRPostprocessError,
+    classify_ocr_error,
+)
 from paperforge.worker.ocr_figures import (
     build_figure_inventory,
     write_figure_inventory,
@@ -79,6 +87,19 @@ from paperforge.worker.sync import (
 
 logger = logging.getLogger(__name__)
 
+OCR_QUEUE_STATUSES = {
+    "pending", "queued", "running", "done", "done_degraded",
+    "retryable_error", "fatal_error", "nopdf", "blocked",
+}
+
+
+def apply_ocr_error_state(row: dict, meta: dict, error_state: dict) -> None:
+    row["queue_status"] = error_state["status"]
+    meta["error_type"] = error_state["error_type"]
+    meta["error_stage"] = error_state["error_stage"]
+    meta["retryable"] = error_state["retryable"]
+    meta["last_error"] = error_state["last_error"]
+
 
 def ensure_ocr_meta(vault: Path, row: dict) -> dict:
     paths = pipeline_paths(vault)
@@ -2184,11 +2205,8 @@ def run_ocr(vault: Path, verbose: bool = False, no_progress: bool = False, selec
             try:
                 response = retry_with_meta(_do_poll, meta_path_poll, meta["ocr_job_id"], token)
             except Exception as e:
-                meta["ocr_status"] = "pending"
-                meta["error"] = str(e)
-                meta["last_error"] = str(e)
-                meta["retry_count"] = int(meta.get("retry_count", 0)) + 1
-                queue_row["queue_status"] = "pending"
+                error_state = classify_ocr_error(OCRNetworkError(str(e)), stage="poll")
+                apply_ocr_error_state(queue_row, meta, error_state)
                 write_json(paths["ocr"] / key / "meta.json", meta)
                 changed += 1
                 active_submitted = max(0, active_submitted - 1)
@@ -2197,13 +2215,17 @@ def run_ocr(vault: Path, verbose: bool = False, no_progress: bool = False, selec
                 payload = response.json()["data"]
                 state = payload["state"]
             except (json.JSONDecodeError, KeyError) as e:
-                meta["ocr_status"] = "pending"
-                meta["error"] = f"API schema mismatch during polling: {e}"
-                meta["last_error"] = meta["error"]
+                error_state = classify_ocr_error(
+                    OCRAPISchemaError(f"API schema mismatch during polling: {e}"), stage="poll"
+                )
+                apply_ocr_error_state(queue_row, meta, error_state)
                 meta["raw_response"] = response.text[:1000]
-                meta["retry_count"] = int(meta.get("retry_count", 0)) + 1
-                queue_row["queue_status"] = "pending"
                 write_json(paths["ocr"] / key / "meta.json", meta)
+                json_dir = paths["ocr"] / key / "json"
+                json_dir.mkdir(parents=True, exist_ok=True)
+                raw_error_dir = paths["ocr"] / key / "raw"
+                raw_error_dir.mkdir(parents=True, exist_ok=True)
+                write_json(raw_error_dir / "api_error_payload.json", {"raw_response": response.text[:5000]})
                 changed += 1
                 active_submitted = max(0, active_submitted - 1)
                 continue
@@ -2224,16 +2246,33 @@ def run_ocr(vault: Path, verbose: bool = False, no_progress: bool = False, selec
                     page_num, markdown_path, json_path, fulltext_md_path = postprocess_ocr_result(
                         vault, key, all_results
                     )
-                except Exception as e:
-                    meta["ocr_status"] = "pending"
-                    meta["error"] = str(e)
-                    meta["retry_count"] = int(meta.get("retry_count", 0)) + 1
-                    queue_row["queue_status"] = "pending"
+                except KeyboardInterrupt:
+                    raise
+                except SystemExit:
+                    raise
+                except BaseException:
+                    raise
+                except (OCRPDFResolveError, OCRArtifactIntegrityError, OCRPostprocessError) as exc:
+                    error_state = classify_ocr_error(exc, stage="postprocess")
+                    apply_ocr_error_state(queue_row, meta, error_state)
                     write_json(paths["ocr"] / key / "meta.json", meta)
                     changed += 1
                     active_submitted = max(0, active_submitted - 1)
                     continue
-                meta["ocr_status"] = "done"
+                except Exception as exc:
+                    error_state = classify_ocr_error(OCRPostprocessError(str(exc)), stage="postprocess")
+                    apply_ocr_error_state(queue_row, meta, error_state)
+                    write_json(paths["ocr"] / key / "meta.json", meta)
+                    changed += 1
+                    active_submitted = max(0, active_submitted - 1)
+                    continue
+                health_path_poll = paths["ocr"] / key / "health" / "ocr_health.json"
+                health_report_poll = read_json(health_path_poll) if health_path_poll.exists() else {}
+                if health_report_poll.get("degraded_reasons"):
+                    meta["ocr_status"] = "done_degraded"
+                    meta["degraded_reason"] = "; ".join(health_report_poll["degraded_reasons"])
+                else:
+                    meta["ocr_status"] = "done"
                 # Per D-01: auto_analyze_after_ocr opt-in workflow streamlining
                 cfg_path = vault / "paperforge.json"
                 if cfg_path.exists():
@@ -2435,7 +2474,13 @@ def run_ocr(vault: Path, verbose: bool = False, no_progress: bool = False, selec
                     page_num, md_path, json_path, fulltext_md_path = postprocess_ocr_result(vault, key, results)
                 except Exception:
                     continue
-                meta["ocr_status"] = "done"
+                health_path_loop = paths["ocr"] / key / "health" / "ocr_health.json"
+                health_report_loop = read_json(health_path_loop) if health_path_loop.exists() else {}
+                if health_report_loop.get("degraded_reasons"):
+                    meta["ocr_status"] = "done_degraded"
+                    meta["degraded_reason"] = "; ".join(health_report_loop["degraded_reasons"])
+                else:
+                    meta["ocr_status"] = "done"
                 meta["ocr_finished_at"] = datetime.now(timezone.utc).isoformat()
                 meta["page_count"] = page_num
                 meta["markdown_path"] = md_path

paperforge/worker/ocr_errors.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+
+class OCRRecoverableError(Exception):
+    pass
+
+
+class OCRFatalError(Exception):
+    pass
+
+
+class OCRNetworkError(OCRRecoverableError):
+    pass
+
+
+class OCRAPISchemaError(OCRRecoverableError):
+    pass
+
+
+class OCRPDFResolveError(OCRFatalError):
+    pass
+
+
+class OCRPostprocessError(OCRFatalError):
+    pass
+
+
+class OCRArtifactIntegrityError(OCRFatalError):
+    pass
+
+
+def classify_ocr_error(exc: Exception, *, stage: str) -> dict:
+    retryable = isinstance(exc, OCRRecoverableError)
+    return {
+        "status": "retryable_error" if retryable else "fatal_error",
+        "error_type": exc.__class__.__name__,
+        "error_stage": stage,
+        "retryable": retryable,
+        "last_error": str(exc),
+    }
+
+
+def normalize_ocr_status_for_reader(status: str) -> str:
+    if status in {"retryable_error", "fatal_error"}:
+        return "error"
+    if status == "done_degraded":
+        return "done"
+    return status

paperforge/worker/ocr_health.py

@@ -107,6 +107,16 @@ def build_ocr_health(
     }
     report.update(decision_summary)
 
+    degraded_reasons = []
+    if span.get("coverage_quality", "weak") == "weak":
+        degraded_reasons.append(f"weak span coverage ({span.get('coverage_ratio', 0):.0%})")
+    if spine.get("_meta", {}).get("quality", "weak") == "weak":
+        degraded_reasons.append("weak body spine")
+    if layout.get("status", "unknown") == "fail":
+        degraded_reasons.append(f"layout audit failed ({layout.get('anomaly_count', 0)} anomalies)")
+
+    report["degraded_reasons"] = degraded_reasons
+
     def _score_distribution(scores: list[float]) -> dict:
         return {
             "high": sum(1 for s in scores if s >= 0.75),

paperforge/worker/status.py

@@ -782,7 +782,7 @@ def run_doctor(vault: Path, verbose: bool = False, json_output: bool = False) ->
                 if not _ws_dir.exists():
                     _missing_workspace += 1
                     continue
-                if _e.get("ocr_status") == "done":
+                if _e.get("ocr_status") in ("done", "done_degraded"):
                     _ft = _ws_dir / "fulltext.md"
                     if not _ft.exists():
                         _missing_fulltext += 1
@@ -1018,7 +1018,7 @@ def run_status(vault: Path, verbose: bool = False, json_output: bool = False) ->
             ocr_failed += 1
             continue
         status = str(meta.get("ocr_status", "")).strip().lower()
-        if status == "done":
+        if status in ("done", "done_degraded"):
             ocr_done += 1
         elif status in ("queued", "running", "processing"):
             ocr_processing += 1
@@ -1038,7 +1038,7 @@ def run_status(vault: Path, verbose: bool = False, json_output: bool = False) ->
                 continue
             ocr_total += 1
             status = str(meta.get("ocr_status", "")).strip().lower()
-            if status == "done":
+            if status in ("done", "done_degraded"):
                 ocr_done += 1
             elif status in ("queued", "running", "processing"):
                 ocr_processing += 1

tests/test_ocr_errors.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+
+def test_network_error_maps_to_retryable_status() -> None:
+    from paperforge.worker.ocr_errors import OCRNetworkError, classify_ocr_error
+
+    result = classify_ocr_error(OCRNetworkError("timeout"), stage="poll")
+    assert result["status"] == "retryable_error"
+    assert result["retryable"] is True
+    assert result["error_type"] == "OCRNetworkError"
+
+
+def test_postprocess_error_maps_to_fatal_status() -> None:
+    from paperforge.worker.ocr_errors import OCRPostprocessError, classify_ocr_error
+
+    result = classify_ocr_error(OCRPostprocessError("bad role graph"), stage="postprocess")
+    assert result["status"] == "fatal_error"
+    assert result["retryable"] is False
+    assert result["error_stage"] == "postprocess"
+
+
+def test_ocr_queue_statuses_include_error_taxonomy() -> None:
+    from paperforge.worker.ocr import OCR_QUEUE_STATUSES
+
+    for status in ["pending", "queued", "running", "done", "done_degraded", "retryable_error", "fatal_error", "nopdf", "blocked"]:
+        assert status in OCR_QUEUE_STATUSES
+
+
+def test_apply_ocr_error_state_updates_queue_and_meta() -> None:
+    from paperforge.worker.ocr import apply_ocr_error_state
+
+    row = {"queue_status": "running"}
+    meta = {}
+    apply_ocr_error_state(row, meta, {"status": "fatal_error", "error_type": "OCRPostprocessError", "error_stage": "postprocess", "retryable": False, "last_error": "bad"})
+    assert row["queue_status"] == "fatal_error"
+    assert meta["error_type"] == "OCRPostprocessError"
+
+
+def test_legacy_error_status_maps_to_new_taxonomy_for_readers() -> None:
+    from paperforge.worker.ocr_errors import normalize_ocr_status_for_reader
+
+    assert normalize_ocr_status_for_reader("retryable_error") == "error"
+    assert normalize_ocr_status_for_reader("fatal_error") == "error"
+    assert normalize_ocr_status_for_reader("done_degraded") == "done"
+
+
+def test_postprocess_failure_does_not_return_to_pending() -> None:
+    from paperforge.worker.ocr_errors import OCRPostprocessError, classify_ocr_error
+
+    state = classify_ocr_error(OCRPostprocessError("bad structure"), stage="postprocess")
+    assert state["status"] == "fatal_error"
+    assert state["retryable"] is False
+
+

tests/test_ocr_health.py

@@ -238,3 +238,18 @@ def test_ocr_health_includes_confidence_distributions() -> None:
     assert tbl_dist["high"] == 1
     assert tbl_dist["medium"] == 1
     assert tbl_dist["low"] == 1
+
+
+def test_health_report_includes_degraded_reasons() -> None:
+    from paperforge.worker.ocr_health import build_ocr_health
+
+    report = build_ocr_health(
+        page_count=1,
+        raw_blocks_count=0,
+        structured_blocks=[],
+        figure_inventory={},
+        table_inventory={},
+    )
+
+    assert "degraded_reasons" in report
+    assert len(report["degraded_reasons"]) > 0