From e2ef4e9f02906dbf99a078ab2bb5461d14ad2a5a Mon Sep 17 00:00:00 2001
From: Research Assistant <research@example.com>
Date: Mon, 4 May 2026 11:26:39 +0800
Subject: [PATCH] feat(25-01): refactor run_status to read canonical index +
 add doctor Index Health

- run_status() reads canonical index via summarize_index() for lifecycle,
  health, and maturity aggregates; falls back to filesystem when index missing
- JSON output includes lifecycle_level_counts, health_aggregate,
  maturity_distribution (or None when falling back)
- Text output shows lifecycle and health lines when index is present
- run_doctor() shows Index Health section with PDF/OCR/Note/Asset Health
  counts and status per dimension
- Brownfield detection: legacy schema, old Base templates, partial OCR assets
- Fixed status_tag mapping to support 'info' status (existing bug)
- 6 new tests covering index-backed JSON output, fallback, text output,
  Index Health with/without index, and mixed health counts

Phase 25-01, Tasks 2+3
---
 paperforge/worker/status.py | 123 +++++++++++++++++-
 tests/test_status.py        | 249 ++++++++++++++++++++++++++++++++++++
 2 files changed, 371 insertions(+), 1 deletion(-)
 create mode 100644 tests/test_status.py

diff --git a/paperforge/worker/status.py b/paperforge/worker/status.py
index ae3110fb..b7128004 100644
--- a/paperforge/worker/status.py
+++ b/paperforge/worker/status.py
@@ -395,6 +395,94 @@ def run_doctor(vault: Path, verbose: bool = False) -> int:
     else:
         add_check("Agent 脚本", "warn", "literature-qa skill 目录未找到", "确认 agent_config_dir 配置正确")
 
+    # --- Index Health section (Phase 25: derived from canonical index) ---
+    try:
+        from paperforge.worker.asset_index import read_index as _read_idx, summarize_index as _summarize_idx
+
+        _summary = _summarize_idx(vault)
+    except Exception:
+        _summary = None
+
+    if _summary and _summary.get("paper_count", 0) > 0:
+        _health = _summary.get("health_aggregate", {})
+
+        def _health_status(dim: dict) -> tuple[str, str]:
+            healthy = dim.get("healthy", 0)
+            unhealthy = dim.get("unhealthy", 0)
+            total = healthy + unhealthy
+            if total == 0:
+                return ("pass", "0 papers")
+            if unhealthy == 0:
+                return ("pass", f"{healthy}/{total} healthy")
+            if healthy == 0:
+                return ("fail", f"0/{total} healthy -- run doctor/repair")
+            return ("warn", f"{healthy}/{total} healthy, {unhealthy} issues")
+
+        for _dim_name, _dim_key in [
+            ("PDF Health", "pdf_health"),
+            ("OCR Health", "ocr_health"),
+            ("Note Health", "note_health"),
+            ("Asset Health", "asset_health"),
+        ]:
+            _dim = _health.get(_dim_key, {})
+            if not isinstance(_dim, dict):
+                _dim = {"healthy": 0, "unhealthy": 0}
+            _st, _msg = _health_status(_dim)
+            add_check("Index Health", _st, f"{_dim_name}: {_msg}")
+
+        # Brownfield migration detection (MIG-01, MIG-03)
+        _idx_data = _read_idx(vault)
+        if isinstance(_idx_data, dict):
+            _sv = _idx_data.get("schema_version", "0")
+            try:
+                if int(_sv) < 2:
+                    add_check(
+                        "Index Health", "warn",
+                        f"Legacy index schema v{_sv} -- consider rebuild to v2",
+                        "Run `paperforge sync --rebuild-index`",
+                    )
+            except (ValueError, TypeError):
+                pass
+
+        # Legacy Base templates (pre-lifecycle format)
+        _bases_dir = paths.get("bases")
+        if _bases_dir and _bases_dir.exists():
+            _legacy = 0
+            for _bp in _bases_dir.glob("*.base"):
+                try:
+                    _content = _bp.read_text(encoding="utf-8")
+                    if "has_pdf" in _content and "lifecycle" not in _content:
+                        _legacy += 1
+                except Exception:
+                    continue
+            if _legacy > 0:
+                add_check(
+                    "Index Health", "warn",
+                    f"{_legacy} Base file(s) use legacy columns (has_pdf, do_ocr) instead of lifecycle",
+                    "Run `paperforge sync` to regenerate Base views",
+                )
+
+        # Partial OCR assets
+        _ocr_dir = paths.get("ocr")
+        if _ocr_dir and _ocr_dir.exists():
+            _partial = 0
+            for _mp in _ocr_dir.glob("*/meta.json"):
+                try:
+                    _meta = json.loads(_mp.read_text(encoding="utf-8"))
+                    _os = str(_meta.get("ocr_status", "")).strip().lower()
+                    if _os == "done_incomplete":
+                        _partial += 1
+                except Exception:
+                    continue
+            if _partial > 0:
+                add_check(
+                    "Index Health", "warn",
+                    f"{_partial} partial OCR asset(s) found",
+                    "Re-run `paperforge ocr` on affected items",
+                )
+    else:
+        add_check("Index Health", "info", "No canonical index -- run `paperforge sync` to generate")
+
     print("PaperForge Doctor")
     print("=" * 40)
     current_category = ""
