[ISSUE-14] Add BuildGraph, BuildSummary, and optional BuildReport

src/buildcompiler/execution/executor.py

@@ -171,17 +171,42 @@ def execute(
             if (not unresolved and not any(pending[s] for s in pending))
             else (BuildStatus.PARTIAL_SUCCESS if products else BuildStatus.FAILED)
         )
-        return FullBuildResult(
+        from buildcompiler.reporting import build_graph, build_report, build_summary
+
+        preliminary_result = FullBuildResult(
+            status=status,
+            plan=plan,
+            build_document=self.context.build_document,
+            stage_results=stage_results,
+            graph=None,
+            final_products=products,
+            missing_inputs=unresolved,
+            required_approvals=list(approvals.values()),
+            warnings=warnings,
+            summary=None,
+            report=None,
+        )
+        graph = build_graph(preliminary_result)
+        report = (
+            build_report(preliminary_result, graph=graph)
+            if self.context.options.reporting.include_detailed_report
+            else None
+        )
+        final_result = FullBuildResult(
             status=status,
             plan=plan,
             build_document=self.context.build_document,
             stage_results=stage_results,
-            graph=self.context.graph,
+            graph=graph,
             final_products=products,
             missing_inputs=unresolved,
             required_approvals=list(approvals.values()),
             warnings=warnings,
+            summary=None,
+            report=report,
         )
+        final_result.summary = build_summary(final_result)
+        return final_result
 
     def _run_stage(self, stage: Any, request: BuildRequest) -> StageResult:
         source_document = (

src/buildcompiler/reporting/__init__.py

@@ -1 +1,27 @@
-"""Package scaffolding for clean architecture."""
+"""Reporting models and builders."""
+
+from .graph import BuildGraph, BuildGraphEdge, BuildGraphNode, build_graph
+from .report import (
+    BuildReport,
+    DependencyChainStep,
+    RecommendedAction,
+    RouteReport,
+    StageReportSection,
+    build_report,
+)
+from .summary import BuildSummary, build_summary
+
+__all__ = [
+    "BuildGraph",
+    "BuildGraphEdge",
+    "BuildGraphNode",
+    "BuildReport",
+    "BuildSummary",
+    "DependencyChainStep",
+    "RecommendedAction",
+    "RouteReport",
+    "StageReportSection",
+    "build_graph",
+    "build_report",
+    "build_summary",
+]

src/buildcompiler/reporting/graph.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from buildcompiler.domain import FullBuildResult
+
+
+@dataclass(frozen=True)
+class BuildGraphNode:
+    id: str
+    kind: str
+    label: str | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class BuildGraphEdge:
+    source: str
+    target: str
+    relationship: str
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class BuildGraph:
+    nodes: list[BuildGraphNode] = field(default_factory=list)
+    edges: list[BuildGraphEdge] = field(default_factory=list)
+
+    def add_node(self, node: BuildGraphNode) -> None:
+        if node.id not in {n.id for n in self.nodes}:
+            self.nodes.append(node)
+
+    def add_edge(self, edge: BuildGraphEdge) -> None:
+        key = (edge.source, edge.target, edge.relationship, tuple(sorted(edge.metadata.items())))
+        keys = {
+            (e.source, e.target, e.relationship, tuple(sorted(e.metadata.items())))
+            for e in self.edges
+        }
+        if key not in keys:
+            self.edges.append(edge)
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "nodes": [
+                {
+                    "id": n.id,
+                    "kind": n.kind,
+                    "label": n.label,
+                    "metadata": n.metadata,
+                }
+                for n in sorted(self.nodes, key=lambda x: x.id)
+            ],
+            "edges": [
+                {
+                    "source": e.source,
+                    "target": e.target,
+                    "relationship": e.relationship,
+                    "metadata": e.metadata,
+                }
+                for e in sorted(
+                    self.edges,
+                    key=lambda x: (x.source, x.target, x.relationship, sorted(x.metadata.items())),
+                )
+            ],
+        }
+
+    def summary(self) -> dict[str, object]:
+        relationship_counts: dict[str, int] = {}
+        for edge in self.edges:
+            relationship_counts[edge.relationship] = relationship_counts.get(edge.relationship, 0) + 1
+        return {
+            "node_count": len(self.nodes),
+            "edge_count": len(self.edges),
+            "relationship_counts": dict(sorted(relationship_counts.items())),
+        }
+
+
+def _kind_for_identity(identity: str) -> str:
+    value = identity.lower()
+    if "moduledefinition" in value or "/module/" in value:
+        return "abstract_design"
+    if "engineered" in value or "region" in value:
+        return "engineered_region"
+    if "plasmid" in value:
+        return "plasmid"
+    if "strain" in value:
+        return "strain"
+    if "plate" in value:
+        return "plate"
+    return "part"
+
+
+def build_graph(result: FullBuildResult) -> BuildGraph:
+    graph = BuildGraph()
+
+    for stage_result in result.stage_results:
+        stage_node_id = f"stage_result:{stage_result.id}"
+        graph.add_node(BuildGraphNode(id=stage_node_id, kind="stage_result", label=stage_result.stage.value))
+        for request_id in sorted(stage_result.request_ids):
+            graph.add_node(BuildGraphNode(id=f"request:{request_id}", kind="abstract_design", label=request_id))
+            graph.add_edge(BuildGraphEdge(source=f"request:{request_id}", target=stage_node_id, relationship="requires"))
+        for product in stage_result.products:
+            graph.add_node(BuildGraphNode(id=product.identity, kind=_kind_for_identity(product.identity), label=product.display_id))
+            graph.add_edge(BuildGraphEdge(source=stage_node_id, target=product.identity, relationship="produces"))
+        for missing in stage_result.missing_inputs:
+            node_id = f"missing:{missing.missing_identity}"
+            graph.add_node(BuildGraphNode(id=node_id, kind="missing_input", label=missing.missing_display_id, metadata={"kind": missing.missing_kind}))
+            graph.add_edge(BuildGraphEdge(source=stage_node_id, target=node_id, relationship="blocks", metadata={"required_stage": str(missing.required_stage)}))
+        for approval in stage_result.required_approvals:
+            approval_id = f"approval:{approval.process}"
+            graph.add_node(BuildGraphNode(id=approval_id, kind="approval", label=approval.process, metadata={"status": approval.status.value}))
+            graph.add_edge(BuildGraphEdge(source=stage_node_id, target=approval_id, relationship="requires"))
+
+    for product in result.final_products:
+        graph.add_node(BuildGraphNode(id=product.identity, kind=_kind_for_identity(product.identity), label=product.display_id))
+
+    for missing in result.missing_inputs:
+        graph.add_node(BuildGraphNode(id=f"missing:{missing.missing_identity}", kind="missing_input", label=missing.missing_display_id, metadata={"kind": missing.missing_kind}))
+
+    for approval in result.required_approvals:
+        graph.add_node(BuildGraphNode(id=f"approval:{approval.process}", kind="approval", label=approval.process, metadata={"status": approval.status.value}))
+
+    return graph

src/buildcompiler/reporting/report.py

@@ -0,0 +1,159 @@
+from __future__ import annotations
+
+import json
+from dataclasses import asdict, dataclass, field
+from typing import Any
+
+from buildcompiler.domain import BuildStatus, FullBuildResult
+from buildcompiler.reporting.graph import BuildGraph, build_graph
+
+
+@dataclass
+class StageReportSection:
+    stage: str
+    status: str
+    request_ids: list[str]
+    product_count: int
+    missing_input_count: int
+    approval_count: int
+    warning_count: int
+    logs: list[str] = field(default_factory=list)
+
+
+@dataclass
+class RouteReport:
+    source_stage_result_id: str
+    selected: bool
+    route: dict[str, Any]
+
+
+@dataclass
+class RecommendedAction:
+    code: str
+    message: str
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class DependencyChainStep:
+    source: str
+    relationship: str
+    target: str
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class BuildReport:
+    status: BuildStatus
+    executive_summary: str
+    stage_sections: list[StageReportSection]
+    selected_routes: list[RouteReport]
+    rejected_alternatives: list[RouteReport]
+    missing_inputs: list[dict[str, Any]]
+    required_approvals: list[dict[str, Any]]
+    warnings: list[dict[str, Any]]
+    next_actions: list[RecommendedAction]
+    dependency_chain: list[DependencyChainStep]
+    graph_summary: dict[str, Any]
+
+    def to_dict(self) -> dict[str, Any]:
+        data = asdict(self)
+        data["status"] = self.status.value
+        return data
+
+    def to_json(self) -> str:
+        return json.dumps(self.to_dict(), sort_keys=True)
+
+    def to_markdown(self) -> str:
+        return "\n".join([
+            "# Build Report",
+            f"- Status: `{self.status.value}`",
+            f"- Stage sections: `{len(self.stage_sections)}`",
+            f"- Selected routes: `{len(self.selected_routes)}`",
+            f"- Rejected alternatives: `{len(self.rejected_alternatives)}`",
+            f"- Missing inputs: `{len(self.missing_inputs)}`",
+            f"- Required approvals: `{len(self.required_approvals)}`",
+            f"- Warnings: `{len(self.warnings)}`",
+            "",
+            "## Executive Summary",
+            self.executive_summary,
+        ])
+
+
+def _recommended_actions(result: FullBuildResult) -> list[RecommendedAction]:
+    actions: list[RecommendedAction] = []
+    for missing in result.missing_inputs:
+        kind = missing.missing_kind
+        if kind == "engineered_region":
+            actions.append(RecommendedAction("build_lvl1_engineered_region", "Build the missing engineered region through assembly level 1.", {"missing_identity": missing.missing_identity}))
+        elif kind in {"promoter", "rbs", "cds", "terminator"}:
+            actions.append(RecommendedAction("run_domestication", "Run domestication for missing part inputs.", {"missing_kind": kind, "missing_identity": missing.missing_identity}))
+        elif kind in {"backbone", "restriction_enzyme", "ligase", "reagent"}:
+            actions.append(RecommendedAction("provide_inventory_or_purchase", "Add missing inventory material or enable explicit purchase support.", {"missing_kind": kind, "missing_identity": missing.missing_identity}))
+    for approval in result.required_approvals:
+        actions.append(RecommendedAction("grant_required_approval", f"Grant required approval for process '{approval.process}'.", {"process": approval.process}))
+    for warning in result.warnings:
+        actions.append(RecommendedAction("inspect_warning", f"Inspect warning {warning.code} for details.", {"code": warning.code}))
+    # deterministic de-dup
+    unique: dict[tuple[str, str, str], RecommendedAction] = {}
+    for action in actions:
+        meta = json.dumps(action.metadata, sort_keys=True)
+        unique[(action.code, action.message, meta)] = action
+    return [unique[k] for k in sorted(unique)]
+
+
+def build_report(result: FullBuildResult, graph: BuildGraph | None = None) -> BuildReport:
+    report_graph = graph or build_graph(result)
+    stage_sections = [
+        StageReportSection(
+            stage=sr.stage.value,
+            status=sr.status.value,
+            request_ids=sorted(sr.request_ids),
+            product_count=len(sr.products),
+            missing_input_count=len(sr.missing_inputs),
+            approval_count=len(sr.required_approvals),
+            warning_count=len(sr.warnings),
+            logs=list(sr.logs),
+        )
+        for sr in result.stage_results
+    ]
+    selected_routes: list[RouteReport] = []
+    rejected: list[RouteReport] = []
+    for sr in result.stage_results:
+        artifacts = sr.protocol_artifacts or {}
+        sel = artifacts.get("selected_route")
+        if isinstance(sel, dict):
+            selected_routes.append(RouteReport(sr.id, True, sel))
+        for route in artifacts.get("rejected_routes", []) or []:
+            if isinstance(route, dict):
+                rejected.append(RouteReport(sr.id, False, route))
+
+    blocker_summary = f"{len(result.missing_inputs)} missing inputs and {len(result.required_approvals)} required approvals"
+    if result.status == BuildStatus.FAILED:
+        executive_summary = (
+            f"Build failed with {blocker_summary}."
+            if result.missing_inputs or result.required_approvals
+            else "Build failed. Review stage logs and warnings for the root cause."
+        )
+    elif result.missing_inputs or result.required_approvals:
+        executive_summary = f"Build is blocked by {blocker_summary}."
+    else:
+        executive_summary = "Build completed without unresolved blockers."
+    dependency_chain = [
+        DependencyChainStep(e.source, e.relationship, e.target, dict(e.metadata))
+        for e in sorted(report_graph.edges, key=lambda x: (x.source, x.target, x.relationship))
+        if e.relationship in {"blocks", "requires", "produces", "satisfies", "transforms", "plates"}
+    ]
+    return BuildReport(
+        status=result.status,
+        executive_summary=executive_summary,
+        stage_sections=stage_sections,
+        selected_routes=selected_routes,
+        rejected_alternatives=rejected,
+        missing_inputs=[asdict(x) | {"source_stage": x.source_stage.value, "required_stage": str(x.required_stage)} for x in result.missing_inputs],
+        required_approvals=[asdict(x) | {"status": x.status.value} for x in result.required_approvals],
+        warnings=[asdict(x) | {"stage": x.stage.value if x.stage else None} for x in result.warnings],
+        next_actions=_recommended_actions(result),
+        dependency_chain=dependency_chain,
+        graph_summary=report_graph.summary(),
+    )