feat(infospace): add composition model for discipline binding (S2.6)

Discipline resolution, viability checking, entity access, stale
mapping detection, and binding management. CLI commands: bind-discipline,
disciplines, stale-mappings.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-19 02:03:54 +01:00
parent ce7f78d57d
commit b76d6d38c1
3 changed files with 641 additions and 0 deletions

View File

@@ -419,3 +419,106 @@ def history_diff(date_a: str, date_b: str, config_path: Optional[str]):
diff = diff_snapshots(snap_a, snap_b)
click.echo(diff.summary())
# ── bind-discipline ─────────────────────────────────────────────────
@infospace_commands.command(name="bind-discipline")
@click.argument("discipline_path")
@click.option("--name", required=True, help="Name for the discipline.")
@click.option("--config", "config_path", default=None, help="Path to infospace.yaml.")
def bind_discipline_cmd(discipline_path: str, name: str, config_path: Optional[str]):
"""Bind a discipline infospace to the current infospace."""
cfg, cfg_path = _load_config_or_exit(config_path)
root = cfg_path.parent
from markitect.infospace.composition import bind_discipline
status = bind_discipline(cfg, name=name, path=discipline_path, root=root)
if status.error:
click.echo(f"Error: {status.error}", err=True)
raise SystemExit(1)
# Persist updated config
save_infospace_config(cfg, cfg_path)
click.echo(f"Bound discipline '{name}' from {discipline_path}")
click.echo(f" Entities: {status.entity_count}")
if status.has_config:
viable_str = "YES" if status.is_viable else "NO"
click.echo(f" Viable: {viable_str}")
# ── disciplines ─────────────────────────────────────────────────────
@infospace_commands.command()
@click.option("--config", "config_path", default=None, help="Path to infospace.yaml.")
def disciplines(config_path: Optional[str]):
"""List bound disciplines and their viability status."""
cfg, cfg_path = _load_config_or_exit(config_path)
root = cfg_path.parent
if not cfg.disciplines:
click.echo("No disciplines bound.")
return
from markitect.infospace.composition import check_discipline_status
click.echo(f"{'Name':<30} {'Entities':>8} {'Viable':>8} {'Path'}")
click.echo("-" * 70)
for binding in cfg.disciplines:
status = check_discipline_status(binding, root)
viable_str = "YES" if status.is_viable else ("NO" if status.has_config else "?")
click.echo(
f"{status.name:<30} {status.entity_count:>8} {viable_str:>8} {status.path}"
)
if status.error:
click.echo(f" Error: {status.error}")
# ── stale-mappings ──────────────────────────────────────────────────
@infospace_commands.command(name="stale-mappings")
@click.option("--config", "config_path", default=None, help="Path to infospace.yaml.")
def stale_mappings(config_path: Optional[str]):
"""Check for stale mappings due to discipline changes."""
cfg, cfg_path = _load_config_or_exit(config_path)
root = cfg_path.parent
if not cfg.disciplines:
click.echo("No disciplines bound — no mappings to check.")
return
from markitect.infospace.composition import find_stale_mappings
# Try to load mapping references from output
mapping_refs = _load_mapping_references(cfg, root)
stale = find_stale_mappings(cfg, root, mapping_references=mapping_refs)
if not stale:
click.echo("No stale mappings detected.")
return
click.echo(f"Found {len(stale)} stale mapping(s):\n")
for s in stale:
click.echo(f" {s.entity_slug} -> {s.discipline_entity}")
click.echo(f" {s.reason}")
def _load_mapping_references(
cfg: InfospaceConfig, root: Path
) -> Optional[dict]:
"""Try to load mapping references from YAML file in output dir."""
mapping_file = root / cfg.metrics_dir / "mapping-references.yaml"
if not mapping_file.is_file():
return None
import yaml
data = yaml.safe_load(mapping_file.read_text(encoding="utf-8"))
if isinstance(data, dict):
return data
return None

View File

