feat(infospace): add lifecycle CLI commands — init, status, entities, viability (S2.2)
Adds 'markitect infospace' command group with init (create config), status (entity count/domains/disciplines), entities (list with sort), and viability (threshold dashboard with pass/fail). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -7147,6 +7147,13 @@ try:
|
||||
except ImportError:
|
||||
pass # Helper module not available
|
||||
|
||||
# Register infospace commands
|
||||
try:
|
||||
from markitect.infospace.cli import infospace_commands
|
||||
cli.add_command(infospace_commands)
|
||||
except ImportError:
|
||||
pass # Infospace module not available
|
||||
|
||||
# Register proxy file system commands
|
||||
try:
|
||||
from markitect.proxy.cli import proxy_group
|
||||
|
||||
210
markitect/infospace/cli.py
Normal file
210
markitect/infospace/cli.py
Normal file
@@ -0,0 +1,210 @@
|
||||
"""
|
||||
CLI commands for infospace lifecycle management.
|
||||
|
||||
Provides ``markitect infospace`` subcommands for initialising,
|
||||
inspecting, and evaluating infospaces.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
import click
|
||||
|
||||
from markitect.infospace.config import (
|
||||
DisciplineBinding,
|
||||
InfospaceConfig,
|
||||
SchemaRegistry,
|
||||
TopicConfig,
|
||||
find_infospace_config,
|
||||
load_infospace_config,
|
||||
save_infospace_config,
|
||||
)
|
||||
from markitect.infospace.entity_parser import parse_entity_directory
|
||||
from markitect.infospace.state import build_state
|
||||
|
||||
|
||||
def _load_config_or_exit(config_path: Optional[str] = None) -> tuple:
|
||||
"""Resolve and load infospace.yaml, or exit with an error."""
|
||||
if config_path:
|
||||
p = Path(config_path)
|
||||
else:
|
||||
p = find_infospace_config()
|
||||
if p is None:
|
||||
click.echo("Error: No infospace.yaml found. Run 'markitect infospace init' first.", err=True)
|
||||
raise SystemExit(1)
|
||||
cfg = load_infospace_config(p)
|
||||
return cfg, p
|
||||
|
||||
|
||||
@click.group(name="infospace")
|
||||
def infospace_commands():
|
||||
"""Manage infospaces — create, inspect, evaluate."""
|
||||
pass
|
||||
|
||||
|
||||
# ── init ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@infospace_commands.command()
|
||||
@click.option("--topic", required=True, help="Topic name for the infospace.")
|
||||
@click.option("--domain", default="", help="Knowledge domain.")
|
||||
@click.option("--sources", default="", help="Path to source material directory.")
|
||||
@click.option("--discipline", multiple=True, help="Discipline name (repeatable).")
|
||||
@click.option("--output", "-o", default="infospace.yaml", help="Output config file path.")
|
||||
def init(topic: str, domain: str, sources: str, discipline: tuple, output: str):
|
||||
"""Initialise a new infospace configuration file."""
|
||||
out_path = Path(output)
|
||||
if out_path.exists():
|
||||
click.echo(f"Error: {out_path} already exists.", err=True)
|
||||
raise SystemExit(1)
|
||||
|
||||
disciplines = [DisciplineBinding(name=d) for d in discipline]
|
||||
config = InfospaceConfig(
|
||||
topic=TopicConfig(name=topic, domain=domain, sources=sources),
|
||||
disciplines=disciplines,
|
||||
)
|
||||
save_infospace_config(config, out_path)
|
||||
click.echo(f"Created {out_path}")
|
||||
|
||||
|
||||
# ── status ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@infospace_commands.command()
|
||||
@click.option("--config", "config_path", default=None, help="Path to infospace.yaml.")
|
||||
def status(config_path: Optional[str]):
|
||||
"""Show infospace status — entity count, domains, evaluation state."""
|
||||
cfg, cfg_path = _load_config_or_exit(config_path)
|
||||
root = cfg_path.parent
|
||||
|
||||
# Parse entities
|
||||
entities_dir = root / cfg.entities_dir
|
||||
entities = []
|
||||
if entities_dir.is_dir():
|
||||
entities = parse_entity_directory(entities_dir)
|
||||
|
||||
# Load latest snapshot if available
|
||||
snapshot = None
|
||||
history_path = root / cfg.metrics_dir / "history.yaml"
|
||||
if history_path.is_file():
|
||||
from markitect.infospace.evaluation_io import read_history
|
||||
history = read_history(history_path)
|
||||
if history:
|
||||
snapshot = history[-1]
|
||||
|
||||
state = build_state(cfg, entities=entities, snapshot=snapshot)
|
||||
|
||||
click.echo(f"Infospace: {state.topic_name}")
|
||||
if cfg.topic.domain:
|
||||
click.echo(f"Domain: {cfg.topic.domain}")
|
||||
click.echo(f"Entities: {state.entity_count}")
|
||||
if state.domains:
|
||||
click.echo(f"Domains: {', '.join(state.domains)}")
|
||||
if cfg.disciplines:
|
||||
names = [d.name for d in cfg.disciplines]
|
||||
click.echo(f"Disciplines: {', '.join(names)}")
|
||||
if state.has_evaluations:
|
||||
click.echo(f"Last evaluated: {state.latest_snapshot.created_at.isoformat()}")
|
||||
else:
|
||||
click.echo("Evaluations: none")
|
||||
|
||||
|
||||
# ── entities ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@infospace_commands.command()
|
||||
@click.option("--config", "config_path", default=None, help="Path to infospace.yaml.")
|
||||
@click.option(
|
||||
"--sort-by", "sort_key",
|
||||
type=click.Choice(["slug", "domain", "words"]),
|
||||
default="slug",
|
||||
help="Sort entities by field.",
|
||||
)
|
||||
def entities(config_path: Optional[str], sort_key: str):
|
||||
"""List entities with metadata summary."""
|
||||
cfg, cfg_path = _load_config_or_exit(config_path)
|
||||
root = cfg_path.parent
|
||||
entities_dir = root / cfg.entities_dir
|
||||
|
||||
if not entities_dir.is_dir():
|
||||
click.echo("No entities directory found.")
|
||||
return
|
||||
|
||||
entity_list = parse_entity_directory(entities_dir)
|
||||
if not entity_list:
|
||||
click.echo("No entities found.")
|
||||
return
|
||||
|
||||
# Sort
|
||||
if sort_key == "domain":
|
||||
entity_list.sort(key=lambda e: (e.domain or "", e.slug))
|
||||
elif sort_key == "words":
|
||||
entity_list.sort(key=lambda e: e.total_word_count, reverse=True)
|
||||
else:
|
||||
entity_list.sort(key=lambda e: e.slug)
|
||||
|
||||
# Format as table
|
||||
click.echo(f"{'Slug':<40} {'Domain':<20} {'Words':>6}")
|
||||
click.echo("-" * 68)
|
||||
for e in entity_list:
|
||||
click.echo(f"{e.slug:<40} {(e.domain or '-'):<20} {e.total_word_count:>6}")
|
||||
click.echo(f"\nTotal: {len(entity_list)} entities")
|
||||
|
||||
|
||||
# ── viability ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@infospace_commands.command()
|
||||
@click.option("--config", "config_path", default=None, help="Path to infospace.yaml.")
|
||||
def viability(config_path: Optional[str]):
|
||||
"""Show viability dashboard — threshold checks and pass/fail."""
|
||||
cfg, cfg_path = _load_config_or_exit(config_path)
|
||||
|
||||
if not cfg.viability:
|
||||
click.echo("No viability thresholds configured in infospace.yaml.")
|
||||
return
|
||||
|
||||
# Try to load latest metrics
|
||||
root = cfg_path.parent
|
||||
metrics: dict = {}
|
||||
metrics_file = root / cfg.metrics_dir / "metrics.yaml"
|
||||
if metrics_file.is_file():
|
||||
import yaml
|
||||
raw = yaml.safe_load(metrics_file.read_text(encoding="utf-8"))
|
||||
if isinstance(raw, dict):
|
||||
metrics = {k: float(v) for k, v in raw.items() if isinstance(v, (int, float))}
|
||||
|
||||
state = build_state(cfg, metrics=metrics if metrics else None)
|
||||
|
||||
if not state.viability_results:
|
||||
click.echo("No metrics available. Run evaluations first.")
|
||||
click.echo("\nConfigured thresholds:")
|
||||
for name, t in cfg.viability.items():
|
||||
bounds = []
|
||||
if t.min is not None:
|
||||
bounds.append(f"min={t.min}")
|
||||
if t.max is not None:
|
||||
bounds.append(f"max={t.max}")
|
||||
click.echo(f" {name}: {', '.join(bounds)}")
|
||||
return
|
||||
|
||||
click.echo(f"{'Metric':<30} {'Value':>8} {'Threshold':>15} {'Status':>8}")
|
||||
click.echo("-" * 63)
|
||||
for r in state.viability_results:
|
||||
bounds = []
|
||||
if r.threshold.min is not None:
|
||||
bounds.append(f"min={r.threshold.min}")
|
||||
if r.threshold.max is not None:
|
||||
bounds.append(f"max={r.threshold.max}")
|
||||
status_str = "PASS" if r.passed else "FAIL"
|
||||
click.echo(
|
||||
f"{r.metric:<30} {r.value:>8.4f} {', '.join(bounds):>15} {status_str:>8}"
|
||||
)
|
||||
|
||||
click.echo()
|
||||
if state.is_viable:
|
||||
click.echo(f"Viable: YES ({state.viability_pass_count}/{state.viability_total_count} thresholds met)")
|
||||
else:
|
||||
click.echo(f"Viable: NO ({state.viability_pass_count}/{state.viability_total_count} thresholds met)")
|
||||
225
tests/unit/infospace/test_cli.py
Normal file
225
tests/unit/infospace/test_cli.py
Normal file
@@ -0,0 +1,225 @@
|
||||
"""Tests for markitect.infospace.cli."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
from markitect.infospace.cli import infospace_commands
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def runner():
|
||||
return CliRunner()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def infospace_dir(tmp_path):
|
||||
"""Create a minimal infospace directory with config and entities."""
|
||||
config_yaml = """\
|
||||
topic:
|
||||
name: "Test Infospace"
|
||||
domain: "Testing"
|
||||
|
||||
disciplines:
|
||||
- name: "Test Discipline"
|
||||
|
||||
viability:
|
||||
coverage_ratio:
|
||||
min: 0.60
|
||||
redundancy_ratio:
|
||||
max: 0.05
|
||||
"""
|
||||
(tmp_path / "infospace.yaml").write_text(config_yaml)
|
||||
|
||||
entities = tmp_path / "output" / "entities"
|
||||
entities.mkdir(parents=True)
|
||||
(entities / "alpha.md").write_text(
|
||||
"# Alpha\n\n## Definition\n\nAlpha is a test entity.\n\n"
|
||||
"## Source Chapter\n\nChapter 1\n\n"
|
||||
"## Domain\n\nProduction\n"
|
||||
)
|
||||
(entities / "beta.md").write_text(
|
||||
"# Beta\n\n## Definition\n\nBeta is another test entity with more words "
|
||||
"to make it longer.\n\n"
|
||||
"## Source Chapter\n\nChapter 2\n\n"
|
||||
"## Domain\n\nDistribution\n"
|
||||
)
|
||||
return tmp_path
|
||||
|
||||
|
||||
# ── init ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestInitCommand:
|
||||
def test_creates_config_file(self, runner, tmp_path):
|
||||
out = tmp_path / "infospace.yaml"
|
||||
result = runner.invoke(
|
||||
infospace_commands,
|
||||
["init", "--topic", "My Topic", "--domain", "Science", "-o", str(out)],
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
assert out.exists()
|
||||
assert "Created" in result.output
|
||||
|
||||
def test_config_contains_topic(self, runner, tmp_path):
|
||||
out = tmp_path / "infospace.yaml"
|
||||
runner.invoke(
|
||||
infospace_commands,
|
||||
["init", "--topic", "My Topic", "-o", str(out)],
|
||||
)
|
||||
text = out.read_text()
|
||||
assert "My Topic" in text
|
||||
|
||||
def test_refuses_overwrite(self, runner, tmp_path):
|
||||
out = tmp_path / "infospace.yaml"
|
||||
out.write_text("existing")
|
||||
result = runner.invoke(
|
||||
infospace_commands,
|
||||
["init", "--topic", "X", "-o", str(out)],
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
assert "already exists" in result.output
|
||||
|
||||
def test_with_disciplines(self, runner, tmp_path):
|
||||
out = tmp_path / "infospace.yaml"
|
||||
result = runner.invoke(
|
||||
infospace_commands,
|
||||
[
|
||||
"init", "--topic", "T",
|
||||
"--discipline", "VSM",
|
||||
"--discipline", "Category Theory",
|
||||
"-o", str(out),
|
||||
],
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
text = out.read_text()
|
||||
assert "VSM" in text
|
||||
assert "Category Theory" in text
|
||||
|
||||
|
||||
# ── status ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestStatusCommand:
|
||||
def test_shows_topic_and_count(self, runner, infospace_dir):
|
||||
result = runner.invoke(
|
||||
infospace_commands,
|
||||
["status", "--config", str(infospace_dir / "infospace.yaml")],
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
assert "Test Infospace" in result.output
|
||||
assert "2" in result.output # 2 entities
|
||||
|
||||
def test_shows_domain_field(self, runner, infospace_dir):
|
||||
result = runner.invoke(
|
||||
infospace_commands,
|
||||
["status", "--config", str(infospace_dir / "infospace.yaml")],
|
||||
)
|
||||
# Domain from config (topic.domain), not entity domains
|
||||
assert "Testing" in result.output
|
||||
|
||||
def test_shows_disciplines(self, runner, infospace_dir):
|
||||
result = runner.invoke(
|
||||
infospace_commands,
|
||||
["status", "--config", str(infospace_dir / "infospace.yaml")],
|
||||
)
|
||||
assert "Test Discipline" in result.output
|
||||
|
||||
def test_no_config_exits(self, runner, tmp_path):
|
||||
result = runner.invoke(
|
||||
infospace_commands,
|
||||
["status", "--config", str(tmp_path / "nonexistent.yaml")],
|
||||
)
|
||||
assert result.exit_code != 0
|
||||
|
||||
|
||||
# ── entities ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestEntitiesCommand:
|
||||
def test_lists_entities(self, runner, infospace_dir):
|
||||
result = runner.invoke(
|
||||
infospace_commands,
|
||||
["entities", "--config", str(infospace_dir / "infospace.yaml")],
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
assert "alpha" in result.output
|
||||
assert "beta" in result.output
|
||||
assert "Total: 2" in result.output
|
||||
|
||||
def test_sort_by_domain(self, runner, infospace_dir):
|
||||
result = runner.invoke(
|
||||
infospace_commands,
|
||||
[
|
||||
"entities",
|
||||
"--config", str(infospace_dir / "infospace.yaml"),
|
||||
"--sort-by", "domain",
|
||||
],
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
lines = result.output.strip().split("\n")
|
||||
# Distribution comes before Production alphabetically
|
||||
data_lines = [l for l in lines if "alpha" in l or "beta" in l]
|
||||
assert len(data_lines) == 2
|
||||
|
||||
def test_no_entities_dir(self, runner, tmp_path):
|
||||
(tmp_path / "infospace.yaml").write_text("topic:\n name: X\n")
|
||||
result = runner.invoke(
|
||||
infospace_commands,
|
||||
["entities", "--config", str(tmp_path / "infospace.yaml")],
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
assert "No entities" in result.output
|
||||
|
||||
|
||||
# ── viability ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestViabilityCommand:
|
||||
def test_no_metrics_shows_thresholds(self, runner, infospace_dir):
|
||||
result = runner.invoke(
|
||||
infospace_commands,
|
||||
["viability", "--config", str(infospace_dir / "infospace.yaml")],
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
assert "coverage_ratio" in result.output
|
||||
|
||||
def test_with_metrics_file(self, runner, infospace_dir):
|
||||
import yaml
|
||||
metrics_dir = infospace_dir / "output" / "metrics"
|
||||
metrics_dir.mkdir(parents=True, exist_ok=True)
|
||||
metrics = {"coverage_ratio": 0.85, "redundancy_ratio": 0.02}
|
||||
(metrics_dir / "metrics.yaml").write_text(yaml.safe_dump(metrics))
|
||||
|
||||
result = runner.invoke(
|
||||
infospace_commands,
|
||||
["viability", "--config", str(infospace_dir / "infospace.yaml")],
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
assert "PASS" in result.output
|
||||
assert "Viable: YES" in result.output
|
||||
|
||||
def test_failing_threshold(self, runner, infospace_dir):
|
||||
import yaml
|
||||
metrics_dir = infospace_dir / "output" / "metrics"
|
||||
metrics_dir.mkdir(parents=True, exist_ok=True)
|
||||
metrics = {"coverage_ratio": 0.3, "redundancy_ratio": 0.02}
|
||||
(metrics_dir / "metrics.yaml").write_text(yaml.safe_dump(metrics))
|
||||
|
||||
result = runner.invoke(
|
||||
infospace_commands,
|
||||
["viability", "--config", str(infospace_dir / "infospace.yaml")],
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
assert "FAIL" in result.output
|
||||
assert "Viable: NO" in result.output
|
||||
|
||||
def test_no_thresholds_configured(self, runner, tmp_path):
|
||||
(tmp_path / "infospace.yaml").write_text("topic:\n name: X\n")
|
||||
result = runner.invoke(
|
||||
infospace_commands,
|
||||
["viability", "--config", str(tmp_path / "infospace.yaml")],
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
assert "No viability thresholds" in result.output
|
||||
Reference in New Issue
Block a user