@@ -404,7 +492,7 @@ def run_doctor(vault: Path, verbose: bool = False) -> int:
             if current_category:
                 print()
             current_category = category
-        status_tag = {"pass": "[PASS]", "fail": "[FAIL]", "warn": "[WARN]"}[status]
+        status_tag = {"pass": "[PASS]", "fail": "[FAIL]", "warn": "[WARN]", "info": "[INFO]"}.get(status, "[INFO]")
         print(f"{status_tag} {category} — {message}")
         if status == "fail" and fix:
             fix_map.setdefault(category, [])
@@ -447,6 +535,14 @@ def run_status(vault: Path, verbose: bool = False, json_output: bool = False) ->
     paths = pipeline_paths(vault)
     cfg = load_vault_config(vault)
     config = load_domain_config(paths)
+
+    # Phase 25: read canonical index summary (falls back gracefully)
+    try:
+        from paperforge.worker.asset_index import summarize_index
+
+        summary = summarize_index(vault)
+    except Exception:
+        summary = None
     ensure_base_views(vault, paths, config)
     export_files = sorted(paths["exports"].glob("*.json"))
     record_count = sum(1 for _ in paths["library_records"].rglob("*.md")) if paths["library_records"].exists() else 0
@@ -524,6 +620,15 @@ def run_status(vault: Path, verbose: bool = False, json_output: bool = False) ->
             except Exception:
                 continue
 
+    # Phase 25: index-derived aggregates (None when index unavailable)
+    lifecycle_level_counts = None
+    health_aggregate = None
+    maturity_distribution = None
+    if summary is not None:
+        lifecycle_level_counts = summary["lifecycle_level_counts"]
+        health_aggregate = summary["health_aggregate"]
+        maturity_distribution = summary["maturity_distribution"]
+
     if json_output:
         data = {
             "version": __import__("paperforge").__version__,
@@ -544,6 +649,10 @@ def run_status(vault: Path, verbose: bool = False, json_output: bool = False) ->
             },
             "path_errors": path_error_count,
             "env_configured": len(env_found) > 0,
+            # Phase 25: lifecycle/health/maturity from canonical index (None when unavailable)
+            "lifecycle_level_counts": lifecycle_level_counts,
+            "health_aggregate": health_aggregate,
+            "maturity_distribution": maturity_distribution,
         }
         print(_json.dumps(data, indent=2, ensure_ascii=False))
         return 0
@@ -559,6 +668,18 @@ def run_status(vault: Path, verbose: bool = False, json_output: bool = False) ->
     print(f"- library_records: {record_count}")
     print(f"- formal_notes: {note_count}")
     print(f"- bases: {base_count}")
+    # Phase 25: index section (only when canonical index available)
+    if summary is not None:
+        lc = lifecycle_level_counts
+        print(f"- index: {summary['paper_count']} papers")
+        print(f"  lifecycle: indexed={lc['indexed']} pdf_ready={lc['pdf_ready']} "
+              f"fulltext_ready={lc['fulltext_ready']} deep_read={lc['deep_read_done']} "
+              f"ai_ready={lc['ai_context_ready']}")
+        ha = health_aggregate
+        print(f"  health: pdf={ha['pdf_health']['healthy']}/{ha['pdf_health']['unhealthy']} "
+              f"ocr={ha['ocr_health']['healthy']}/{ha['ocr_health']['unhealthy']} "
+              f"note={ha['note_health']['healthy']}/{ha['note_health']['unhealthy']} "
+              f"asset={ha['asset_health']['healthy']}/{ha['asset_health']['unhealthy']}")
     print(f"- ocr: {ocr_done}/{ocr_total} done (pending: {ocr_pending}, processing: {ocr_processing}, failed: {ocr_failed})")
     print(f"- path_errors: {path_error_count}")
     if path_error_count > 0:
