feat: add layered runtime health computation

paperforge/memory/runtime_health.py

@@ -0,0 +1,217 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+
+def get_runtime_health(vault: Path) -> dict:
+    """Compute full runtime health for a vault. Returns layers + summary + capabilities."""
+    layers = {
+        "bootstrap": _check_bootstrap(vault),
+        "read": _check_read(vault),
+        "write": _check_write(vault),
+        "index": _check_index(vault),
+        "vector": _check_vector(vault),
+    }
+    summary = _derive_summary(layers)
+    capabilities = _derive_capabilities(layers)
+    return {
+        "summary": summary,
+        "layers": layers,
+        "capabilities": capabilities,
+    }
+
+
+def _layer(status: str, evidence: list, next_action: str = "", repair_command: str = "") -> dict:
+    return {
+        "status": status,
+        "evidence": evidence,
+        "next_action": next_action,
+        "repair_command": repair_command,
+    }
+
+
+def _check_bootstrap(vault: Path) -> dict:
+    pf_json = vault / "paperforge.json"
+    if not pf_json.exists():
+        return _layer("blocked", [f"paperforge.json not found at {vault}"],
+                      "Run paperforge setup to initialize vault",
+                      "paperforge setup --headless")
+    try:
+        import json
+        cfg = json.loads(pf_json.read_text(encoding="utf-8"))
+    except Exception as e:
+        return _layer("blocked", [f"Cannot read paperforge.json: {e}"],
+                      "Fix paperforge.json syntax",
+                      "paperforge doctor")
+    system_dir = cfg.get("vault_config", {}).get("system_dir") or cfg.get("system_dir", "System")
+    pf_root = vault / system_dir / "PaperForge"
+    if not pf_root.exists():
+        return _layer("degraded", [f"PaperForge directory not found at {pf_root}"],
+                      "Run setup to create directory structure",
+                      "paperforge doctor")
+    return _layer("ok", [f"paperforge.json ok at {pf_json}"])
+
+
+def _check_read(vault: Path) -> dict:
+    from paperforge.memory.db import get_memory_db_path
+    from paperforge.config import paperforge_paths
+
+    paths = paperforge_paths(vault)
+    index_path = paths.get("index")
+    if index_path and not Path(index_path).exists():
+        return _layer("degraded", ["Canonical index not found"],
+                      "Run paperforge sync --rebuild-index",
+                      "paperforge sync --rebuild-index")
+
+    db_path = get_memory_db_path(vault)
+    if not db_path.exists():
+        return _layer("degraded", ["Memory DB not found, can be rebuilt"],
+                      "Run paperforge memory build",
+                      "paperforge memory build")
+
+    logs_dir = paths.get("paperforge", vault / "System" / "PaperForge") / "logs"
+    if not logs_dir.exists():
+        return _layer("degraded", ["JSONL logs directory missing"],
+                      "Create logs directory or run sync",
+                      "mkdir -p \"$logs_dir\"")
+
+    return _layer("ok", ["Canonical index found", "Memory DB exists", "JSONL logs dir found"])
+
+
+def _check_write(vault: Path) -> dict:
+    from paperforge.config import paperforge_paths
+    paths = paperforge_paths(vault)
+    pf_root = paths.get("paperforge", vault / "System" / "PaperForge")
+    logs_dir = pf_root / "logs"
+    evidence = []
+    try:
+        logs_dir.mkdir(parents=True, exist_ok=True)
+        test_file = logs_dir / ".health-check"
+        test_file.write_text("ok", encoding="utf-8")
+        test_file.unlink()
+        evidence.append("JSONL logs dir writable")
+    except Exception as e:
+        return _layer("blocked", [f"JSONL logs dir not writable: {e}"],
+                      "Check filesystem permissions", "")
+    return _layer("ok", evidence)
+
+
+def _check_index(vault: Path) -> dict:
+    from paperforge.memory.db import get_connection, get_memory_db_path
+    db_path = get_memory_db_path(vault)
+    if not db_path.exists():
+        return _layer("blocked", ["Memory DB not found"],
+                      "Run paperforge memory build",
+                      "paperforge memory build")
+    try:
+        conn = get_connection(db_path, read_only=True)
+        version = conn.execute("SELECT value FROM meta WHERE key='schema_version'").fetchone()
+        conn.close()
+        if version:
+            return _layer("ok", [f"Schema version: {version['value']}"])
+        return _layer("degraded", ["Schema version unknown"],
+                      "Rebuild memory DB", "paperforge memory build --force")
+    except Exception as e:
+        return _layer("blocked", [f"DB query failed: {e}"],
+                      "Rebuild memory DB",
+                      "paperforge memory build --force")
+
+
+def _check_vector(vault: Path) -> dict:
+    from paperforge.memory.vector_db import get_vector_db_path, read_vector_build_state
+
+    settings_path = vault / ".obsidian" / "plugins" / "paperforge" / "data.json"
+    vector_enabled = False
+    if settings_path.exists():
+        try:
+            import json
+            data = json.loads(settings_path.read_text(encoding="utf-8"))
+            vector_enabled = bool(data.get("features", {}).get("vector_db", False))
+        except Exception:
+            pass
+
+    if not vector_enabled:
+        return _layer("ok", ["Vector DB disabled by user"])
+
+    build_state = read_vector_build_state(vault)
+    job_status = build_state.get("status", "idle")
+    if job_status == "running":
+        return _layer("degraded", ["Vector build in progress"],
+                      "Wait for build to complete",
+                      "paperforge embed status --json")
+    if job_status == "failed":
+        return _layer("degraded", [f"Last build failed: {build_state.get('message', '')}"],
+                      "Check error and rebuild",
+                      "paperforge embed build --resume")
+
+    db_path = get_vector_db_path(vault)
+    if not db_path.exists():
+        return _layer("degraded", ["Vector DB not built yet"],
+                      "Run embed build",
+                      "paperforge embed build --resume")
+
+    return _layer("ok", ["Vector DB exists"])
+
+
+def _derive_summary(layers: dict) -> dict:
+    bootstrap_s = layers["bootstrap"]["status"]
+    read_s = layers["read"]["status"]
+    write_s = layers["write"]["status"]
+    index_s = layers["index"]["status"]
+    vector_s = layers["vector"]["status"]
+
+    blocked = False
+    degraded = False
+    reasons = []
+
+    if bootstrap_s == "blocked":
+        blocked = True
+        reasons.append("Bootstrap blocked")
+    if read_s == "blocked" and write_s == "blocked":
+        blocked = True
+        reasons.append("Read and write both blocked")
+    if read_s in ("degraded", "blocked"):
+        degraded = True
+        reasons.append("Read layer degraded")
+    if write_s in ("degraded", "blocked"):
+        degraded = True
+        reasons.append("Write layer degraded")
+    if index_s in ("degraded", "blocked"):
+        degraded = True
+        reasons.append("Index layer degraded")
+    if vector_s in ("degraded", "blocked"):
+        degraded = True
+        if vector_s == "degraded":
+            reasons.append("Vector build degraded")
+        else:
+            reasons.append("Vector DB blocked")
+
+    if blocked:
+        status = "blocked"
+    elif degraded:
+        status = "degraded"
+    else:
+        status = "ok"
+        reasons.append("All systems operational")
+
+    return {
+        "status": status,
+        "reason": "; ".join(reasons),
+        "safe_read": read_s == "ok",
+        "safe_write": write_s == "ok",
+        "safe_build": index_s == "ok",
+        "safe_vector": vector_s == "ok",
+    }
+
+
+def _derive_capabilities(layers: dict) -> dict:
+    read_ok = layers["read"]["status"] == "ok"
+    write_ok = layers["write"]["status"] == "ok"
+    vector_ok = layers["vector"]["status"] == "ok"
+    return {
+        "paper_context": read_ok,
+        "reading_log_write": write_ok,
+        "project_log_write": write_ok,
+        "fts_search": read_ok,
+        "vector_retrieve": vector_ok,
+    }