@@ -0,0 +1,281 @@
"""
Infospace composition model.
Allows one infospace to use another as a discipline — a reusable
framework of concepts applied as an analytical lens.
Key operations:
- Resolve and validate discipline bindings
- Check discipline viability (must meet its own thresholds)
- List discipline entities as mapping targets
- Detect stale mappings when discipline content changes
"""
from __future__ import annotations
import hashlib
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional
from markitect.infospace.config import (
DisciplineBinding,
InfospaceConfig,
load_infospace_config,
)
from markitect.infospace.entity_parser import parse_entity_directory
from markitect.infospace.history import get_latest_snapshot, read_metrics_file
from markitect.infospace.models import EntityMeta
from markitect.infospace.state import InfospaceState, ViabilityResult, build_state
@dataclass
class DisciplineStatus:
"""Status of a bound discipline infospace."""
name: str
path: str
resolved_path: Optional[Path] = None
exists: bool = False
has_config: bool = False
entity_count: int = 0
is_viable: bool = False
viability_results: List[ViabilityResult] = field(default_factory=list)
error: str = ""
def to_dict(self) -> Dict[str, Any]:
d: Dict[str, Any] = {
"name": self.name,
"path": self.path,
"exists": self.exists,
"has_config": self.has_config,
"entity_count": self.entity_count,
"is_viable": self.is_viable,
}
if self.viability_results:
d["viability"] = [r.to_dict() for r in self.viability_results]
if self.error:
d["error"] = self.error
return d
@dataclass
class StaleMappingInfo:
"""Information about a mapping that may be stale."""
entity_slug: str
discipline_entity: str
reason: str
def to_dict(self) -> Dict[str, Any]:
return {
"entity_slug": self.entity_slug,
"discipline_entity": self.discipline_entity,
"reason": self.reason,
}
# ── Resolution ───────────────────────────────────────────────────────
def resolve_discipline_path(
binding: DisciplineBinding, root: Path
) -> Optional[Path]:
"""Resolve a discipline binding to an absolute path.
Tries the binding's path relative to *root*, then as an absolute path.
Returns ``None`` if the directory doesn't exist.
"""
if not binding.path:
return None
# Try relative to root first
candidate = root / binding.path
if candidate.is_dir():
return candidate.resolve()
# Try as absolute
candidate = Path(binding.path)
if candidate.is_dir():
return candidate.resolve()
return None
def load_discipline_config(
binding: DisciplineBinding, root: Path
) -> Optional[InfospaceConfig]:
"""Load the infospace config for a bound discipline.
Returns ``None`` if the discipline path cannot be resolved or
has no ``infospace.yaml``.
"""
disc_path = resolve_discipline_path(binding, root)
if disc_path is None:
return None
config_file = disc_path / "infospace.yaml"
if not config_file.is_file():
return None
return load_infospace_config(config_file)
# ── Viability checking ───────────────────────────────────────────────
def check_discipline_status(
binding: DisciplineBinding, root: Path
) -> DisciplineStatus:
"""Check the full status of a bound discipline.
Resolves the path, loads config, counts entities, and checks
viability against the discipline's own thresholds.
"""
status = DisciplineStatus(name=binding.name, path=binding.path)
disc_path = resolve_discipline_path(binding, root)
if disc_path is None:
status.error = f"Path not found: {binding.path}"
return status
status.resolved_path = disc_path
status.exists = True
# Load config
config_file = disc_path / "infospace.yaml"
if not config_file.is_file():
status.error = "No infospace.yaml found"
return status
disc_config = load_infospace_config(config_file)
status.has_config = True
# Count entities
entities_dir = disc_path / disc_config.entities_dir
if entities_dir.is_dir():
entities = parse_entity_directory(entities_dir)
status.entity_count = len(entities)
# Check viability
if disc_config.viability:
metrics = read_metrics_file(disc_path / disc_config.metrics_dir / "metrics.yaml")
if metrics:
state = build_state(disc_config, metrics=metrics)
status.viability_results = state.viability_results
status.is_viable = state.is_viable
return status
def get_discipline_entities(
binding: DisciplineBinding, root: Path
) -> List[EntityMeta]:
"""Get all entities from a bound discipline infospace."""
disc_path = resolve_discipline_path(binding, root)
if disc_path is None:
return []
disc_config = load_discipline_config(binding, root)
if disc_config is None:
return []
entities_dir = disc_path / disc_config.entities_dir
if not entities_dir.is_dir():
return []
return parse_entity_directory(entities_dir)
# ── Stale mapping detection ─────────────────────────────────────────
def _content_digest(entity: EntityMeta) -> str:
"""Compute a short content digest for an entity."""
content = f"{entity.slug}|{entity.definition}|{entity.domain}"
return hashlib.sha256(content.encode()).hexdigest()[:12]
def compute_discipline_digests(
binding: DisciplineBinding, root: Path
) -> Dict[str, str]:
"""Compute content digests for all entities in a discipline.
Returns ``{slug: digest}`` mapping.
"""
entities = get_discipline_entities(binding, root)
return {e.slug: _content_digest(e) for e in entities}
def find_stale_mappings(
config: InfospaceConfig,
root: Path,
mapping_references: Optional[Dict[str, List[str]]] = None,
) -> List[StaleMappingInfo]:
"""Find mappings that may be stale due to discipline changes.
Args:
config: The infospace configuration.
root: Project root directory.
mapping_references: ``{entity_slug: [discipline_entity_slugs]}``
mapping of local entities to the discipline entities they
reference. If ``None``, returns an empty list (no mapping
data available).
Returns:
List of stale mapping info objects.
"""
if not mapping_references:
return []
stale: List[StaleMappingInfo] = []
for binding in config.disciplines:
disc_entities = get_discipline_entities(binding, root)
disc_slugs = {e.slug for e in disc_entities}
for entity_slug, refs in mapping_references.items():
for ref_slug in refs:
if ref_slug not in disc_slugs:
stale.append(StaleMappingInfo(
entity_slug=entity_slug,
discipline_entity=ref_slug,
reason=f"Discipline entity '{ref_slug}' no longer exists in '{binding.name}'",
))
return stale
# ── Binding management ───────────────────────────────────────────────
def bind_discipline(
config: InfospaceConfig,
name: str,
path: str,
root: Path,
) -> DisciplineStatus:
"""Add a discipline binding to the config and validate it.
Does NOT persist the config — the caller should save it.
Args:
config: The infospace configuration to update.
name: Discipline name.
path: Path to the discipline infospace.
root: Project root for path resolution.
Returns:
Status of the newly bound discipline.
"""
# Check for duplicates
existing = {d.name for d in config.disciplines}
if name in existing:
return DisciplineStatus(
name=name, path=path, error=f"Discipline '{name}' already bound"
)
binding = DisciplineBinding(name=name, path=path)
config.disciplines.append(binding)
return check_discipline_status(binding, root)

