feat(infospace): add L3 relation graph with VSM-aware triplets (S2.8)

Implements the L3 relation graph layer — a directed graph of (Subject,
Predicate, Object) triplets annotated with VSM channel codes and feedback
roles. Triplets are authored as markdown files under output/relations/,
parsed into RelationMeta dataclasses, and analysed with networkx.

New modules:
- markitect/infospace/relation_models.py — RelationMeta dataclass +
  RELATION_TYPES controlled vocabulary (15 relation classes → VSM codes)
- markitect/infospace/relation_parser.py — parse_relation_file() and
  parse_relations_directory()

New schema: examples/infospace-with-history/schemas/relation-schema-v1.0.md
  — file naming convention, required sections, controlled vocabulary table

15 seed relation files covering the three core WoN feedback loops:
  - Capital Accumulation loop (positive reinforcement, S1/S3)
  - Market Price Balancing loop (negative feedback, S2/S3)
  - Market Extent mutual dependency (S1/S2)
  Plus structural relations: wages regulation, rent residual, price
  decomposition, invisible hand coordination

CLI: markitect infospace relations [--entity SLUG] [--vsm FILTER]
     [--loops] [--stats]
  - Builds directed graph from parsed files
  - Detects feedback loops via nx.simple_cycles()
  - 6 loops found from 15 seed relations (3 intended + 3 emergent)
  - --stats aggregates by VSM system code (strips parentheticals)

Config: InfospaceConfig gains relations_dir (default output/relations)
infospace.yaml: schemas.relation references relation-schema-v1.0.md

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-23 06:04:28 +01:00
parent fa27572f43
commit 2d45425b25
21 changed files with 1058 additions and 0 deletions

View File

@@ -299,6 +299,126 @@ def eval_summary(config_path: Optional[str], update_metrics: bool):
click.echo(f"\nUpdated metrics.yaml: per_entity_mean = {mean_overall:.4f}")
# ── relations ─────────────────────────────────────────────────────────
@infospace_commands.command()
@click.option("--config", "config_path", default=None, help="Path to infospace.yaml.")
@click.option("--entity", "entity_slug", default=None,
help="Show only relations involving this entity slug.")
@click.option("--vsm", "vsm_filter", default=None,
help="Show only relations whose VSM channel contains this string (e.g. S2, S3).")
@click.option("--loops", "loops_only", is_flag=True, default=False,
help="Show only feedback loops (cycles in the relation graph).")
@click.option("--stats", "stats_only", is_flag=True, default=False,
help="Show aggregate statistics only, no individual relations.")
def relations(config_path: Optional[str], entity_slug: Optional[str],
vsm_filter: Optional[str], loops_only: bool, stats_only: bool):
"""Show the L3 relation graph — triplets, feedback loops, and VSM channels."""
cfg, cfg_path = _load_config_or_exit(config_path)
root = cfg_path.parent
from markitect.infospace.relation_parser import parse_relations_directory
relations_dir = root / cfg.relations_dir
if not relations_dir.is_dir():
click.echo("No relations directory found. Create output/relations/ and add relation files.")
return
all_relations = parse_relations_directory(relations_dir)
if not all_relations:
click.echo("No relation files found in " + str(relations_dir))
return
# Build directed graph for cycle detection
try:
import networkx as nx
G = nx.DiGraph()
for r in all_relations:
G.add_edge(r.subject_slug, r.object_slug,
predicate=r.predicate,
relation_type=r.relation_type,
vsm_channel=r.vsm_channel,
slug=r.slug)
except ImportError:
G = None
# Find feedback loops
loops = []
if G is not None:
try:
loops = list(nx.simple_cycles(G))
except Exception:
loops = []
# Stats summary
import re as _re
def _vsm_code(channel: str) -> str:
"""Strip parenthetical description, returning just the system code (e.g. 'S3 → S1')."""
return _re.sub(r'\s*\(.*', '', channel).strip() or channel
n = len(all_relations)
vsm_counts: dict = {}
type_counts: dict = {}
for r in all_relations:
vsm_counts[_vsm_code(r.vsm_channel)] = vsm_counts.get(_vsm_code(r.vsm_channel), 0) + 1
type_counts[r.relation_type] = type_counts.get(r.relation_type, 0) + 1
click.echo(f"Relation graph — {n} relations")
if G is not None:
click.echo(f" Entities in graph: {G.number_of_nodes()}")
click.echo(f" Feedback loops: {len(loops)}")
click.echo()
if stats_only:
click.echo("Relation types:")
for rt, count in sorted(type_counts.items(), key=lambda x: -x[1]):
click.echo(f" {rt:<25} {count:>4}")
click.echo()
click.echo("VSM channels:")
for ch, count in sorted(vsm_counts.items(), key=lambda x: -x[1]):
click.echo(f" {ch:<20} {count:>4}")
return
# Feedback loops section
if loops or loops_only:
if loops:
click.echo(f"Feedback loops ({len(loops)}):")
for i, cycle in enumerate(loops, 1):
click.echo(f" Loop {i}: {''.join(cycle)}{cycle[0]}")
click.echo()
elif loops_only:
click.echo("No feedback loops detected in current relation set.")
return
if loops_only:
return
# Filter relations
filtered = all_relations
if entity_slug:
filtered = [r for r in filtered
if entity_slug in (r.subject_slug, r.object_slug)]
if not filtered:
click.echo(f"No relations found involving '{entity_slug}'.")
return
if vsm_filter:
filtered = [r for r in filtered if vsm_filter in r.vsm_channel]
if not filtered:
click.echo(f"No relations with VSM channel containing '{vsm_filter}'.")
return
# Display relations
click.echo(f"{'Subject':<35} {'Predicate':<30} {'Object':<35} {'VSM'}")
click.echo("-" * 110)
for r in filtered:
subj = r.subject[:33] + ".." if len(r.subject) > 35 else r.subject
obj = r.object[:33] + ".." if len(r.object) > 35 else r.object
pred = r.predicate[:28] + ".." if len(r.predicate) > 30 else r.predicate
click.echo(f"{subj:<35} {pred:<30} {obj:<35} {r.vsm_channel}")
# ── viability ────────────────────────────────────────────────────────

