agent/tools/check_project.py

#!/usr/bin/env python3
"""Run deterministic repository integrity checks without external services."""

from __future__ import annotations

import ast
import json
from collections import Counter
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
COLLECTIONS = ("jobs", "papers", "industry")
REQUIRED_FILES = (
    "AGENTS.md",
    "README.md",
    "docs/06-project-status.md",
    "docs/07-exploration-history.md",
    "docs/08-next-work.md",
    "data/index.json",
    "data/summary.json",
    "research/memory/findings.md",
    "research/memory/evidence-ledger.md",
    "research/evaluation/k1412-agent-evaluation-v1.md",
    "data/evaluation/k1412-agent-eval-task-contracts-v1.json",
    "tools/evaluation/score_agent_runs.py",
    "web/app.py",
    "web/agent-knowledge-atlas.service",
    "web/manage.sh",
)


def load_json(path: str):
    return json.loads((ROOT / path).read_text(encoding="utf-8"))


def require(condition: bool, message: str) -> None:
    if not condition:
        raise AssertionError(message)


def check_required_files() -> None:
    missing = [path for path in REQUIRED_FILES if not (ROOT / path).is_file()]
    require(not missing, f"missing required files: {', '.join(missing)}")


def check_collection_index() -> dict[str, int]:
    index = load_json("data/index.json")
    summary = load_json("data/summary.json")
    require(isinstance(index, list), "data/index.json must be an array")

    paths = [str(item.get("path") or "") for item in index]
    require(len(paths) == len(set(paths)), "data/index.json contains duplicate paths")
    missing_paths = [path for path in paths if not (ROOT / path).is_file()]
    require(not missing_paths, f"index points to missing files: {missing_paths[:5]}")

    indexed = Counter(str(item.get("collection") or "") for item in index)
    filesystem = {
        collection: sum(
            1
            for path in (ROOT / collection / "items").glob("*.md")
            if path.name != "README.md"
        )
        for collection in COLLECTIONS
    }
    require(dict(indexed) == filesystem, f"index/file count mismatch: {dict(indexed)} != {filesystem}")
    require(summary.get("total") == len(index), "summary total does not match index")
    require(summary.get("by_collection") == filesystem, "summary collection counts are stale")
    return filesystem


def check_research_data() -> dict[str, int]:
    expanded = load_json("data/research/agent-memory-expanded.json")
    screened = load_json("data/research/agent-memory-screened.json")
    landscape = load_json("data/research/memory-landscape.json")
    corpus = load_json("data/research/memory-corpus.json")

    candidate_count = len(expanded.get("papers", []))
    require(expanded.get("unique_papers") == candidate_count, "expanded memory count mismatch")
    require(screened.get("total") == candidate_count, "screen total mismatch")
    require(screened.get("completed") == candidate_count, "memory screening is incomplete")
    require(len(screened.get("papers", [])) == candidate_count, "screen paper count mismatch")
    require(landscape.get("source_candidates") == candidate_count, "landscape source count mismatch")
    require(corpus.get("resolved") == len(corpus.get("papers", [])), "local memory corpus mismatch")

    return {
        "memory_candidates": candidate_count,
        "memory_title_central": int(landscape.get("title_central") or 0),
        "memory_local_topic": int(corpus.get("resolved") or 0),
    }


def check_python_syntax() -> int:
    paths = sorted((ROOT / "tools").rglob("*.py")) + [ROOT / "web" / "app.py"]
    for path in paths:
        ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
    return len(paths)


def check_ollama_contract() -> None:
    source = (ROOT / "web" / "app.py").read_text(encoding="utf-8")
    require('"num_ctx"' not in source, "web app must not override Ollama num_ctx")
    require('"keep_alive"' not in source, "web app must not override Ollama keep_alive")


def check_experiment_state() -> None:
    module = load_json("learning/agent-memory/module.json")
    require(module.get("status") == "failed", "old learning pilot must remain marked failed")
    pilot = (ROOT / "experiments/knowledge-compilation/2026-07-10-agent-memory-pilot.md").read_text(
        encoding="utf-8"
    )
    require("status: failed" in pilot, "pilot experiment status is stale")


def check_evaluation_contract() -> int:
    suite = load_json("data/evaluation/k1412-agent-eval-task-contracts-v1.json")
    require(
        suite.get("schema_version") == "agent-eval-task-contracts/v1",
        "unknown Agent evaluation task schema",
    )
    tasks = suite.get("tasks")
    require(isinstance(tasks, list) and tasks, "Agent evaluation task suite is empty")
    budget_profiles = suite.get("budget_profiles") or {}
    require(isinstance(budget_profiles, dict) and budget_profiles, "evaluation budget profiles are empty")
    task_ids = [str(task.get("task_id") or "") for task in tasks]
    require(all(task_ids), "Agent evaluation task is missing task_id")
    require(len(task_ids) == len(set(task_ids)), "Agent evaluation task ids are not unique")
    for task in tasks:
        require(task.get("title_zh"), f"{task['task_id']} is missing Chinese title")
        require(task.get("version"), f"{task['task_id']} is missing version")
        require(task.get("runner_status"), f"{task['task_id']} is missing runner_status")
        require(
            task.get("budget_profile") in budget_profiles,
            f"{task['task_id']} references an unknown budget profile",
        )
        require(task.get("observable_completion"), f"{task['task_id']} has no completion contract")
        require(task.get("forbidden_effects") is not None, f"{task['task_id']} has no forbidden effects")
    return len(tasks)


def main() -> int:
    check_required_files()
    collections = check_collection_index()
    research = check_research_data()
    python_files = check_python_syntax()
    check_ollama_contract()
    check_experiment_state()
    evaluation_tasks = check_evaluation_contract()
    print(
        json.dumps(
            {
                "ok": True,
                "collections": collections,
                "research": research,
                "evaluation_tasks": evaluation_tasks,
                "python_files_checked": python_files,
            },
            ensure_ascii=False,
            indent=2,
        )
    )
    return 0


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