View File

@@ -0,0 +1,257 @@
"""
Tests for infospace composition model (S2.6).
"""
from __future__ import annotations
from pathlib import Path
import pytest
import yaml
from markitect.infospace.composition import (
DisciplineStatus,
StaleMappingInfo,
bind_discipline,
check_discipline_status,
compute_discipline_digests,
find_stale_mappings,
get_discipline_entities,
load_discipline_config,
resolve_discipline_path,
)
from markitect.infospace.config import (
DisciplineBinding,
InfospaceConfig,
TopicConfig,
ViabilityThreshold,
save_infospace_config,
)
# ── helpers ──────────────────────────────────────────────────────────
def _create_discipline(tmp_path: Path, name: str = "test-discipline") -> Path:
"""Create a minimal discipline infospace directory."""
disc_dir = tmp_path / name
disc_dir.mkdir(parents=True, exist_ok=True)
disc_config = InfospaceConfig(
topic=TopicConfig(name=name.replace("-", " ").title(), domain="Testing"),
viability={"coverage_ratio": ViabilityThreshold(metric="coverage_ratio", min=0.5)},
)
save_infospace_config(disc_config, disc_dir / "infospace.yaml")
# Create some entities
entities_dir = disc_dir / "output" / "entities"
entities_dir.mkdir(parents=True, exist_ok=True)
for slug in ["concept_a", "concept_b", "concept-c"]:
title = slug.replace("-", " ").title()
(entities_dir / f"{slug}.md").write_text(
f"# {title}\n\n## Definition\n\nA test concept for {slug}.\n\n"
f"## Source Chapter\n\nch01\n\n## Domain\n\nTesting\n",
encoding="utf-8",
)
return disc_dir
def _parent_config(tmp_path: Path, disc_path: str = "") -> InfospaceConfig:
"""Create a parent infospace config."""
return InfospaceConfig(
topic=TopicConfig(name="Parent", domain="Testing"),
disciplines=[DisciplineBinding(name="Test Discipline", path=disc_path)]
if disc_path
else [],
)
# ── resolve_discipline_path ─────────────────────────────────────────
class TestResolveDisciplinePath:
def test_relative_path(self, tmp_path):
disc_dir = _create_discipline(tmp_path)
binding = DisciplineBinding(name="test", path="test-discipline")
result = resolve_discipline_path(binding, tmp_path)
assert result is not None
assert result == disc_dir.resolve()
def test_absolute_path(self, tmp_path):
disc_dir = _create_discipline(tmp_path)
binding = DisciplineBinding(name="test", path=str(disc_dir))
result = resolve_discipline_path(binding, tmp_path / "other")
assert result is not None
assert result == disc_dir.resolve()
def test_missing_path(self, tmp_path):
binding = DisciplineBinding(name="test", path="nonexistent")
assert resolve_discipline_path(binding, tmp_path) is None
def test_empty_path(self, tmp_path):
binding = DisciplineBinding(name="test", path="")
assert resolve_discipline_path(binding, tmp_path) is None
# ── load_discipline_config ──────────────────────────────────────────
class TestLoadDisciplineConfig:
def test_loads_config(self, tmp_path):
disc_dir = _create_discipline(tmp_path)
binding = DisciplineBinding(name="test", path="test-discipline")
config = load_discipline_config(binding, tmp_path)
assert config is not None
assert config.topic.domain == "Testing"
def test_missing_config_file(self, tmp_path):
(tmp_path / "no-config").mkdir()
binding = DisciplineBinding(name="test", path="no-config")
assert load_discipline_config(binding, tmp_path) is None
def test_missing_directory(self, tmp_path):
binding = DisciplineBinding(name="test", path="gone")
assert load_discipline_config(binding, tmp_path) is None
# ── check_discipline_status ─────────────────────────────────────────
class TestCheckDisciplineStatus:
def test_valid_discipline(self, tmp_path):
_create_discipline(tmp_path)
binding = DisciplineBinding(name="Test Discipline", path="test-discipline")
status = check_discipline_status(binding, tmp_path)
assert status.exists
assert status.has_config
assert status.entity_count == 3
assert status.error == ""
def test_missing_discipline(self, tmp_path):
binding = DisciplineBinding(name="Missing", path="nope")
status = check_discipline_status(binding, tmp_path)
assert not status.exists
assert "not found" in status.error.lower()
def test_no_config(self, tmp_path):
(tmp_path / "bare").mkdir()
binding = DisciplineBinding(name="Bare", path="bare")
status = check_discipline_status(binding, tmp_path)
assert status.exists
assert not status.has_config
def test_viable_with_metrics(self, tmp_path):
disc_dir = _create_discipline(tmp_path)
# Write metrics that meet the threshold
metrics_dir = disc_dir / "output" / "metrics"
metrics_dir.mkdir(parents=True, exist_ok=True)
(metrics_dir / "metrics.yaml").write_text(
yaml.safe_dump({"coverage_ratio": 0.8}), encoding="utf-8"
)
binding = DisciplineBinding(name="Test", path="test-discipline")
status = check_discipline_status(binding, tmp_path)
assert status.is_viable
def test_not_viable_below_threshold(self, tmp_path):
disc_dir = _create_discipline(tmp_path)
metrics_dir = disc_dir / "output" / "metrics"
metrics_dir.mkdir(parents=True, exist_ok=True)
(metrics_dir / "metrics.yaml").write_text(
yaml.safe_dump({"coverage_ratio": 0.2}), encoding="utf-8"
)
binding = DisciplineBinding(name="Test", path="test-discipline")
status = check_discipline_status(binding, tmp_path)
assert not status.is_viable
def test_to_dict(self, tmp_path):
_create_discipline(tmp_path)
binding = DisciplineBinding(name="Test", path="test-discipline")
status = check_discipline_status(binding, tmp_path)
d = status.to_dict()
assert d["name"] == "Test"
assert d["exists"] is True
assert d["entity_count"] == 3
# ── get_discipline_entities ─────────────────────────────────────────
class TestGetDisciplineEntities:
def test_returns_entities(self, tmp_path):
_create_discipline(tmp_path)
binding = DisciplineBinding(name="Test", path="test-discipline")
entities = get_discipline_entities(binding, tmp_path)
assert len(entities) == 3
slugs = {e.slug for e in entities}
assert "concept_a" in slugs
def test_missing_discipline(self, tmp_path):
binding = DisciplineBinding(name="Test", path="nope")
assert get_discipline_entities(binding, tmp_path) == []
# ── compute_discipline_digests ──────────────────────────────────────
class TestComputeDisciplineDigests:
def test_returns_digests(self, tmp_path):
_create_discipline(tmp_path)
binding = DisciplineBinding(name="Test", path="test-discipline")
digests = compute_discipline_digests(binding, tmp_path)
assert len(digests) == 3
assert "concept_a" in digests
assert isinstance(digests["concept_a"], str)
assert len(digests["concept_a"]) == 12
# ── find_stale_mappings ─────────────────────────────────────────────
class TestFindStaleMappings:
def test_no_references(self, tmp_path):
cfg = _parent_config(tmp_path, disc_path="test-discipline")
assert find_stale_mappings(cfg, tmp_path) == []
def test_no_stale(self, tmp_path):
_create_discipline(tmp_path)
cfg = _parent_config(tmp_path, disc_path="test-discipline")
refs = {"entity_x": ["concept_a", "concept_b"]}
stale = find_stale_mappings(cfg, tmp_path, mapping_references=refs)
assert stale == []
def test_detects_stale(self, tmp_path):
_create_discipline(tmp_path)
cfg = _parent_config(tmp_path, disc_path="test-discipline")
refs = {"entity_x": ["concept_a", "deleted_concept"]}
stale = find_stale_mappings(cfg, tmp_path, mapping_references=refs)
assert len(stale) == 1
assert stale[0].entity_slug == "entity_x"
assert stale[0].discipline_entity == "deleted_concept"
def test_stale_to_dict(self):
info = StaleMappingInfo(
entity_slug="e1", discipline_entity="d1", reason="gone"
)
d = info.to_dict()
assert d["entity_slug"] == "e1"
# ── bind_discipline ─────────────────────────────────────────────────
class TestBindDiscipline:
def test_adds_binding(self, tmp_path):
_create_discipline(tmp_path)
cfg = InfospaceConfig(topic=TopicConfig(name="Parent"))
status = bind_discipline(cfg, name="Test", path="test-discipline", root=tmp_path)
assert status.exists
assert len(cfg.disciplines) == 1
assert cfg.disciplines[0].name == "Test"
def test_duplicate_rejected(self, tmp_path):
_create_discipline(tmp_path)
cfg = _parent_config(tmp_path, disc_path="test-discipline")
status = bind_discipline(cfg, name="Test Discipline", path="x", root=tmp_path)
assert "already bound" in status.error