View File

@@ -254,6 +254,7 @@ class InfospaceConfig:
entities_dir: str = "output/entities"
evaluations_dir: str = "output/evaluations"
metrics_dir: str = "output/metrics"
relations_dir: str = "output/relations"
def to_dict(self) -> Dict[str, Any]:
d: Dict[str, Any] = {"topic": self.topic.to_dict()}
@@ -276,6 +277,8 @@ class InfospaceConfig:
d["evaluations_dir"] = self.evaluations_dir
if self.metrics_dir != "output/metrics":
d["metrics_dir"] = self.metrics_dir
if self.relations_dir != "output/relations":
d["relations_dir"] = self.relations_dir
return d
@classmethod
@@ -301,6 +304,7 @@ class InfospaceConfig:
entities_dir=data.get("entities_dir", "output/entities"),
evaluations_dir=data.get("evaluations_dir", "output/evaluations"),
metrics_dir=data.get("metrics_dir", "output/metrics"),
relations_dir=data.get("relations_dir", "output/relations"),
)

View File

@@ -0,0 +1,72 @@
"""
Data models for L3 relation triplets.
A relation triplet is the fundamental unit of the relation graph:
Subject --[Predicate]--> Object
Each triplet is stored as a markdown file in ``output/relations/``.
"""
from dataclasses import dataclass, field
from typing import List, Optional
# Controlled relation vocabulary — maps semantic class to VSM channel
RELATION_TYPES = {
"enables": "S1 → S1",
"constrains": "S1 ← S1",
"regulates": "S3 → S1",
"is regulated by": "S1 ← S3",
"coordinates": "S2",
"produces": "S1",
"consumes": "S1",
"monitors": "S3*",
"audits": "S3*",
"adapts to": "S4",
"anticipates": "S4",
"defines": "S5 → any",
"is defined by": "any ← S5",
"contradicts": "any",
"tensions with": "any",
}
@dataclass
class RelationMeta:
"""Structured metadata for a single relation triplet.
Attributes:
slug: Unique identifier, e.g.
``division_of_labour--constrains--market_extent``
subject: Human-readable title of the subject entity
subject_slug: Slug of the subject entity (links to L1)
predicate: Human-readable predicate phrase, e.g. "limited by"
object: Human-readable title of the object entity
object_slug: Slug of the object entity (links to L1)
relation_type: Semantic class from the controlled vocabulary
vsm_channel: VSM systems involved, e.g. "S1 → S2"
evidence: Source text quote or chapter reference
feedback_role: Description of role in a feedback loop (if any)
source_path: Absolute path to the ``.md`` file
"""
slug: str
subject: str
subject_slug: str
predicate: str
object: str
object_slug: str
relation_type: str
vsm_channel: str
evidence: str = ""
feedback_role: str = ""
source_path: str = ""
@property
def is_feedback_member(self) -> bool:
"""True if this relation participates in a named feedback loop."""
return bool(self.feedback_role.strip())
def edge(self) -> tuple:
"""Return a (subject_slug, object_slug, predicate) edge tuple."""
return (self.subject_slug, self.object_slug, self.predicate)

