Graph Status

akms.graph.graph_status

graph_status.py — Health Check & Review Report (§2.8 of system design).

Read-only health check tool. Run on demand or as part of review cycle.

Reports

• Degraded nodes (confidence < 0.5)
• Coverage flags (missing-detail, outdated)
• Tentative nodes awaiting promotion
• Dedup events persisted in local overlay history
• Blocked downstream tasks
• Id collisions between global and local nodes
• Orphaned nodes (no edges)
• Stale nodes (last_activated > N days)
• Orphaned overlay entries
• Docstring drift warnings (consumed from generate_mirror output)

Does NOT modify anything — pure read-only reporting.

format_report

format_report(report: dict[str, Any]) -> str

Format a health report as human-readable text.

Source code in packages/akms/src/akms/graph/graph_status.py

def format_report(report: dict[str, Any]) -> str:
    """Format a health report as human-readable text."""
    lines = ["═══ AKMS Graph Health Report ═══", ""]

    # Degraded nodes
    degraded = report.get("degraded_nodes", [])
    lines.append(f"## Degraded Nodes (confidence < 0.5): {len(degraded)}")
    for d in degraded:
        overlay_str = (
            f" (overlay: {d['overlay_confidence']:.2f})"
            if d.get("overlay_confidence") is not None
            else ""
        )
        lines.append(
            f"  - {d['node_id']}: {d['effective_confidence']:.2f}{overlay_str} [{d['origin']}]"
        )
    lines.append("")

    # Tentative
    tentative = report.get("tentative_nodes", [])
    lines.append(f"## Tentative Nodes Awaiting Promotion: {len(tentative)}")
    for t in tentative:
        lines.append(
            f"  - {t['node_id']} ({t['origin']}, domain={t['domain']}, conf={t['confidence']:.2f})"
        )
        # FR-R05: show the agent-written content_draft inline for quick review.
        draft = t.get("content_draft", "")
        if draft:
            lines.append("    ┌─ content_draft ─────────────────────────────")
            for draft_line in draft.splitlines():
                lines.append(f"    │ {draft_line}")
            lines.append("    └─────────────────────────────────────────────")
    lines.append("")

    # Id collisions
    collisions = report.get("id_collisions", [])
    lines.append(f"## Id Collisions: {len(collisions)}")
    for c in collisions:
        lines.append(f"  - {c['node_id']}: global + local both exist")
    lines.append("")

    # Orphaned nodes
    orphaned = report.get("orphaned_nodes", [])
    lines.append(f"## Orphaned Nodes (no edges): {len(orphaned)}")
    for o in orphaned:
        lines.append(f"  - {o['node_id']} ({o['domain']}, {o['origin']})")
    lines.append("")

    # Stale nodes
    stale = report.get("stale_nodes", [])
    lines.append(f"## Stale Nodes: {len(stale)}")
    for s in stale:
        lines.append(
            f"  - {s['node_id']}: {s['days_inactive']} days inactive (last: {s['last_activated']})"
        )
    lines.append("")

    # Orphaned overlay
    orphaned_overlay = report.get("orphaned_overlay_entries", [])
    lines.append(f"## Orphaned Overlay Entries: {len(orphaned_overlay)}")
    for o in orphaned_overlay:
        lines.append(f"  - {o['node_id']}: in overlay but not in graph")
    lines.append("")

    # Coverage flags
    coverage_flags = report.get("coverage_flags", [])
    lines.append(f"## Coverage Flags (missing-detail/outdated): {len(coverage_flags)}")
    for c in coverage_flags:
        phase_note = f", phase={c['phase']}" if c.get("phase") is not None else ""
        lines.append(
            f"  - {c['node_id']}: {c['coverage']} (source={c.get('source_id', '?')}{phase_note})"
        )
    lines.append("")

    # Dedup events
    dedup_events = report.get("dedup_events", [])
    lines.append(f"## Dedup Events: {len(dedup_events)}")
    for d in dedup_events:
        score_note = ""
        if d.get("score") is not None and d.get("threshold") is not None:
            score_note = f" (score={d['score']}, threshold={d['threshold']})"
        lines.append(
            f"  - {d.get('action', 'dedup')} merged into {d.get('merged_into', '?')}{score_note}"
        )
    lines.append("")

    # Blocked tasks
    blocked_tasks = report.get("blocked_tasks", [])
    lines.append(f"## Blocked Downstream Tasks: {len(blocked_tasks)}")
    for b in blocked_tasks:
        reason = f": {b['reason']}" if b.get("reason") else ""
        lines.append(f"  - {b.get('task', '?')}{reason}")
    lines.append("")

    # Drift warnings
    drift = report.get("drift_warnings", [])
    lines.append(f"## Docstring Drift Warnings: {len(drift)}")
    for d in drift:
        lines.append(f"  - {d.get('file', '?')}::{d['function']}: {d['detail']}")
    lines.append("")

    # Mirror provider identity (A2-6) — non-secret only
    mp = report.get("mirror_provider") or {}
    if mp:
        lines.append("## Mirror Provider")
        lines.append(
            f"  - provider: {mp.get('resolved_provider') or mp.get('provider', 'legacy')}"
        )
        if "success" in mp:
            lines.append(f"  - last_refresh_success: {mp.get('success')}")
        if mp.get("fallback_used"):
            lines.append("  - fallback_used: true")
        for err in mp.get("errors") or []:
            if isinstance(err, dict):
                lines.append(
                    f"  - error: [{err.get('code', '?')}] {err.get('message', '')}"
                )
            else:
                lines.append(f"  - error: {err}")
        lines.append("")

    # Skipped files (malformed frontmatter surfaced by build_graph)
    skipped = report.get("skipped_files", [])
    lines.append(f"## Skipped Files: {len(skipped)}")
    for s in skipped:
        lines.append(f"  - {s.get('path', '?')}: {s.get('reason', '')}")
    lines.append("")

    # Summary
    total_issues = (
        len(degraded)
        + len(tentative)
        + len(collisions)
        + len(orphaned)
        + len(stale)
        + len(orphaned_overlay)
        + len(coverage_flags)
        + len(dedup_events)
        + len(blocked_tasks)
        + len(drift)
    )
    lines.append(f"## Summary: {total_issues} total issues")
    lines.append(f"  Nodes in graph: {report.get('total_nodes', 0)}")
    lines.append(f"  Edges in graph: {report.get('total_edges', 0)}")

    return "\n".join(lines)

