command parity and migration guide

This commit is contained in:
2026-05-14 17:16:39 +02:00
parent 0753c32c1b
commit 3de72eb0d2
10 changed files with 552 additions and 16 deletions

View File

@@ -1,6 +1,7 @@
from __future__ import annotations
import argparse
from collections import Counter
import json
import sys
from pathlib import Path
@@ -8,7 +9,15 @@ from pathlib import Path
from .checks import run_collection_checks
from .engine import engine_capability_contract, plan_asset_sync, sync_assets
from .errors import InfospaceError
from .history import find_snapshot, get_history, metric_trend, record_check_results
from .history import (
build_viability_report,
find_snapshot,
get_history,
metric_trend,
read_metrics_file,
record_check_results,
)
from .inspection import export_mermaid, relationship_summary
from .lifecycle import add_artifact, create_infospace, load_infospace
from .markdown_adapter import validate_infospace_artifacts
from .semantics import list_entities, list_relations
@@ -28,6 +37,9 @@ def build_parser() -> argparse.ArgumentParser:
inspect = sub.add_parser("inspect", help="Inspect an infospace")
inspect.add_argument("root")
status = sub.add_parser("status", help="Summarize replacement readiness state")
status.add_argument("root")
add = sub.add_parser("add-artifact", help="Add an artifact to an infospace")
add.add_argument("root")
add.add_argument("source")
@@ -64,6 +76,19 @@ def build_parser() -> argparse.ArgumentParser:
)
metrics.add_argument("root")
check = sub.add_parser(
"check",
help="Run collection checks and persist metrics/history",
)
check.add_argument("root")
viability = sub.add_parser("viability", help="Evaluate configured thresholds")
viability.add_argument("root")
graph = sub.add_parser("graph", help="Export the artifact relationship graph")
graph.add_argument("root")
graph.add_argument("--format", choices=["json", "mermaid"], default="json")
workflow = sub.add_parser("workflow", help="Inspect, plan, and run workflows")
workflow_sub = workflow.add_subparsers(dest="workflow_command", required=True)
@@ -126,6 +151,8 @@ def main(argv: list[str] | None = None) -> int:
_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 == "status":
_write_json(_status_payload(Path(args.root)))
elif args.command == "add-artifact":
artifact = add_artifact(
Path(args.root),
@@ -191,12 +218,26 @@ def main(argv: list[str] | None = None) -> int:
)
_write_json({"diff": before.diff(after).to_dict()})
elif args.command == "metrics":
_write_json(_record_checks(Path(args.root)).to_dict())
elif args.command == "check":
infospace = load_infospace(Path(args.root))
result = record_check_results(
infospace.root,
run_collection_checks(infospace.artifacts),
report = run_collection_checks(infospace.artifacts)
result = record_check_results(infospace.root, report)
_write_json(
{
**result.to_dict(),
"details": report.details,
}
)
_write_json(result.to_dict())
elif args.command == "viability":
_write_json(build_viability_report(Path(args.root)))
elif args.command == "graph":
infospace = load_infospace(Path(args.root))
summary = relationship_summary(infospace.artifacts)
if args.format == "mermaid":
print(export_mermaid(summary), end="")
else:
_write_json(_relationship_summary_payload(summary))
elif args.command == "workflow":
if args.workflow_command == "inspect":
_write_json(
@@ -241,5 +282,67 @@ def main(argv: list[str] | None = None) -> int:
return 0
def _status_payload(root: Path) -> dict:
infospace = load_infospace(root)
validation = validate_infospace_artifacts(infospace.root)
entities = list_entities(infospace.root)
relations = list_relations(infospace.root)
history = get_history(infospace.root)
metrics = read_metrics_file(
infospace.root / "output" / "metrics" / "metrics.yaml"
)
viability = (
build_viability_report(infospace.root) if infospace.config.viability else None
)
graph = relationship_summary(infospace.artifacts)
return {
"slug": infospace.config.slug,
"name": infospace.config.name,
"topic": infospace.config.topic.to_dict(),
"artifact_count": len(infospace.artifacts),
"artifact_kinds": dict(
sorted(Counter(item.kind for item in infospace.artifacts).items())
),
"entity_count": len(entities),
"relation_count": len(relations),
"validation": {
"checked_artifact_count": len(validation),
"valid": all(result.valid for result in validation),
},
"history": {
"snapshot_count": len(history),
"latest_snapshot_id": history[-1].snapshot_id if history else "",
},
"metrics": metrics,
"viability": viability,
"graph": {
"node_count": graph.node_count,
"edge_count": graph.edge_count,
"relationship_types": graph.relationship_types,
},
}
def _record_checks(root: Path):
infospace = load_infospace(root)
return record_check_results(
infospace.root,
run_collection_checks(infospace.artifacts),
)
def _relationship_summary_payload(summary) -> dict:
return {
"node_count": summary.node_count,
"edge_count": summary.edge_count,
"nodes": summary.nodes,
"edges": [
{"source": edge.source, "target": edge.target, "type": edge.type}
for edge in summary.edges
],
"relationship_types": summary.relationship_types,
}
def _write_json(payload: dict) -> None:
print(json.dumps(payload, indent=2))