tests/unit/memory/test_runtime_health.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from pathlib import Path
+import json
+
+from paperforge.memory.runtime_health import (
+    get_runtime_health,
+    _check_bootstrap,
+    _check_write,
+)
+
+
+def test_runtime_health_blocks_without_paperforge_json(tmp_path):
+    vault = tmp_path / "novault"
+    vault.mkdir()
+    health = get_runtime_health(vault)
+    assert health["summary"]["status"] == "blocked"
+    assert health["layers"]["bootstrap"]["status"] == "blocked"
+    assert health["capabilities"]["paper_context"] == False
+
+
+def test_bootstrap_with_paperforge_json(tmp_path):
+    vault = tmp_path / "vault"
+    vault.mkdir()
+    pf_json = vault / "paperforge.json"
+    pf_json.write_text(json.dumps({"system_dir": "System"}), encoding="utf-8")
+    (vault / "System" / "PaperForge").mkdir(parents=True)
+    result = _check_bootstrap(vault)
+    assert result["status"] == "ok"
+
+
+def test_write_layer(tmp_path):
+    vault = tmp_path / "vault"
+    vault.mkdir()
+    result = _check_write(vault)
+    assert result["status"] == "ok"
+    assert any("writable" in e for e in result["evidence"])
+
+
+def test_runtime_health_summary_has_expected_keys(tmp_path):
+    vault = tmp_path / "vault"
+    vault.mkdir()
+    health = get_runtime_health(vault)
+    summary = health["summary"]
+    for key in ("status", "reason", "safe_read", "safe_write", "safe_build", "safe_vector"):
+        assert key in summary
+    for layer in ("bootstrap", "read", "write", "index", "vector"):
+        assert layer in health["layers"]
+        for key in ("status", "evidence", "next_action", "repair_command"):
+            assert key in health["layers"][layer]