diff --git a/src/atocore/api/routes.py b/src/atocore/api/routes.py index b0f23d8..b3e2416 100644 --- a/src/atocore/api/routes.py +++ b/src/atocore/api/routes.py @@ -30,6 +30,7 @@ from atocore.interactions.service import ( list_interactions, record_interaction, ) +from atocore.engineering.mirror import generate_project_overview from atocore.engineering.service import ( ENTITY_TYPES, RELATIONSHIP_TYPES, @@ -1074,6 +1075,25 @@ def api_create_relationship(req: RelationshipCreateRequest) -> dict: } +@router.get("/projects/{project_name}/mirror") +def api_project_mirror(project_name: str) -> dict: + """Generate a human-readable project overview from structured data. + + Layer 3 of the AtoCore architecture. The mirror is DERIVED from + entities, project state, and memories — it is not canonical truth. + Returns markdown that can be rendered, saved to a file, or served + as a dashboard page. + """ + from atocore.projects.registry import resolve_project_name as _resolve + + canonical = _resolve(project_name) + try: + markdown = generate_project_overview(canonical) + except Exception as e: + raise HTTPException(status_code=500, detail=f"Mirror generation failed: {e}") + return {"project": canonical, "format": "markdown", "content": markdown} + + @router.get("/admin/backup/{stamp}/validate") def api_validate_backup(stamp: str) -> dict: """Validate that a previously created backup is structurally usable.""" diff --git a/src/atocore/engineering/mirror.py b/src/atocore/engineering/mirror.py new file mode 100644 index 0000000..55a59fc --- /dev/null +++ b/src/atocore/engineering/mirror.py @@ -0,0 +1,220 @@ +"""Human Mirror — derived readable project views from structured data. + +Layer 3 of the AtoCore architecture. Generates human-readable markdown +pages from the engineering entity graph, Trusted Project State, and +active memories. These pages are DERIVED — they are not canonical +machine truth. They are support surfaces for human inspection and +audit comfort. + +The mirror never invents content. Every line traces back to an entity, +a state entry, or a memory. If the structured data is wrong, the +mirror is wrong — fix the source, not the page. +""" + +from __future__ import annotations + +from atocore.context.project_state import get_state +from atocore.engineering.service import ( + get_entities, + get_relationships, +) +from atocore.memory.service import get_memories +from atocore.observability.logger import get_logger + +log = get_logger("mirror") + + +def generate_project_overview(project: str) -> str: + """Generate a full project overview page in markdown.""" + sections = [ + _header(project), + _state_section(project), + _system_architecture(project), + _decisions_section(project), + _requirements_section(project), + _materials_section(project), + _vendors_section(project), + _active_memories_section(project), + _footer(project), + ] + return "\n\n".join(s for s in sections if s) + + +def _header(project: str) -> str: + return ( + f"# {project} — Project Overview\n\n" + f"> This page is auto-generated from AtoCore structured data.\n" + f"> It is a **derived view**, not canonical truth. " + f"If something is wrong here, fix the source data." + ) + + +def _state_section(project: str) -> str: + entries = get_state(project) + if not entries: + return "" + + lines = ["## Trusted Project State"] + by_category: dict[str, list] = {} + for e in entries: + by_category.setdefault(e.category.upper(), []).append(e) + + for cat in ["DECISION", "REQUIREMENT", "STATUS", "FACT", "MILESTONE", "CONFIG", "CONTACT"]: + items = by_category.get(cat, []) + if not items: + continue + lines.append(f"\n### {cat.title()}") + for item in items: + value = item.value[:300] + lines.append(f"- **{item.key}**: {value}") + if item.source: + lines.append(f" *(source: {item.source})*") + + return "\n".join(lines) + + +def _system_architecture(project: str) -> str: + systems = get_entities(entity_type="system", project=project) + subsystems = get_entities(entity_type="subsystem", project=project) + components = get_entities(entity_type="component", project=project) + interfaces = get_entities(entity_type="interface", project=project) + + if not systems and not subsystems and not components: + return "" + + lines = ["## System Architecture"] + + for system in systems: + lines.append(f"\n### {system.name}") + if system.description: + lines.append(f"{system.description}") + + rels = get_relationships(system.id, direction="outgoing") + children = [] + for rel in rels: + if rel.relationship_type == "contains": + child = next( + (s for s in subsystems + components if s.id == rel.target_entity_id), + None, + ) + if child: + children.append(child) + + if children: + lines.append("\n**Contains:**") + for child in children: + desc = f" — {child.description}" if child.description else "" + lines.append(f"- [{child.entity_type}] **{child.name}**{desc}") + + child_rels = get_relationships(child.id, direction="both") + for cr in child_rels: + if cr.relationship_type in ("uses_material", "interfaces_with", "constrained_by"): + other_id = ( + cr.target_entity_id + if cr.source_entity_id == child.id + else cr.source_entity_id + ) + other = next( + (e for e in get_entities(project=project, limit=200) + if e.id == other_id), + None, + ) + if other: + lines.append( + f" - *{cr.relationship_type}* → " + f"[{other.entity_type}] {other.name}" + ) + + return "\n".join(lines) + + +def _decisions_section(project: str) -> str: + decisions = get_entities(entity_type="decision", project=project) + if not decisions: + return "" + + lines = ["## Decisions"] + for d in decisions: + lines.append(f"\n### {d.name}") + if d.description: + lines.append(d.description) + rels = get_relationships(d.id, direction="outgoing") + for rel in rels: + if rel.relationship_type == "affected_by_decision": + affected = next( + (e for e in get_entities(project=project, limit=200) + if e.id == rel.target_entity_id), + None, + ) + if affected: + lines.append( + f"- Affects: [{affected.entity_type}] {affected.name}" + ) + + return "\n".join(lines) + + +def _requirements_section(project: str) -> str: + reqs = get_entities(entity_type="requirement", project=project) + constraints = get_entities(entity_type="constraint", project=project) + if not reqs and not constraints: + return "" + + lines = ["## Requirements & Constraints"] + for r in reqs: + lines.append(f"- **{r.name}**: {r.description}" if r.description else f"- **{r.name}**") + for c in constraints: + lines.append(f"- [constraint] **{c.name}**: {c.description}" if c.description else f"- [constraint] **{c.name}**") + + return "\n".join(lines) + + +def _materials_section(project: str) -> str: + materials = get_entities(entity_type="material", project=project) + if not materials: + return "" + + lines = ["## Materials"] + for m in materials: + desc = f" — {m.description}" if m.description else "" + lines.append(f"- **{m.name}**{desc}") + + return "\n".join(lines) + + +def _vendors_section(project: str) -> str: + vendors = get_entities(entity_type="vendor", project=project) + if not vendors: + return "" + + lines = ["## Vendors"] + for v in vendors: + desc = f" — {v.description}" if v.description else "" + lines.append(f"- **{v.name}**{desc}") + + return "\n".join(lines) + + +def _active_memories_section(project: str) -> str: + memories = get_memories(project=project, active_only=True, limit=20) + if not memories: + return "" + + lines = ["## Active Memories"] + for m in memories: + conf = f" (conf: {m.confidence:.2f})" if m.confidence < 1.0 else "" + refs = f" | refs: {m.reference_count}" if m.reference_count > 0 else "" + lines.append(f"- [{m.memory_type}]{conf}{refs} {m.content[:200]}") + + return "\n".join(lines) + + +def _footer(project: str) -> str: + from datetime import datetime, timezone + + now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") + return ( + f"---\n\n" + f"*Generated by AtoCore Human Mirror at {now}. " + f"This is a derived view — not canonical truth.*" + )