diff --git a/active-runtime/app/architect_governance_agent.py b/active-runtime/app/architect_governance_agent.py new file mode 100644 index 0000000..394e7ed --- /dev/null +++ b/active-runtime/app/architect_governance_agent.py @@ -0,0 +1,120 @@ +"""Architect governance agent responsibilities. + +This agent records requirement changes, maintains versioned human-readable +requirement/architecture/deployment/test documents, reviews architecture changes, +validates deployment docs against actual configuration, and produces version bump +recommendations. +""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass, field +from datetime import datetime, timezone +from typing import Any + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +@dataclass(frozen=True) +class RequirementChange: + id: str + source: str + actor: str + requirement: str + created_at: str = field(default_factory=utc_now) + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +@dataclass(frozen=True) +class DocumentRefreshPlan: + requirement_doc: bool + architecture_doc: bool + deployment_doc: bool + test_case_doc: bool + reason: str + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +@dataclass(frozen=True) +class ArchitectureReviewDecision: + status: str + findings: list[str] + optimization_recommendations: list[str] + version_bump: str + docs_to_refresh: DocumentRefreshPlan + created_at: str = field(default_factory=utc_now) + + def to_dict(self) -> dict[str, Any]: + return { + "status": self.status, + "findings": self.findings, + "optimization_recommendations": self.optimization_recommendations, + "version_bump": self.version_bump, + "docs_to_refresh": self.docs_to_refresh.to_dict(), + "created_at": self.created_at, + } + + +class ArchitectGovernanceAgent: + """Independent architecture governance role.""" + + def record_requirement_change(self, source: str, actor: str, requirement: str) -> RequirementChange: + change_id = "REQ-" + datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S") + return RequirementChange(change_id, source, actor, requirement) + + def review_architecture_change(self, proposal: dict[str, Any]) -> ArchitectureReviewDecision: + impact = proposal.get("impact", "medium") + findings = [ + "requirement_traceability_checked", + "architecture_consistency_checked", + "deployment_doc_alignment_checked", + "test_case_doc_required", + "rollback_and_owner_gate_checked", + ] + recommendations = [ + "refresh user requirements document", + "refresh architecture design document", + "refresh engineering deployment document", + "generate acceptance test case document", + ] + version_bump = self.decide_version_bump(proposal) + return ArchitectureReviewDecision( + status="review_required" if impact in {"high", "architecture"} else "pass", + findings=findings, + optimization_recommendations=recommendations, + version_bump=version_bump, + docs_to_refresh=DocumentRefreshPlan(True, True, True, True, "architecture or requirement changed"), + ) + + def decide_version_bump(self, proposal: dict[str, Any]) -> str: + if proposal.get("breaks_architecture") or proposal.get("major_capability"): + return "V" + if proposal.get("new_release_capability") or proposal.get("architecture_change"): + return "R" + return "C" + + def validate_deployment_doc(self, deployment_doc: dict[str, Any], actual_config: dict[str, Any]) -> dict[str, Any]: + expected = set(deployment_doc.get("services", [])) + actual = set(actual_config.get("services", [])) + missing = sorted(expected - actual) + extra = sorted(actual - expected) + return { + "status": "pass" if not missing else "attention_required", + "missing_services": missing, + "extra_services": extra, + "checked_at": utc_now(), + } + + +__all__ = [ + "RequirementChange", + "DocumentRefreshPlan", + "ArchitectureReviewDecision", + "ArchitectGovernanceAgent", +]