graph_status

graph_status(
    repo_root: str | Path,
    global_vault: str | Path | None = None,
    config: PropagationConfig | None = None,
    drift_warnings: list[dict] | None = None,
    blocked_tasks: list[dict | str] | None = None,
    today: date | None = None,
    mirror_provider: dict[str, Any] | None = None,
    *,
    allow_graph_rebuild: bool = True,
) -> dict[str, Any]

Run the full health check and return structured report.

Parameters:

Name	Type	Description	Default
repo_root	str | Path	Path to the repository root.	required
global_vault	str | Path | None	Override global vault path.	None
config	PropagationConfig | None	PropagationConfig (loads from file or defaults if None).	None
drift_warnings	list[dict] | None	Pre-computed drift warnings (from generate_mirror).	None
blocked_tasks	list[dict | str] | None	Optional blocked-task entries to include in report.	None
today	date | None	Override date for stale check (for testing).	None
mirror_provider	dict[str, Any] | None	Optional non-secret mirror provider identity / last refresh status (A2-6). When omitted, derived from config.mirror.	None
allow_graph_rebuild	bool	When False, skip build_graph if graph.json is missing (used after a required mirror-provider failure so a partial mirror set cannot enter the graph).	True

Returns:

Type	Description
dict[str, Any]	Dict with all check results.

Source code in packages/akms/src/akms/graph/graph_status.py

