infospace-bench/src/infospace_bench/cli.py

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

from .checks import run_collection_checks
from .errors import InfospaceError
from .history import find_snapshot, get_history, metric_trend, record_check_results
from .lifecycle import add_artifact, create_infospace, load_infospace
from .markdown_adapter import validate_infospace_artifacts
from .semantics import list_entities, list_relations


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="infospace-bench")
    sub = parser.add_subparsers(dest="command", required=True)

    create = sub.add_parser("create", help="Create an infospace")
    create.add_argument("workspace")
    create.add_argument("slug")
    create.add_argument("--name", required=True)
    create.add_argument("--topic-domain", default="")

    inspect = sub.add_parser("inspect", help="Inspect an infospace")
    inspect.add_argument("root")

    add = sub.add_parser("add-artifact", help="Add an artifact to an infospace")
    add.add_argument("root")
    add.add_argument("source")
    add.add_argument("--kind", required=True)
    add.add_argument("--title", default="")

    export = sub.add_parser("export", help="Print the infospace representation")
    export.add_argument("root")

    validate = sub.add_parser("validate", help="Validate infospace artifacts")
    validate.add_argument("root")

    entities = sub.add_parser("entities", help="List parsed entity artifacts")
    entities.add_argument("root")

    relations = sub.add_parser("relations", help="List parsed relation artifacts")
    relations.add_argument("root")

    history = sub.add_parser("history", help="List evaluation snapshot history")
    history.add_argument("root")
    history.add_argument("--metric", default="")

    history_diff = sub.add_parser(
        "history-diff",
        help="Diff two evaluation snapshots by snapshot ID or date",
    )
    history_diff.add_argument("root")
    history_diff.add_argument("before")
    history_diff.add_argument("after")

    metrics = sub.add_parser(
        "metrics",
        help="Run collection checks and persist metrics/history",
    )
    metrics.add_argument("root")

    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        if args.command == "create":
            infospace = create_infospace(
                Path(args.workspace),
                args.slug,
                name=args.name,
                topic_domain=args.topic_domain,
            )
            _write_json({"slug": infospace.config.slug, "root": str(infospace.root)})
        elif args.command == "inspect":
            _write_json(load_infospace(Path(args.root)).to_dict())
        elif args.command == "add-artifact":
            artifact = add_artifact(
                Path(args.root),
                Path(args.source),
                kind=args.kind,
                title=args.title,
            )
            _write_json({"artifact": artifact.to_dict()})
        elif args.command == "export":
            _write_json(load_infospace(Path(args.root)).to_dict())
        elif args.command == "validate":
            results = validate_infospace_artifacts(Path(args.root))
            valid = all(result.valid for result in results)
            _write_json(
                {
                    "valid": valid,
                    "results": [result.to_dict() for result in results],
                }
            )
            return 0 if valid else 1
        elif args.command == "entities":
            _write_json(
                {
                    "entities": [
                        entity.to_dict() for entity in list_entities(Path(args.root))
                    ]
                }
            )
        elif args.command == "relations":
            _write_json(
                {
                    "relations": [
                        relation.to_dict()
                        for relation in list_relations(Path(args.root))
                    ]
                }
            )
        elif args.command == "history":
            history = get_history(Path(args.root))
            if args.metric:
                _write_json(
                    {
                        "metric": args.metric,
                        "trend": metric_trend(history, args.metric),
                    }
                )
            else:
                _write_json({"history": [item.to_dict() for item in history]})
        elif args.command == "history-diff":
            history = get_history(Path(args.root))
            before = find_snapshot(history, args.before)
            after = find_snapshot(history, args.after)
            if before is None or after is None:
                missing = []
                if before is None:
                    missing.append(args.before)
                if after is None:
                    missing.append(args.after)
                raise InfospaceError(
                    "missing_snapshot",
                    "Could not resolve requested snapshot reference",
                    {"missing_refs": missing},
                )
            _write_json({"diff": before.diff(after).to_dict()})
        elif args.command == "metrics":
            infospace = load_infospace(Path(args.root))
            result = record_check_results(
                infospace.root,
                run_collection_checks(infospace.artifacts),
            )
            _write_json(result.to_dict())
        else:
            parser.error(f"Unhandled command: {args.command}")
    except InfospaceError as exc:
        print(json.dumps(exc.to_dict(), indent=2), file=sys.stderr)
        return 2
    return 0


def _write_json(payload: dict) -> None:
    print(json.dumps(payload, indent=2))