Jyotisha/scripts/preflight_fragment_scan.py

#!/usr/bin/env python3
"""Preflight scan for fragment discipline, redundancy, and real-capability boundaries.

This script is intentionally stdlib-only and reuses existing project audits.
It exists to enforce a "scan before work" rule for multi-window development,
so high-value drafts, mirrors, and external-work-brain fragments do not get
forgotten or confused with source truth.
"""

from __future__ import annotations

import json
import os
import re
import subprocess
import sys
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
SCRIPT_DIR = ROOT / "scripts"
if str(SCRIPT_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPT_DIR))

import oracle_boundary_audit  # noqa: E402


LOCAL_DRAFTS_DIR = ROOT / "docs" / "research" / "local_drafts" / "2026-06"
EXTERNAL_WORK_BRAIN_DIR = Path("<home>/.gemini/antigravity-ide/brain")
DISTRIBUTION_MIRROR_DIR = Path("<home>/.workbuddy/skills/jyotish-vedic-astrology")
ORACLE_FILE = ROOT / "references" / "oracle" / "dasha_shadbala_oracle_cases.json"
REPO_CLEANUP_MAP = ROOT / "docs" / "research" / "repo_cleanup_promotion_map_2026_07_01.md"
ERROR_LEDGER = ROOT / "docs" / "research" / "pre_work_error_ledger.md"
LATEST_SWEEP = ROOT / "docs" / "research" / "whole_machine_fragment_sweep_2026_07_05.md"
ROUND25_SWEEP = ROOT / "docs" / "research" / "whole_machine_fragment_sweep_round25_2026_06_25.md"


def _read_text(path: Path) -> str:
    try:
        return path.read_text(encoding="utf-8", errors="ignore")
    except OSError:
        return ""


def _run_json_script(*args: str) -> dict[str, Any]:
    completed = subprocess.run(
        [sys.executable, *args],
        cwd=ROOT,
        text=True,
        capture_output=True,
        timeout=120,
        check=False,
    )
    if completed.returncode != 0:
        raise RuntimeError(completed.stderr or completed.stdout or f"Failed: {' '.join(args)}")
    return json.loads(completed.stdout)


def _list_files(base: Path, patterns: tuple[str, ...]) -> list[str]:
    if not base.exists():
        return []
    items: list[str] = []
    for pattern in patterns:
        for path in base.rglob(pattern):
            if path.is_file():
                items.append(str(path))
    return sorted(set(items))


def _summarize_local_drafts() -> list[dict[str, Any]]:
    if not LOCAL_DRAFTS_DIR.exists():
        return []
    promoted_names = _promoted_draft_names()
    preferred_tokens = (
        "reuse_audit",
        "three_fronts",
        "tajika",
        "shadbala",
        "dasha",
        "yogi",
        "fragment",
        "truth",
        "benchmark",
    )
    rows: list[dict[str, Any]] = []
    for path in sorted(LOCAL_DRAFTS_DIR.glob("*.md")):
        name = path.name
        if name in promoted_names:
            continue
        if not any(token in name for token in preferred_tokens):
            continue
        rows.append(
            {
                "category": "repo_local_draft",
                "path": str(path),
                "reason": "High-value draft not promoted into repo truth yet.",
            }
        )
    return rows[:12]


def _promoted_draft_names() -> set[str]:
    docs_dir = ROOT / "docs" / "research"
    names: set[str] = set()
    for path in docs_dir.glob("*repo_truth_pack_2026_07_01.md"):
        text = _read_text(path)
        names.update(re.findall(r"`([^`]+\.md)`", text))
    return names


def _summarize_external_work_brain() -> list[dict[str, Any]]:
    if not EXTERNAL_WORK_BRAIN_DIR.exists():
        return []
    rows: list[dict[str, Any]] = []
    for path in _list_files(
        EXTERNAL_WORK_BRAIN_DIR,
        ("*.md", "*.py"),
    ):
        name = os.path.basename(path).lower()
        if not any(token in name for token in ("vedastro", "audit", "skill", "workflow", "oracle")):
            continue
        rows.append(
            {
                "category": "external_work_brain",
                "path": path,
                "reason": "Recovery-only work-brain artifact; re-anchor before reuse.",
            }
        )
    return rows[:12]