@traced("akms.graph_status")
def graph_status(
    repo_root: str | Path,
    global_vault: str | Path | None = None,
    config: PropagationConfig | None = None,
    drift_warnings: list[dict] | None = None,
    blocked_tasks: list[dict | str] | None = None,
    today: date | None = None,
    mirror_provider: dict[str, Any] | None = None,
    *,
    allow_graph_rebuild: bool = True,
) -> dict[str, Any]:
    """Run the full health check and return structured report.

    Args:
        repo_root: Path to the repository root.
        global_vault: Override global vault path.
        config: PropagationConfig (loads from file or defaults if None).
        drift_warnings: Pre-computed drift warnings (from generate_mirror).
        blocked_tasks: Optional blocked-task entries to include in report.
        today: Override date for stale check (for testing).
        mirror_provider: Optional non-secret mirror provider identity / last
            refresh status (A2-6). When omitted, derived from config.mirror.
        allow_graph_rebuild: When False, skip ``build_graph`` if graph.json is
            missing (used after a required mirror-provider failure so a partial
            mirror set cannot enter the graph).

    Returns:
        Dict with all check results.
    """
    repo_root = Path(repo_root)
    knowledge_dir = repo_root / "knowledge"
    graph_dir = knowledge_dir / "graph"
    graph_json = graph_dir / "graph.json"
    overlay_path = graph_dir / "local_state.yaml"

    # Load config
    if config is None:
        config_path = graph_dir / "propagation_config.yaml"
        if config_path.exists():
            config = parse_propagation_config(config_path)
        else:
            config = PropagationConfig()

    # Resolve global vault via shared helper (F-06 precedence).
    # Subsumes the earlier `.expanduser()` patch — the helper expands `~`
    # on every branch (explicit > env > config > default).
    vault_path = resolve_global_vault(explicit=global_vault, config=config)

    # Load graph — pass the resolved vault + config so a rebuild here sees
    # the exact same source tree the health checks will read.
    if graph_json.exists():
        G = load_graph(graph_json)
    elif allow_graph_rebuild:
        G = build_graph(repo_root, global_vault=vault_path, config=config)
    else:
        G = nx.DiGraph()
        G.graph["skipped_files"] = []
        logger.warning(
            "graph_status: graph.json missing and allow_graph_rebuild=False; "
            "returning empty graph diagnostics"
        )

    # Load overlay
    overlay: dict = {}
    if overlay_path.exists():
        with open(overlay_path) as f:
            overlay = yaml.safe_load(f) or {}

    local_nodes_dir = knowledge_dir / "local-nodes"

    # Provider identity (non-secret) for status surfaces.
    if mirror_provider is None:
        try:
            from akms.graph.mirror_provider import (
                public_provider_identity,
                resolve_mirror_config,
            )

            mirror_provider = public_provider_identity(resolve_mirror_config(config))
        except Exception:
            mirror_provider = {"provider": "legacy"}

    report = {
        "total_nodes": G.number_of_nodes(),
        "total_edges": G.number_of_edges(),
        "degraded_nodes": _check_degraded_nodes(G, overlay),
        "tentative_nodes": _check_tentative_nodes(G, repo_root, local_nodes_dir),
        "id_collisions": _check_id_collisions(vault_path, local_nodes_dir),
        "orphaned_nodes": _check_orphaned_nodes(G),
        "stale_nodes": _check_stale_nodes(
            G,
            overlay,
            config.graph.stale_node_days,
            today,
        ),
        "orphaned_overlay_entries": _check_orphaned_overlay_entries(G, overlay),
        "coverage_flags": _check_coverage_flags(overlay),
        "dedup_events": _check_dedup_events(overlay),
        "blocked_tasks": _check_blocked_tasks(overlay, blocked_tasks),
        "drift_warnings": drift_warnings or [],
        # Malformed-frontmatter files skipped by build_graph, surfaced here so
        # health reports expose non-fatal parse failures instead of silently
        # omitting sources.
        "skipped_files": list(G.graph.get("skipped_files", [])),
        # A2-6: configured / last-refresh mirror provider identity (non-secret).
        "mirror_provider": mirror_provider,
    }

    logger.info(
        "graph_status: %d nodes, %d edges, %d degraded, %d tentative, "
        "%d collisions, %d orphaned, %d stale, %d orphaned overlay, "
        "%d coverage_flags, %d dedup_events, %d blocked_tasks",
        report["total_nodes"],
        report["total_edges"],
        len(report["degraded_nodes"]),
        len(report["tentative_nodes"]),
        len(report["id_collisions"]),
        len(report["orphaned_nodes"]),
        len(report["stale_nodes"]),
        len(report["orphaned_overlay_entries"]),
        len(report["coverage_flags"]),
        len(report["dedup_events"]),
        len(report["blocked_tasks"]),
    )

    return report