View File

@@ -0,0 +1,137 @@
"""
Relation triplet parser.
Reads structured :class:`RelationMeta` objects from relation markdown
files in ``output/relations/``.
File format::
# Subject — predicate — Object
## Subject
Subject Entity Title
## Predicate
predicate phrase
## Object
Object Entity Title
## Relation Type
constrains
## VSM Channel
S1 → S2
## Evidence
Book I, Chapter 3: "..."
## Feedback Role
Part of the Market Expansion loop: ...
"""
from __future__ import annotations
import logging
import re
from pathlib import Path
from typing import List, Optional, Sequence
from markitect.core.parser import parse_markdown_to_ast
from markitect.core.section_tree import (
build_section_tree,
extract_section_text,
slugify,
)
from .relation_models import RelationMeta
logger = logging.getLogger(__name__)
def _find_h2(tree_root: dict, slug: str) -> Optional[dict]:
for child in tree_root.get("children", []):
if child["level"] == 2 and child["slug"] == slug:
return child
return None
def _section_text(root: dict, slug: str) -> str:
node = _find_h2(root, slug)
return extract_section_text(node).strip() if node else ""
def _slug_from_title(title: str) -> str:
"""Convert entity title to slug (same as slugify used in entity_parser)."""
return slugify(title)
def parse_relation_file(path: Path) -> RelationMeta:
"""Parse a single relation markdown file into :class:`RelationMeta`.
Raises:
ValueError: If required sections are missing.
"""
content = path.read_text(encoding="utf-8")
tokens = parse_markdown_to_ast(content)
tree = build_section_tree(tokens)
# Find H1
h1 = next(
(c for c in tree["children"] if c["level"] == 1),
None,
)
if h1 is None:
raise ValueError(f"No H1 heading in {path}")
root = h1
subject = _section_text(root, "subject")
predicate = _section_text(root, "predicate")
obj = _section_text(root, "object")
relation_type = _section_text(root, "relation_type")
vsm_channel = _section_text(root, "vsm_channel")
evidence = _section_text(root, "evidence")
feedback_role = _section_text(root, "feedback_role")
if not subject:
raise ValueError(f"Missing ## Subject in {path}")
if not predicate:
raise ValueError(f"Missing ## Predicate in {path}")
if not obj:
raise ValueError(f"Missing ## Object in {path}")
subject_slug = _slug_from_title(subject)
object_slug = _slug_from_title(obj)
# Derive canonical slug from file stem
slug = path.stem
return RelationMeta(
slug=slug,
subject=subject,
subject_slug=subject_slug,
predicate=predicate,
object=obj,
object_slug=object_slug,
relation_type=relation_type,
vsm_channel=vsm_channel,
evidence=evidence,
feedback_role=feedback_role,
source_path=str(path),
)
def parse_relations_directory(
directory: Path,
) -> List[RelationMeta]:
"""Parse all relation files in *directory*.
Malformed files are skipped with a warning.
"""
relations: List[RelationMeta] = []
for md_file in sorted(directory.glob("*.md")):
try:
relations.append(parse_relation_file(md_file))
except Exception as exc:
logger.warning("Skipping relation file %s: %s", md_file.name, exc)
return relations