def _redundant_or_mirror_rows(fragment_audit: dict[str, Any]) -> list[dict[str, Any]]:
    rows = [
        {
            "category": "distribution_mirror",
            "path": str(DISTRIBUTION_MIRROR_DIR),
            "reason": "Historical skill mirror; do not reverse-sync over main repo truth.",
        }
    ]
    for rel in fragment_audit.get("workspace_residue", {}).get("untracked_files", [])[:12]:
        rows.append(
            {
                "category": "workspace_residue",
                "path": str(ROOT / rel),
                "reason": "Untracked residue or generated artifact; review before treating as source truth.",
            }
        )
    return rows


def _real_capability_risks(oracle_boundary: dict[str, Any]) -> list[dict[str, Any]]:
    summary = oracle_boundary.get("summary", {})
    risks = [
        {
            "id": "external_oracle_not_closed",
            "severity": "high",
            "reason": "External oracle boundary remains open; production tuning is still blocked.",
            "evidence": {
                "production_tuning_recommended": summary.get("production_tuning_recommended"),
                "open_items": summary.get("open_items", []),
            },
        },
        {
            "id": "historical_event_accuracy_not_proven",
            "severity": "high",
            "reason": "Engineering surfaces are covered, but historical life-event accuracy is not yet proven by external oracle closure.",
            "evidence": {
                "routes_present": ["career_timing_strict", "event_verification_strict", "full_reading_strict"],
                "needs_real_backtest": True,
            },
        },
        {
            "id": "official_vs_local_boundary",
            "severity": "medium",
            "reason": "VedAstro official ingestion exists, but some real-capability claims still depend on local fallback and incomplete external comparison packs.",
            "evidence": {
                "longitude_cases": summary.get("longitude_cases", 0),
                "dasha_cases": summary.get("dasha_cases", 0),
                "shadbala_cases": summary.get("shadbala_cases", 0),
            },
        },
    ]
    return risks


def _cleanup_priorities(fragment_audit: dict[str, Any], high_value_unpromoted: list[dict[str, Any]]) -> list[dict[str, Any]]:
    workspace_residue = fragment_audit.get("workspace_residue", {}).get("untracked_files", [])
    priorities = [
        {
            "id": "triage_workspace_residue",
            "severity": "high" if workspace_residue else "medium",
            "why": "Untracked residue can hide generated truth, stale artifacts, or partial experiments across windows.",
            "evidence": {
                "untracked_count": len(workspace_residue),
                "sample_paths": workspace_residue[:5],
            },
        },
        {
            "id": "promote_or_archive_high_value_drafts",
            "severity": "medium",
            "why": "High-value draft audits should either be promoted into repo truth or explicitly archived to reduce rediscovery cost.",
            "evidence": {
                "draft_count": len(high_value_unpromoted),
                "sample_paths": [item["path"] for item in high_value_unpromoted[:5]],
            },
        },
    ]
    return priorities