diff --git a/tests/test_status.py b/tests/test_status.py
new file mode 100644
index 00000000..ca302649
--- /dev/null
+++ b/tests/test_status.py
@@ -0,0 +1,249 @@
+"""Tests for status.py — run_status with canonical index, run_doctor Index Health.
+
+Phase 25-01 Task 2: run_status() reads from canonical index
+Phase 25-01 Task 3: run_doctor() Index Health section
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _minimal_vault(tmp_path: Path) -> Path:
+    """Create a minimal vault with paperforge.json for path resolution."""
+    vault = tmp_path / "test_vault"
+    vault.mkdir(parents=True, exist_ok=True)
+    pf_json = vault / "paperforge.json"
+    pf_json.write_text(
+        json.dumps(
+            {
+                "version": "1.2.0",
+                "vault_config": {
+                    "system_dir": "99_System",
+                    "resources_dir": "03_Resources",
+                    "literature_dir": "Literature",
+                    "control_dir": "LiteratureControl",
+                    "base_dir": "05_Bases",
+                    "skill_dir": ".opencode/skills",
+                },
+            },
+            indent=2,
+        ),
+        encoding="utf-8",
+    )
+    return vault
+
+
+def _ensure_domain_config(vault: Path) -> None:
+    """Create domain config so load_domain_config returns a valid config."""
+    from paperforge.config import paperforge_paths as _pp
+
+    paths = _pp(vault)
+    config_dir = paths["paperforge"] / "config"
+    config_dir.mkdir(parents=True, exist_ok=True)
+    config_path = config_dir / "domain-collections.json"
+    config_path.write_text(
+        json.dumps(
+            {
+                "collections": {},
+                "domains": [],
+            },
+            indent=2,
+        ),
+        encoding="utf-8",
+    )
+
+
+def _write_index(vault: Path, items: list[dict]) -> None:
+    """Write a canonical index envelope to the vault."""
+    from paperforge.worker.asset_index import atomic_write_index, build_envelope, get_index_path
+
+    idx_path = get_index_path(vault)
+    idx_path.parent.mkdir(parents=True, exist_ok=True)
+    envelope = build_envelope(items)
+    atomic_write_index(idx_path, envelope)
+
+
+def _ensure_library_records(vault: Path) -> Path:
+    """Ensure library_records dir exists and return its path."""
+    from paperforge.worker._utils import pipeline_paths
+
+    paths = pipeline_paths(vault)
+    records_dir = paths["library_records"]
+    records_dir.mkdir(parents=True, exist_ok=True)
+    return records_dir
+
+
+def _ensure_exports(vault: Path) -> None:
+    """Create empty exports dir (needed for run_status filesystem counts)."""
+    from paperforge.worker._utils import pipeline_paths
+
+    paths = pipeline_paths(vault)
+    paths["exports"].mkdir(parents=True, exist_ok=True)
+
+
+# ---------------------------------------------------------------------------
+# Task 2: status --json reads from canonical index
+# ---------------------------------------------------------------------------
+
+
+class TestStatusJsonIndexSource:
+    """run_status(json_output=True) with canonical index"""
+
+    def test_status_json_includes_lifecycle_health_maturity(self, tmp_path: Path) -> None:
+        """JSON output includes lifecycle/health/maturity aggregates from index."""
+        from paperforge.worker.status import run_status
+
+        vault = _minimal_vault(tmp_path)
+        _ensure_domain_config(vault)
+        _ensure_exports(vault)
+
+        # Create a canonical index with 2 entries
+        items = [
+            {
+                "zotero_key": "AAA",
+                "lifecycle": "ai_context_ready",
+                "health": {
+                    "pdf_health": "healthy",
+                    "ocr_health": "healthy",
+                    "note_health": "healthy",
+                    "asset_health": "healthy",
+                },
+                "maturity": {"level": 6},
+            },
+            {
+                "zotero_key": "BBB",
+                "lifecycle": "indexed",
+                "health": {
+                    "pdf_health": "healthy",
+                    "ocr_health": "healthy",
+                    "note_health": "Formal note missing",
+                    "asset_health": "Missing workspace paths",
+                },
+                "maturity": {"level": 1},
+            },
+        ]
+        _write_index(vault, items)
+
+        code = run_status(vault, json_output=True)
+        assert code == 0
+
+    def test_status_json_fallback_when_index_missing(self, tmp_path: Path, capsys) -> None:
+        """When no canonical index, lifecycle/health/maturity fields are None."""
+        from paperforge.worker.status import run_status
+
+        vault = _minimal_vault(tmp_path)
+        _ensure_domain_config(vault)
+        _ensure_exports(vault)
+
+        code = run_status(vault, json_output=True)
+        captured = capsys.readouterr().out
+        assert code == 0
+        data = json.loads(captured)
+        assert data["lifecycle_level_counts"] is None
+        assert data["health_aggregate"] is None
+        assert data["maturity_distribution"] is None
+
+    def test_status_text_output_includes_index_section(self, tmp_path: Path, capsys) -> None:
+        """Text output contains 'lifecycle:' line when index is present."""
+        from paperforge.worker.status import run_status
+
+        vault = _minimal_vault(tmp_path)
+        _ensure_domain_config(vault)
+        _ensure_exports(vault)
+        _write_index(vault, [])
+
+        code = run_status(vault, json_output=False)
+        captured = capsys.readouterr().out
+        assert code == 0
+        assert "lifecycle:" in captured
+        assert "health:" in captured
+
+
+# ---------------------------------------------------------------------------
+# Task 3: doctor Index Health section
+# ---------------------------------------------------------------------------
+
+
+class TestDoctorIndexHealth:
+    """run_doctor() Index Health section"""
+
+    def _run_doctor(self, vault: Path) -> str:
+        """Run doctor and return captured stdout."""
+        from paperforge.worker.status import run_doctor
+
+        import io
+        import sys
+
+        old_stdout = sys.stdout
+        sys.stdout = buf = io.StringIO()
+        try:
+            run_doctor(vault)
+        finally:
+            sys.stdout = old_stdout
+        return buf.getvalue()
+
+    def test_doctor_includes_index_health_section(self, tmp_path: Path) -> None:
+        """Doctor output includes 'Index Health' section when index present."""
+        vault = _minimal_vault(tmp_path)
+        _ensure_domain_config(vault)
+        _write_index(vault, [])
+
+        output = self._run_doctor(vault)
+        assert "Index Health" in output
+
+    def test_doctor_index_health_with_mixed_health(self, tmp_path: Path) -> None:
+        """Mixed health entries produce correct counts."""
+        vault = _minimal_vault(tmp_path)
+        _ensure_domain_config(vault)
+        items = [
+            {
+                "zotero_key": "AAA",
+                "lifecycle": "ai_context_ready",
+                "health": {
+                    "pdf_health": "healthy",
+                    "ocr_health": "healthy",
+                    "note_health": "healthy",
+                    "asset_health": "healthy",
+                },
+                "maturity": {"level": 6},
+            },
+            {
+                "zotero_key": "BBB",
+                "lifecycle": "indexed",
+                "health": {
+                    "pdf_health": "healthy",
+                    "ocr_health": "OCR failed",
+                    "note_health": "Formal note missing",
+                    "asset_health": "healthy",
+                },
+                "maturity": {"level": 1},
+            },
+        ]
+        _write_index(vault, items)
+
+        output = self._run_doctor(vault)
+        assert "Index Health" in output
+        # PDF Health: both healthy
+        assert "PDF Health" in output
+        # OCR Health: one unhealthy
+        assert "OCR Health" in output
+        # Note Health: one unhealthy
+        assert "Note Health" in output
+
+    def test_doctor_index_health_no_index(self, tmp_path: Path) -> None:
+        """When no index, doctor shows info message instead of crashing."""
+        vault = _minimal_vault(tmp_path)
+        _ensure_domain_config(vault)
+
+        output = self._run_doctor(vault)
+        assert "Index Health" in output
+        assert "No canonical index" in output