def build_report() -> dict[str, Any]:
    fragment_audit = _run_json_script("scripts/audit_fragments.py", "--strict")
    capability_audit = _run_json_script(
        "scripts/jyotish_engine.py",
        "audit-capabilities",
        "--mode",
        "validate",
    )
    oracle = json.loads(_read_text(ORACLE_FILE))
    oracle_boundary = oracle_boundary_audit.build_report(oracle)

    high_value_unpromoted = _summarize_local_drafts() + _summarize_external_work_brain()
    redundant_or_mirror = _redundant_or_mirror_rows(fragment_audit)
    real_capability_risks = _real_capability_risks(oracle_boundary)
    cleanup_priorities = _cleanup_priorities(fragment_audit, high_value_unpromoted)

    return {
        "scope": "preflight_fragment_scan",
        "summary": {
            "authority_layers_scanned": 4,
            "production_truth_layer": "main_repo_truth",
            "high_value_unpromoted_count": len(high_value_unpromoted),
            "redundant_or_mirror_count": len(redundant_or_mirror),
            "workspace_residue_count": len(fragment_audit.get("workspace_residue", {}).get("untracked_files", [])),
            "real_capability_risk_count": len(real_capability_risks),
            "real_capability_status": "engineering_surfaces_covered_but_external_accuracy_not_closed",
        },
        "layers": {
            "main_repo_truth": {
                "status": "authoritative",
                "paths": [
                    str(ROOT / "SKILL.md"),
                    str(ROOT / "AGENTS.md"),
                    str(ROOT / "scripts"),
                    str(ROOT / "tests"),
                    str(ROOT / "references"),
                    str(ROOT / "docs" / "research"),
                ],
            },
            "repo_local_drafts": {
                "status": "draft_reference_only",
                "path": str(LOCAL_DRAFTS_DIR),
            },
            "external_work_brain": {
                "status": "recovery_reference_only",
                "path": str(EXTERNAL_WORK_BRAIN_DIR),
            },
            "distribution_mirror": {
                "status": "mirror_do_not_reverse_sync",
                "path": str(DISTRIBUTION_MIRROR_DIR),
            },
        },
        "upstream_audits": {
            "fragment_audit": {
                "valid": fragment_audit.get("valid"),
                "candidate_count": fragment_audit.get("fragments", {}).get("candidate_count"),
                "workspace_residue_count": fragment_audit.get("workspace_residue", {}).get("untracked_count"),
            },
            "capability_audit": {
                "valid": capability_audit.get("valid"),
                "technique_count": capability_audit.get("technique_count"),
                "problem_count": capability_audit.get("problem_count"),
            },
        },
        "findings": {
            "high_value_unpromoted_count": len(high_value_unpromoted),
            "redundant_or_mirror_count": len(redundant_or_mirror),
            "workspace_residue_count": len(fragment_audit.get("workspace_residue", {}).get("untracked_files", [])),
            "real_capability_risk_count": len(real_capability_risks),
        },
        "high_value_unpromoted": high_value_unpromoted,
        "redundant_or_mirror": redundant_or_mirror,
        "real_capability_boundary": {
            "status": "not_fully_closed",
            "oracle_boundary": {
                "scope": oracle_boundary.get("scope"),
                "production_tuning_recommended": oracle_boundary.get("summary", {}).get("production_tuning_recommended"),
                "dasha_cases": oracle_boundary.get("summary", {}).get("dasha_cases"),
                "shadbala_cases": oracle_boundary.get("summary", {}).get("shadbala_cases"),
                "longitude_cases": oracle_boundary.get("summary", {}).get("longitude_cases"),
                "open_items": oracle_boundary.get("summary", {}).get("open_items", []),
            },
        },
        "real_capability_risks": real_capability_risks,
        "cleanup_priorities": cleanup_priorities,
        "cleanup_map": {
            "path": str(REPO_CLEANUP_MAP),
            "exists": REPO_CLEANUP_MAP.exists(),
            "focus_layers": [
                "docs/research/local_drafts/2026-06",
                "<home>/.gemini/antigravity-ide/brain",
                "<home>/.workbuddy/skills/jyotish-vedic-astrology",
            ],
        },
        "governance": {
            "pre_work_error_ledger": {
                "path": str(ERROR_LEDGER),
                "exists": ERROR_LEDGER.exists(),
            },
            "latest_fragment_sweep": {
                "path": str(LATEST_SWEEP),
                "exists": LATEST_SWEEP.exists(),
                "remote_ref_parity": "blocked_until_git_ls_remote_succeeds",
            },
            "prior_fragment_sweep": {
                "path": str(ROUND25_SWEEP),
                "exists": ROUND25_SWEEP.exists(),
            },
            "acceptance_command": (
                "python3 -m pytest -q tests/test_runtime_import_boundaries.py "
                "tests/test_project_fragment_governance.py tests/test_preflight_fragment_scan.py "
                "tests/test_remote_repo_visibility_check.py tests/test_pre_work_check.py"
            ),
        },
        "boundary": (
            "Run this preflight scan before major work so drafts, mirrors, and external-work-brain "
            "fragments are reviewed deliberately, and so engineering-surface success is not mistaken "
            "for externally validated historical accuracy."
        ),
    }


def main() -> int:
    print(json.dumps(build_report(), ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())