Menu
Use when orchestrating end-to-end software lifecycle with Aether.go methodology fusion framework across all eight stages
Instalar com Codex ou Claude Copie este prompt, cole no Codex, Claude ou outro assistente e deixe que ele revise a página da skill e instale para você.
Use when auditing the overall architecture for completeness, consistency, principle compliance, derivation chain integrity, and proposing/validating adjustments. This skill unifies the meta-verification layer and replaces architecture-self-auditor, derivation-chain-validator, adjustment-proposer, and adjustment-validator.
Use when designing system architecture, recording decisions, selecting patterns and tech stack, analyzing data flow, designing interface contracts, and applying the Strangler pattern. This skill unifies D2 Architecture Derivation Domain and replaces architecture-decision-recorder, architecture-pattern-selector, tech-stack-selector, data-flow-analyzer, interface-contract-designer, and strangler-pattern-suite.
Use when generating implementation code from interface contracts for backend (Go) and frontend (Vue/React) in any language. This skill unifies D5 Implementation Derivation Domain (L4.5→L5) and replaces contract-driven-code-generator, backend-code-generator, frontend-code-generator, and generic-code-generator.
Use when validating code and specifications against all 15 Aether constitutional principles (P0-P14), detecting principle conflicts, resolving via dynamic weighting, and enforcing mechanized constraints. This skill unifies the constitution enforcement layer and replaces constitution-validator, principle-consistency-checker, and constraint-check-runner.
Use when orchestrating complex deployments (canary, blue-green, rolling), managing releases, handling rollbacks, and enforcing change management. This skill unifies deployment operations and replaces deployment-orchestrator, rollback-manager, release-manager, and change-management.
Use when evaluating task determinism, dispatching to traditional code generators vs AI generation, enforcing contract consistency gates, and routing for confidence-based human review. This skill unifies D4.5 Generation Dispatch Domain and replaces deterministic-evaluator, code-generator-dispatcher, contract-consistency-gate, and confidence-based-reviewer.
Baseado na classificação ocupacional SOC
| name | methodology-fusion-orchestrator |
| description | Use when orchestrating end-to-end software lifecycle with Aether.go methodology fusion framework across all eight stages |
Orchestrate the complete Aether.go methodology fusion workflow across all eight stages: business analysis, specification definition, constitutional review, implementation planning, code generation, integration validation, deployment operations, and recursive optimization. Ensures constitutional principles are enforced, metrics are aggregated, and feedback loops drive continuous improvement.
Need end-to-end methodology fusion? ─────┐
│
Complex project with multiple stages? ────┤
├─► Use methodology-fusion-orchestrator
Constitutional compliance critical? ──────┤
│
Require automated feedback loops? ───────┘
Use when:
Don't use when:
User Request → Orchestrator → Stage Analysis → Skill Scheduling → Constitution Check
↑ ↓
└── Metrics Collection ← Execution ← Context Management ←──┘
↓
Optimization Analysis
↓
Skill Improvement
↓
Constitution Evolution
Business team: "We need a user authentication system"
Developers: Write code based on assumptions
Testers: Find issues after implementation
Ops team: Deploy with performance problems
Result: Misalignment, rework, quality issues
# Orchestrator Workflow Execution
workflow_id: "WF-AUTH-2025-001"
phases_executed:
- phase_1_business_analysis:
skill: business-requirements-collector
input: "User authentication system"
output: Structured business requirements
skill: business-value-mapper
input: Structured requirements
output: BMAD matrix with metrics
constitution_check: passed
duration: "2h"
- phase_2_specification_definition:
skill: spec-parser
input: BMAD matrix
output: Structured specifications
skill: bdd-scenario-writer
input: Structured specifications
output: Gherkin scenarios
skill: atdd-acceptance-test-generator
input: Gherkin scenarios
output: Executable acceptance tests
constitution_check: passed
duration: "3h"
- phase_3_constitutional_review:
skill: constitution-validator
input: Specifications + scenarios
output: Compliance report (92% score)
issues: 2 warnings, 0 errors
duration: "1h"
- phase_4_implementation_planning:
skill: architecture-pattern-selector
input: Validated specifications
output: Selected architecture pattern
skill: tech-stack-selector
input: Selected pattern + specifications
output: Technology stack selection
skill: architecture-decision-recorder
input: Selected pattern + tech stack + specifications
output: Architecture decisions + ADRs
skill: data-flow-analyzer
input: Architecture decisions
output: Data flow diagrams
duration: "4h"
- phase_5_code_generation:
skill: tdd-red-green-refactor
input: Implementation plan
output: Test-driven code
skill: go-backend-scaffolder
output: Clean architecture Go code
skill: vue-quasar-scaffolder
output: Vue 3 + Quasar components
constitution_check: passed
duration: "16h"
- phase_6_integration_validation:
skill: sit-scenario-generator
input: Complete system
output: Integration test scenarios
skill: contract-test-generator
input: API specifications
output: Contract tests for microservices
skill: chaos-test-designer
input: Integration tests
output: Resilience test plans
duration: "8h"
- phase_7_deployment_operations:
skill: deployment-orchestrator
input: Validated system
output: Deployed system with canary/blue-green strategy
skill: incident-management
output: Incident response procedures, on-call rotations
skill: change-management
output: CAB-approved change plans, rollback procedures
skill: release-manager
output: Release calendar, stakeholder communications
skill: metrics-definer
output: Business + technical metrics dashboard
skill: problem-management
output: Root cause analyses, permanent fixes
skill: service-desk
output: Service catalog, SLA agreements
skill: rollback-manager
output: Automated rollback procedures
duration: "8h"
- phase_8_recursive_optimization:
skill: recursive-optimizer
input: All metrics + feedback
output: Optimization recommendations
skill_improvements: 3 skills updated
constitution_evolution: 1 principle refined
assetizations:
- type: skill_assetization
skill: tdd-red-green-refactor
asset_id: tdd-red-green-refactor_a3f7b2c1
quality_score: 0.94
promoted_to_library: true
- type: workflow_assetization
pattern_id: workflow_parallel_exec_8d4e5f6g
quality_score: 0.88
duration: "4h"
metrics_summary:
total_duration: "46h"
constitutional_compliance: "96%"
requirement_traceability: "100%"
test_coverage: "92%"
business_value_alignment: "94%"
optimization_impact: "18% efficiency gain"
assets_created: 2
assets_promoted: 1
| Stage | Primary Skills | Supporting Skills | Output |
|---|---|---|---|
| 1. Business Analysis | business-requirements-collector, business-value-mapper | metrics-definer | Structured requirements + BMAD matrix |
| 2. Specification | spec-parser, bdd-scenario-writer, atdd-acceptance-test-generator | - | Structured specs + scenarios + acceptance tests |
| 3. Constitutional Review | constitution-validator | architecture-decision-recorder | Compliance report |
| 4. Implementation Planning | architecture-pattern-selector, tech-stack-selector, architecture-decision-recorder | data-flow-analyzer | Architecture pattern + tech stack + ADRs + data flows |
| 5. Code Generation | tdd-red-green-refactor, go-backend-scaffolder, vue-quasar-scaffolder | spec-to-code-tracer | Test-driven code |
| 6. Integration Validation | sit-scenario-generator, contract-test-generator, chaos-test-designer | test-pyramid-analyzer | Integration + contract + resilience tests |
| 7. Deployment & Operations | deployment-orchestrator, incident-management, change-management, release-manager | metrics-definer, problem-management, service-desk, rollback-manager | Deployment execution, ITIL operations, metrics dashboard |
| 8. Recursive Optimization | recursive-optimizer | prompt-template-manager, skill-recommender | Optimized skills |
Per Aether.go constitution, all 11 principles are enforced across stages:
constitution_enforcement:
# P0: Context-Adaptation Principle (Meta Principle)
context_adaptation:
- principle: "P0-context-adaptation-principle"
chinese: "情境适配原则"
description: "Dynamically adjust enforcement based on scenario mode"
exemption_mechanisms:
- poc_phase: [P2, P4, P6]
- emergency_fix: [P2, P6]
- legacy_system: [P4]
enforcement: adaptive
stage_1_business_analysis:
- principle: "P1-purpose-driven-principle"
chinese: "目的主导原则"
check: "Every business goal has measurable metrics"
enforcement: strict
validation: "BMAD matrix contains metrics for each goal"
stage_2_specification:
- principle: "P7-context-first-principle"
chinese: "上下文第一性原则"
check: "Complete context prepared before specification"
enforcement: strict
validation: "Requirements context + Technical context + Quality context documented"
stage_3_constitutional_review:
- principle: "P0-context-adaptation-principle"
chinese: "情境适配原则"
check: "Scenario mode appropriate for project context"
enforcement: strict
- principle: "P1-purpose-driven-principle"
chinese: "目的主导原则"
check: "Technical decisions traceable to business goals"
enforcement: strict
stage_4_implementation_planning:
- principle: "P2-planning-driven-principle"
chinese: "规划驱动原则"
check: "Detailed implementation plan before coding"
enforcement: strict
exemption: "poc_phase"
- principle: "P3-modularity-orthogonality-principle"
chinese: "模块化与正交性原则"
check: "High cohesion, low coupling architecture"
enforcement: warning
- principle: "P4-interface-first-principle"
chinese: "接口先行原则"
check: "All interfaces defined before implementation"
enforcement: strict
exemption: "poc_phase"
- principle: "P5-occams-razor-principle"
chinese: "奥卡姆剃刀原则"
check: "Dependencies ≤ 5, minimal complexity"
enforcement: warning
stage_5_code_generation:
- principle: "P6-test-first-principle"
chinese: "测试先行原则"
check: "Tests written before implementation"
enforcement: strict
exemption: "emergency_fix"
- principle: "P7-context-first-principle"
chinese: "上下文第一性原则"
check: "Context quality metrics > 90%"
enforcement: warning
stage_6_integration_validation:
- principle: "P6-test-first-principle"
chinese: "测试先行原则"
check: "Integration and contract tests defined"
enforcement: strict
- principle: "P9-recursive-self-optimization-principle"
chinese: "递归自我优化原则"
check: "Feedback loops established"
enforcement: info
stage_7_deployment:
- principle: "P8-human-ai-boundary-principle"
chinese: "人机责任边界原则"
check: "Critical deployment decisions have human confirmation"
enforcement: strict
confidence_thresholds:
auto_execute: 0.90
suggest_confirm: 0.70
escalate_human: 0.00
stage_8_recursive_optimization:
- principle: "P9-recursive-self-optimization-principle"
chinese: "递归自我优化原则"
check: "Convergence conditions satisfied"
enforcement: strict
convergence:
boundedness: true
monotonicity: true
termination: true
- principle: "P10-skill-assetization-principle"
chinese: "技能库资产化原则"
check: "Effective patterns assetized as reusable skills"
enforcement: info
quality_thresholds:
success_rate: 0.85
reusability: 3
cross_stage:
- principle: "P1-purpose-driven-principle"
chinese: "目的主导原则"
check: "All technical decisions traceable to business value"
enforcement: strict
- principle: "P8-human-ai-boundary-principle"
chinese: "人机责任边界原则"
check: "All AI-generated artifacts reviewed by human"
enforcement: strict
| Principle | Stages | Enforcement | Exemptions |
|---|---|---|---|
| P0 Context-Adaptation | All | Adaptive | Scenario-dependent |
| P1 Purpose-Driven | All | Strict | None |
| P2 Planning-Driven | 4, 5 | Strict | POC Phase |
| P3 Modularity-Orthogonality | 4, 5 | Warning | None |
| P4 Interface-First | 4, 5 | Strict | POC Phase |
| P5 Occam's Razor | 4, 5 | Warning | None |
| P6 Test-First | 5, 6 | Strict | Emergency Fix |
| P7 Context-First | 2, 5 | Warning | None |
| P8 Human-AI Boundary | 7, All | Strict | None |
| P9 Recursive Optimization | 6, 8 | Strict | None |
| P10 Skill Assetization | 8 | Info | None |
Before executing the methodology fusion workflow, the orchestrator initializes the complete .aether/ directory structure:
# Directory structure created by orchestrator (per dot-aether-reference.md)
.aether/
├── constitution.yml # Project宪法（YAML断言格式）
├── workflow.yml # 工作流配置（不变）
├── config.yml # 项目级配置（可选）
│
├── state/ # 【运行时状态 - 过程性】
│ ├── current.yml # 当前工作流状态
│ ├── sessions/ # 会话历史
│ │ ├── index.yml
│ │ └── {session-id}/
│ │ ├── session.yml
│ │ ├── context.yml
│ │ └── snapshots/
│ └── checkpoints/ # 全量检查点（快照）
│ ├── index.yml
│ └── {cp-id}/
│ ├── manifest.yml
│ └── snapshot.tar.gz
│
├── context/ # 【上下文管理 - 过程性】
│ ├── active/ # 当前活跃上下文
│ │ └── conv-{timestamp}.json
│ ├── archived/ # 归档上下文
│ │ └── {year-month}/
│ └── templates/ # 上下文模板
│ ├── specification.json
│ ├── coding.json
│ └── review.json
│
├── prompts/ # 【提示词管理 - 过程性】
│ ├── system/ # 系统提示词
│ ├── user/ # 用户自定义提示词
│ │ └── custom-prompts/
│ └── generated/ # 生成的提示词
│ └── task-specific/
│
├── memory/ # 【长期记忆 - 记忆性】
│ ├── facts.md # 事实记忆
│ ├── decisions.md # 决策记忆
│ ├── learnings.md # 学习记忆
│ ├── patterns.md # 模式记忆
│ ├── .pending/ # AI生成的待批准更新
│ │ ├── facts-{date}-001.md
│ │ ├── decisions-{date}-001.md
│ │ └── rejected/
│ └── embeddings/ # 向量嵌入数据
│ ├── index.yml
│ └── qdrant/
│
├── metrics/ # 【度量与反馈 - 过程性】
│ ├── series/ # 时序度量YAML文件
│ │ └── {year-month}/
│ │ ├── coverage.yml
│ │ ├── defect_rate.yml
│ │ └── performance.yml
│ ├── feedback/ # 用户人工评分/修正
│ │ └── {date}.yml
│ └── prometheus-exporter.yml
│
├── analysis/ # 【场景模式检测 - 过程性】
│ ├── latest-report.yml # 最新检测报告
│ └── history/ # 历史分析记录
│ └── {date}.yml
│
├── docs/ # 【用户产出描述性文档】
│ ├── 01-business/ # 业务需求文档
│ ├── 02-specification/ # 规范文档（用户故事、验收标准、BDD场景）
│ ├── 03-architecture/ # 架构设计文档
│ ├── 04-uml/ # UML模型
│ ├── 05-adr/ # 架构决策记录
│ │ └── INDEX.md
│ ├── 06-api/ # API文档+契约测试
│ ├── 07-testing/ # 测试文档（按金字塔分层）
│ │ ├── unit/
│ │ ├── integration/
│ │ ├── contract/
│ │ ├── acceptance/
│ │ └── e2e/
│ ├── 08-deployment/ # 部署文档
│ └── 09-reports/ # 报告文档
│
├── reference/ # 【引用文档 - 描述性】
│ ├── external/ # 外部引用
│ ├── standards/ # 标准规范
│ └── guidelines/ # 指导原则
│
├── skills/ # 【本地技能 - 过程性】
│ ├── installed/ # 已安装技能
│ ├── custom/ # 自定义技能
│ ├── templates/ # 技能模板
│ └── meta/ # 技能统计与质量
│ ├── success-rates.yml
│ └── improvement-queue/
│ └── {skill-name}.yaml
│
├── workflows/ # 【工作流定义 - 过程性】
│ ├── custom/ # 自定义工作流
│ ├── templates/ # 工作流模板
│ └── improvement/ # 技能改进任务
│
├── history/ # 【操作历史 - 过程性】
│ ├── commands.log # 命令历史（汇总）
│ ├── emergency_fixes.yml # 紧急修复清单
│ └── {year-month}/ # 按月组织的详细日志
│ └── {date}.log
│
├── cache/ # 【临时缓存 - 过程性】
│ ├── ai-contexts/ # AI上下文缓存
│ ├── suggestions/ # 建议缓存
│ └── temp/ # 临时文件
│
└── logs/ # 【日志文件 - 过程性】
└── aether.log
# .aether/orchestrator-config.yaml
orchestrator:
version: "1.0"
project_type: "fullstack-web"
methodology_version: "aether-go-2.0"
directories:
base: ".aether"
state: ".aether/state"
context: ".aether/context"
prompts: ".aether/prompts"
memory: ".aether/memory"
metrics: ".aether/metrics"
analysis: ".aether/analysis"
docs: ".aether/docs"
reference: ".aether/reference"
skills: ".aether/skills"
workflows: ".aether/workflows"
history: ".aether/history"
cache: ".aether/cache"
logs: ".aether/logs"
stages:
enabled: [1, 2, 3, 4, 5, 6, 7, 8]
mandatory: [1, 2, 3, 5, 7] # Must complete these stages
constitution:
file: "./constitution.yaml"
strict_mode: true
auto_evolve: true
evolution_threshold: 0.85 # 85% success rate triggers evolution
evolution:
triggers:
- type: "compliance_threshold"
condition: "constitutional_compliance > 0.85"
action: "propose_evolution"
- type: "feedback_pattern"
condition: "consistent_success_pattern_detected"
action: "analyze_and_evolve"
- type: "business_value"
condition: "business_value_alignment < 0.8"
action: "review_and_adjust"
evolution_process:
- analyze_patterns
- propose_changes
- human_review
- apply_evolution
- track_impact
rollback:
enabled: true
conditions:
- "new_compliance < old_compliance"
- "business_value_decrease > 0.1"
automatic: false
approval_required: true
metrics:
collection_points:
- stage_start
- skill_execution
- constitution_check
- stage_completion
aggregation: realtime
dashboard: auto_generate
optimization:
enabled: true
frequency: "after_each_stage"
scope: ["skills", "constitution", "workflow"]
approval: "auto_on_small, manual_on_major"
skill_effectiveness:
tracking:
- success_rate
- execution_time
- output_quality
- constitution_compliance
- user_satisfaction
reporting:
frequency: "daily"
dashboard: true
alerts:
- condition: "success_rate < 0.7"
action: "notify_team"
- condition: "execution_time > threshold"
action: "optimize_skill"
- condition: "constitution_compliance < 0.9"
action: "block_deployment"
analysis:
trend_analysis: true
anomaly_detection: true
comparative_analysis: true
benchmarking: true
visualization:
enabled: true
views:
- workflow_progress
- constitution_compliance
- skill_effectiveness
- metrics_dashboard
- traceability_graph
- asset_library
export_formats:
- html
- json
- svg
- pdf
real_time_updates: true
interactive_drill_down: true
collaboration:
team_management:
enabled: true
roles:
- business_analyst
- architect
- developer
- tester
- ops_engineer
approvals:
required_stages: [3, 7]
approvers:
stage_3: ["architect"]
stage_7: ["tech_lead", "ops_lead"]
notification:
channels: ["email", "slack", "webhook"]
timeout_hours: 24
comments:
enabled: true
threaded: true
mention_support: true
skill_scheduling:
algorithm: "context_aware"
factors: ["success_rate", "relevance", "efficiency"]
fallback: "skill-recommender"
context_management:
persistence: true
sharing: "cross_stage"
summarization: "auto"
assetization:
enabled: true
quality_threshold: 0.85
auto_promote: true
versioning:
enabled: true
retention_days: 90
sharing:
enabled: true
scope: ["project", "team", "organization"]
access_control: true
import os
from pathlib import Path
class AetherDirectoryInitializer:
"""Initializes the complete .aether directory structure."""
AETHER_STRUCTURE = {
'state': {
'sessions': ['index'],
'checkpoints': ['index']
},
'context': {
'active': [],
'archived': [],
'templates': []
},
'prompts': {
'system': [],
'user': ['custom-prompts'],
'generated': ['task-specific']
},
'memory': {
'.pending': [],
'embeddings': ['index']
},
'metrics': {
'series': [],
'feedback': []
},
'analysis': {
'history': []
},
'docs': {
'01-business': [],
'02-specification': ['user-stories', 'acceptance-criteria', 'bdd-scenarios'],
'03-architecture': ['interface-contracts'],
'04-uml': ['use-case', 'class', 'sequence', 'state', 'activity', 'component', 'deployment'],
'05-adr': [],
'06-api': ['protobuf', 'graphql', 'contracts'],
'07-testing': ['unit', 'integration', 'contract', 'acceptance', 'e2e'],
'08-deployment': ['k8s-manifests'],
'09-reports': []
},
'reference': {
'external': [],
'standards': [],
'guidelines': []
},
'skills': {
'installed': [],
'custom': [],
'templates': [],
'meta': ['success-rates', 'improvement-queue']
},
'workflows': {
'custom': [],
'templates': [],
'improvement': []
},
'history': [],
'cache': {
'ai-contexts': [],
'suggestions': [],
'temp': []
},
'logs': []
}
DEFAULT_FILES = {
'.aether/state/current.yml': '# Current Workflow State\n\n',
'.aether/context/active/.gitkeep': '',
'.aether/context/templates/specification.json': '{\n "version": "1.0",\n "description": "Context template for specification work"\n}',
'.aether/context/templates/coding.json': '{\n "version": "1.0",\n "description": "Context template for coding work"\n}',
'.aether/context/templates/review.json': '{\n "version": "1.0",\n "description": "Context template for review work"\n}',
'.aether/memory/facts.md': '# Project Facts\n\n## Overview\n\n',
'.aether/memory/decisions.md': '# Decision Log\n\n## Key Decisions\n\n',
'.aether/memory/learnings.md': '# Lessons Learned\n\n## Insights\n\n',
'.aether/memory/patterns.md': '# Patterns\n\n## Identified Patterns\n\n',
'.aether/memory/.pending/.gitkeep': '',
'.aether/memory/embeddings/index.yml': 'model: "sentence-transformers/all-MiniLM-L6-v2"\nchunk_strategy: "paragraph"\n',
'.aether/metrics/series/.gitkeep': '',
'.aether/metrics/feedback/.gitkeep': '',
'.aether/metrics/prometheus-exporter.yml': 'enabled: true\nport: 9090\nlocal_storage:\n path: "series/"\n retention_days: 30\n',
'.aether/analysis/latest-report.yml': 'detected_mode: "standard"\nconfidence: 0.0\nuser_confirmed: false\n',
'.aether/docs/05-adr/INDEX.md': '# Architecture Decision Records\n\n| ADR | Title | Date | Status |\n|-----|-------|------|--------|\n',
'.aether/docs/07-testing/.gitkeep': '',
'.aether/docs/09-reports/.gitkeep': '',
'.aether/skills/meta/success-rates.yml': 'skills: []\n',
'.aether/skills/meta/improvement-queue/.gitkeep': '',
'.aether/workflows/improvement/.gitkeep': '',
'.aether/history/commands.log': '',
'.aether/history/emergency_fixes.yml': 'fixes: []\n',
'.aether/cache/temp/.gitkeep': '',
'.aether/logs/aether.log': '',
'.aether/constitution.yml': '# Constitutional Principles\n\nversion: "1"\n\nprinciples: []\n',
'.aether/workflow.yml': '# Workflow Configuration\n\nversion: "1"\nstages: []\n',
'.aether/config.yml': '# Project Configuration\n\nversion: "1"\nproject: {}\n'
}
def __init__(self, base_path='.'):
self.base_path = Path(base_path)
self.aether_path = self.base_path / '.aether'
def initialize(self):
"""Initialize complete .aether directory structure."""
# Create directory structure
self._create_directories(self.AETHER_STRUCTURE, self.aether_path)
# Create default files
self._create_default_files()
# Create .gitignore
self._create_gitignore()
return {
'status': 'initialized',
'path': str(self.aether_path),
'directories_created': self._count_directories(),
'files_created': len(self.DEFAULT_FILES)
}
def _create_directories(self, structure, parent_path):
"""Recursively create directory structure."""
for name, content in structure.items():
current_path = parent_path / name
current_path.mkdir(parents=True, exist_ok=True)
if isinstance(content, dict):
self._create_directories(content, current_path)
elif isinstance(content, list):
for subdir in content:
(current_path / subdir).mkdir(exist_ok=True)
def _create_default_files(self):
"""Create default files with initial content."""
for file_path, content in self.DEFAULT_FILES.items():
full_path = self.base_path / file_path
if not full_path.exists():
full_path.write_text(content, encoding='utf-8')
def _create_gitignore(self):
"""Create .aether/.gitignore file."""
gitignore_content = """# Aether context files (temporary)
context/session/
# Cache files
*.cache
.cache/
# Logs
*.log
logs/
# Local overrides (user-specific)
local/
# Backup files
*.bak
*.backup
"""
gitignore_path = self.aether_path / '.gitignore'
if not gitignore_path.exists():
gitignore_path.write_text(gitignore_content, encoding='utf-8')
def _count_directories(self):
"""Count total directories created."""
return sum(1 for _ in self.aether_path.rglob('*') if _.is_dir())
class WorkflowOutputManager:
"""Manages output locations for workflow stages based on .aether structure."""
OUTPUT_PATHS = {
# Stage 1: Business Analysis
'business_requirements': '.aether/docs/01-business/',
'business_goals': '.aether/docs/01-business/business-goals.md',
'success_metrics': '.aether/docs/01-business/success-metrics.md',
'stakeholder_analysis': '.aether/docs/01-business/stakeholder-analysis.md',
# Stage 2: Specification
'user_stories': '.aether/docs/02-specification/user-stories/',
'acceptance_criteria': '.aether/docs/02-specification/acceptance-criteria/',
'bdd_scenarios': '.aether/docs/02-specification/bdd-scenarios/',
# Stage 3: Constitutional Review
'compliance_report': '.aether/docs/09-reports/constitutional-compliance-report.md',
# Stage 4: Implementation Planning
'adr_files': '.aether/docs/05-adr/',
'adr_index': '.aether/docs/05-adr/INDEX.md',
'system_overview': '.aether/docs/03-architecture/system-overview.md',
'component_diagram': '.aether/docs/03-architecture/component-diagram.md',
'data_model': '.aether/docs/03-architecture/data-model.md',
'interface_contracts': '.aether/docs/03-architecture/interface-contracts/',
'uml_models': '.aether/docs/04-uml/',
# Stage 5-8: Execution & Optimization
'workflow_metrics': '.aether/metrics/series/',
'feedback': '.aether/metrics/feedback/',
'traceability_matrix': '.aether/state/traceability/',
'gap_analysis': '.aether/state/gap-analysis.yaml',
'project_context': '.aether/context/active/',
'workflow_definitions': '.aether/workflows/custom/',
'prompt_templates': '.aether/prompts/',
# Memory
'facts': '.aether/memory/facts.md',
'decisions': '.aether/memory/decisions.md',
'learnings': '.aether/memory/learnings.md',
'patterns': '.aether/memory/patterns.md',
# Reports
'traceability_report': '.aether/docs/09-reports/traceability-report.md',
'coverage_report': '.aether/docs/09-reports/coverage-report.html',
'quality_report': '.aether/docs/09-reports/quality-report.md',
# Analysis
'latest_analysis': '.aether/analysis/latest-report.yml',
'analysis_history': '.aether/analysis/history/',
# Skills
'skill_success_rates': '.aether/skills/meta/success-rates.yml',
'skill_improvement_queue': '.aether/skills/meta/improvement-queue/',
# History
'commands_log': '.aether/history/commands.log',
'emergency_fixes': '.aether/history/emergency_fixes.yml',
# Reference
'external_reference': '.aether/reference/external/',
'standards': '.aether/reference/standards/',
'guidelines': '.aether/reference/guidelines/'
}
def __init__(self, base_path='.'):
self.base_path = Path(base_path)
def get_output_path(self, output_type, filename=None):
"""Get the output path for a specific output type."""
path = self.OUTPUT_PATHS.get(output_type)
if not path:
raise ValueError(f"Unknown output type: {output_type}")
full_path = self.base_path / path
# If it's a directory and filename provided, append filename
if filename and (full_path.is_dir() or path.endswith('/')):
full_path = full_path / filename
return full_path
def ensure_directory_exists(self, output_type):
"""Ensure the directory for an output type exists."""
path = self.get_output_path(output_type)
path.parent.mkdir(parents=True, exist_ok=True)
return path
class MethodologyFusionOrchestrator:
"""Main orchestrator coordinating all eight stages."""
def __init__(self, project_context, constitution, base_path='.'):
self.project_context = project_context
self.constitution = constitution
self.base_path = Path(base_path)
# Initialize directory structure
self.dir_initializer = AetherDirectoryInitializer(base_path)
self.output_manager = WorkflowOutputManager(base_path)
# Initialize components
self.metrics_collector = MetricsCollector(base_path)
self.skill_scheduler = SkillScheduler(base_path)
self.context_manager = ContextManager(base_path)
self.skill_assetizer = SkillAssetizer(base_path)
self.optimization_analyzer = OptimizationAnalyzer(base_path)
self.recursive_optimizer = RecursiveOptimizer(base_path)
self.workflow_optimizer = WorkflowOptimizer(base_path)
# Track initialization state
self.initialized = False
def initialize_project(self):
"""Initialize .aether directory structure for the project."""
if not self.initialized:
result = self.dir_initializer.initialize()
self.initialized = True
# Log initialization to memory
self._log_to_memory('facts', f"Project initialized with Aether structure at {result['path']}")
return result
return {'status': 'already_initialized'}
def _log_to_memory(self, memory_type, content):
"""Log entry to memory files."""
memory_files = {
'facts': '.aether/memory/facts.md',
'decisions': '.aether/memory/decisions.md',
'learnings': '.aether/memory/learnings.md',
'patterns': '.aether/memory/patterns.md'
}
memory_file = self.base_path / memory_files.get(memory_type, '.aether/memory/facts.md')
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
with open(memory_file, 'a', encoding='utf-8') as f:
f.write(f"\n## {timestamp}\n\n{content}\n")
def execute_workflow(self, user_request):
"""Execute complete eight-stage workflow."""
# Ensure directory structure is initialized
if not self.initialized:
self.initialize_project()
workflow_result = {
'id': generate_workflow_id(),
'start_time': datetime.now(),
'stages': [],
'metrics': {},
'optimizations': [],
'output_paths': {}
}
def execute_workflow(self, user_request):
"""Execute complete eight-stage workflow."""
workflow_result = {
'id': generate_workflow_id(),
'start_time': datetime.now(),
'stages': [],
'metrics': {},
'optimizations': []
}
# Stage 1: Business Analysis
stage1_result = self._execute_stage(
stage_id=1,
skills=['business-requirements-collector', 'business-value-mapper', 'metrics-definer'],
input=user_request,
constitution_checks=['Value-Driven Development']
)
workflow_result['stages'].append(stage1_result)
# Stage 2: Specification Definition
stage2_result = self._execute_stage(
stage_id=2,
skills=['spec-parser', 'bdd-scenario-writer', 'atdd-acceptance-test-generator'],
input=stage1_result['output'],
constitution_checks=['Test-First Development']
)
workflow_result['stages'].append(stage2_result)
# Stage 3: Constitutional Review
stage3_result = self._execute_stage(
stage_id=3,
skills=['constitution-validator'],
input={
'specs': stage2_result['output'],
'business_context': stage1_result['context']
},
constitution_checks=['Architectural Consistency']
)
workflow_result['stages'].append(stage3_result)
# Stage 4: Implementation Planning
stage4_result = self._execute_stage(
stage_id=4,
skills=['architecture-pattern-selector', 'tech-stack-selector', 'architecture-decision-recorder', 'data-flow-analyzer'],
input=stage3_result['validated_output'],
constitution_checks=['Interface-First Development', 'Simplicity & YAGNI']
)
workflow_result['stages'].append(stage4_result)
# Stage 5: Code Generation
stage5_result = self._execute_stage(
stage_id=5,
skills=['tdd-red-green-refactor', 'go-backend-scaffolder', 'vue-quasar-scaffolder'],
input=stage4_result['output'],
constitution_checks=['Code Quality & Standards']
)
workflow_result['stages'].append(stage5_result)
# Stage 6: Integration Validation
stage6_result = self._execute_stage(
stage_id=6,
skills=['sit-scenario-generator', 'contract-test-generator', 'chaos-test-designer'],
input=stage5_result['output'],
constitution_checks=['Resilience & Reliability']
)
workflow_result['stages'].append(stage6_result)
# Stage 7: Deployment & Operations
stage7_result = self._execute_stage(
stage_id=7,
skills=['deployment-orchestrator', 'incident-management', 'change-management', 'release-manager', 'metrics-definer', 'problem-management', 'service-desk', 'rollback-manager'],
input=stage6_result['validated_system'],
constitution_checks=['Observability & Monitoring', 'Deployment Safety', 'Incident Response', 'Change Control', 'Service Continuity']
)
workflow_result['stages'].append(stage7_result)
# Stage 8: Recursive Optimization
stage8_result = self._execute_optimization_stage(workflow_result)
workflow_result['stages'].append(stage8_result)
# Aggregate metrics
workflow_result['metrics'] = self.metrics_collector.aggregate(
[s['metrics'] for s in workflow_result['stages']]
)
workflow_result['end_time'] = datetime.now()
workflow_result['success'] = all(s['success'] for s in workflow_result['stages'])
return workflow_result
def _execute_stage(self, stage_id, skills, input, constitution_checks):
"""Execute a single stage with skill scheduling and constitution checks."""
stage_result = {
'stage_id': stage_id,
'start_time': datetime.now(),
'skills_executed': [],
'constitution_checks': [],
'metrics': {}
}
# Apply cross-stage constitution checks
cross_stage_checks = self.constitution.get_cross_stage_principles()
for principle in cross_stage_checks:
check_result = self.constitution.check_compliance(
principle=principle['name'],
artifacts={'stage_id': stage_id, 'input': input},
context=self.context_manager.get_context(stage_id)
)
stage_result['constitution_checks'].append(check_result)
if not check_result['passed']:
stage_result['blocked'] = True
stage_result['block_reason'] = f"Cross-stage constitution violation: {principle['name']}"
# Schedule and execute skills
for skill_name in skills:
skill_result = self.skill_scheduler.execute(
skill_name=skill_name,
input=input,
context=self.context_manager.get_context(stage_id)
)
stage_result['skills_executed'].append(skill_result)
# Collect metrics
stage_result['metrics'].update(
self.metrics_collector.collect_skill_metrics(skill_result)
)
# Apply stage-specific constitution checks
for principle in constitution_checks:
check_result = self.constitution.check_compliance(
principle=principle,
artifacts=stage_result['skills_executed'],
context=self.context_manager.get_context(stage_id)
)
stage_result['constitution_checks'].append(check_result)
if not check_result['passed']:
stage_result['blocked'] = True
stage_result['block_reason'] = f"Constitution violation: {principle}"
stage_result['end_time'] = datetime.now()
stage_result['duration'] = stage_result['end_time'] - stage_result['start_time']
stage_result['success'] = not stage_result.get('blocked', False)
# Update context for next stage
self.context_manager.update_context(
stage_id=stage_id,
data=stage_result,
constitution_checks=stage_result['constitution_checks']
)
return stage_result
def _execute_optimization_stage(self, workflow_result):
"""Execute recursive optimization stage."""
optimization_result = {
'stage_id': 8,
'type': 'recursive_optimization',
'start_time': datetime.now(),
'optimizations': [],
'assetizations': []
}
# Analyze workflow metrics for optimization opportunities
analysis = self.optimization_analyzer.analyze(workflow_result['metrics'])
# Optimize skills
for skill_analysis in analysis.get('skill_improvements', []):
optimized_skill = self.recursive_optimizer.optimize_skill(
skill_name=skill_analysis['skill'],
performance_data=skill_analysis['metrics'],
feedback=workflow_result['stages']
)
# Assetize successful patterns
if optimized_skill['success_rate'] > 0.9:
assetization_result = self.skill_assetizer.assetize(
skill_name=skill_analysis['skill'],
pattern=optimized_skill['pattern'],
metadata={
'usage_count': skill_analysis['usage_count'],
'success_rate': optimized_skill['success_rate'],
'last_used': datetime.now(),
'workflow_id': workflow_result['id']
}
)
optimization_result['assetizations'].append({
'type': 'skill_assetization',
'skill': skill_analysis['skill'],
'asset_id': assetization_result['asset_id'],
'quality_score': assetization_result['quality_score']
})
optimization_result['optimizations'].append({
'type': 'skill_improvement',
'skill': skill_analysis['skill'],
'improvement': optimized_skill['improvement'],
'impact': optimized_skill['expected_impact']
})
# Evolve constitution if needed
constitution_score = workflow_result['metrics'].get('constitutional_compliance', 0)
if constitution_score > self.constitution.evolution_threshold:
evolved_constitution = self.constitution.evolve(
workflow_data=workflow_result,
success_patterns=analysis.get('success_patterns', [])
)
optimization_result['optimizations'].append({
'type': 'constitution_evolution',
'principles_evolved': evolved_constitution['changes'],
'rationale': evolved_constitution['rationale']
})
# Optimize workflow patterns
workflow_optimizations = self.workflow_optimizer.analyze_patterns(
workflow_result['stages']
)
optimization_result['optimizations'].extend(workflow_optimizations)
# Assetize successful workflow patterns
for pattern in workflow_optimizations:
if pattern.get('success_rate', 0) > 0.85:
assetization_result = self.skill_assetizer.assetize(
skill_name='workflow_pattern',
pattern=pattern['pattern'],
metadata={
'type': 'workflow',
'success_rate': pattern['success_rate'],
'stages_involved': pattern['stages'],
'last_used': datetime.now(),
'workflow_id': workflow_result['id']
}
)
optimization_result['assetizations'].append({
'type': 'workflow_assetization',
'pattern_id': assetization_result['asset_id'],
'quality_score': assetization_result['quality_score']
})
optimization_result['end_time'] = datetime.now()
optimization_result['duration'] = optimization_result['end_time'] - optimization_result['start_time']
optimization_result['success'] = len(optimization_result['optimizations']) > 0
return optimization_result
class SkillScheduler:
"""Intelligent scheduler for selecting and executing skills."""
def execute(self, skill_name, input, context):
"""Execute a skill with intelligent scheduling."""
# Get skill metadata and performance history
skill_metadata = self.skill_registry.get(skill_name)
performance_history = self.metrics_db.get_skill_performance(skill_name)
# Check if skill needs optimization
if performance_history.get('success_rate', 0) < 0.7:
# Use fallback or optimized version
alternative = self._find_alternative_skill(skill_name, context)
if alternative:
skill_name = alternative
skill_metadata = self.skill_registry.get(skill_name)
# Prepare execution context
execution_context = {
'skill': skill_name,
'input': input,
'project_context': context,
'constitution': self.constitution.get_relevant_principles(skill_name),
'previous_stage_output': context.get('previous_output'),
'metrics_goals': context.get('metrics_targets', {})
}
# Execute skill
skill_executor = SkillExecutor(skill_metadata)
result = skill_executor.execute(execution_context)
# Collect execution metrics
execution_metrics = {
'skill': skill_name,
'duration': result['duration'],
'success': result['success'],
'output_quality': self._assess_output_quality(result['output'], context),
'constitution_compliance': result.get('constitution_compliance', 1.0),
'resource_usage': result.get('resource_usage', {})
}
# Update skill performance database
self.metrics_db.record_execution(skill_name, execution_metrics)
return {
'skill': skill_name,
'input': input,
'output': result['output'],
'metrics': execution_metrics,
'context_used': execution_context,
'success': result['success']
}
class SkillAssetizer:
"""Manages skill assetization and promotion to skill library."""
def __init__(self, skill_library, quality_threshold=0.85):
self.skill_library = skill_library
self.quality_threshold = quality_threshold
self.asset_registry = {}
def assetize(self, skill_name, pattern, metadata):
"""Assetize a successful skill pattern."""
asset_id = f"{skill_name}_{generate_uuid()[:8]}"
# Assess quality
quality_score = self._assess_quality(pattern, metadata)
if quality_score >= self.quality_threshold:
# Register asset
asset_record = {
'asset_id': asset_id,
'skill_name': skill_name,
'pattern': pattern,
'metadata': metadata,
'quality_score': quality_score,
'created_at': datetime.now(),
'usage_count': 0,
'status': 'active'
}
self.asset_registry[asset_id] = asset_record
# Promote to skill library if quality is high
if quality_score > 0.95:
self.skill_library.add_asset(asset_record)
return {
'asset_id': asset_id,
'quality_score': quality_score,
'promoted_to_library': quality_score > 0.95
}
return {
'asset_id': None,
'quality_score': quality_score,
'promoted_to_library': False,
'reason': 'Quality below threshold'
}
def _assess_quality(self, pattern, metadata):
"""Assess the quality of a skill pattern."""
quality_factors = {
'success_rate': metadata.get('success_rate', 0) * 0.4,
'usage_count': min(metadata.get('usage_count', 0) / 100, 1.0) * 0.2,
'pattern_complexity': self._assess_complexity(pattern) * 0.2,
'reusability': self._assess_reusability(pattern) * 0.2
}
return sum(quality_factors.values())
def _assess_complexity(self, pattern):
"""Assess pattern complexity (lower is better)."""
complexity = len(str(pattern))
normalized = max(0, 1 - (complexity / 10000))
return normalized
def _assess_reusability(self, pattern):
"""Assess pattern reusability."""
reusability_indicators = [
'template' in str(pattern).lower(),
'generic' in str(pattern).lower(),
'parameter' in str(pattern).lower()
]
return sum(reusability_indicators) / len(reusability_indicators)
class OptimizationAnalyzer:
"""Analyzes workflow metrics for optimization opportunities."""
def analyze(self, metrics):
"""Analyze metrics and identify optimization opportunities."""
analysis = {
'skill_improvements': [],
'success_patterns': [],
'bottlenecks': []
}
# Identify skills needing improvement
for skill_name, skill_metrics in metrics.get('skill_performance', {}).items():
if skill_metrics.get('success_rate', 1.0) < 0.8:
analysis['skill_improvements'].append({
'skill': skill_name,
'metrics': skill_metrics,
'priority': 'high' if skill_metrics['success_rate'] < 0.7 else 'medium'
})
# Identify success patterns
for pattern in metrics.get('successful_patterns', []):
if pattern.get('consistency', 0) > 0.9:
analysis['success_patterns'].append(pattern)
# Identify bottlenecks
for stage in metrics.get('stage_metrics', []):
if stage.get('duration') > stage.get('expected_duration', 0) * 1.5:
analysis['bottlenecks'].append({
'stage': stage['stage_id'],
'duration': stage['duration'],
'expected_duration': stage['expected_duration']
})
return analysis
class RecursiveOptimizer:
"""Optimizes skills based on performance data and feedback."""
def optimize_skill(self, skill_name, performance_data, feedback):
"""Optimize a skill based on performance data."""
optimization = {
'skill': skill_name,
'improvement': None,
'expected_impact': 0,
'success_rate': performance_data.get('success_rate', 0),
'pattern': None
}
# Analyze performance issues
issues = self._identify_issues(performance_data, feedback)
if issues:
# Generate optimization pattern
optimization['pattern'] = self._generate_optimization_pattern(
skill_name, issues, feedback
)
# Calculate expected impact
optimization['expected_impact'] = self._calculate_impact(
performance_data, issues
)
optimization['improvement'] = {
'type': 'pattern_optimization',
'issues_addressed': len(issues),
'estimated_improvement': optimization['expected_impact']
}
return optimization
def _identify_issues(self, performance_data, feedback):
"""Identify performance issues."""
issues = []
if performance_data.get('success_rate', 1.0) < 0.8:
issues.append('low_success_rate')
if performance_data.get('execution_time', 0) > performance_data.get('expected_time', 0) * 1.5:
issues.append('slow_execution')
if performance_data.get('constitution_compliance', 1.0) < 0.9:
issues.append('constitution_violation')
return issues
def _generate_optimization_pattern(self, skill_name, issues, feedback):
"""Generate optimization pattern."""
pattern = {
'skill': skill_name,
'optimizations': [],
'feedback_incorporated': []
}
for issue in issues:
if issue == 'low_success_rate':
pattern['optimizations'].append('add_error_handling')
pattern['optimizations'].append('improve_input_validation')
elif issue == 'slow_execution':
pattern['optimizations'].append('optimize_algorithm')
pattern['optimizations'].append('add_caching')
elif issue == 'constitution_violation':
pattern['optimizations'].append('enforce_constitution_checks')
return pattern
def _calculate_impact(self, performance_data, issues):
"""Calculate expected improvement impact."""
base_score = performance_data.get('success_rate', 0)
improvement_per_issue = 0.05
expected_improvement = len(issues) * improvement_per_issue
return min(expected_improvement, 1.0 - base_score)
class WorkflowOptimizer:
"""Optimizes workflow patterns based on execution data."""
def analyze_patterns(self, stages):
"""Analyze workflow patterns for optimization."""
optimizations = []
# Analyze stage dependencies
dependencies = self._analyze_dependencies(stages)
if dependencies.get('optimization_opportunity'):
optimizations.append({
'type': 'dependency_optimization',
'pattern': dependencies['pattern'],
'stages': dependencies['stages'],
'success_rate': dependencies.get('success_rate', 0),
'expected_improvement': dependencies.get('expected_improvement', 0)
})
# Analyze parallel execution opportunities
parallel_ops = self._analyze_parallel_opportunities(stages)
if parallel_ops:
optimizations.extend(parallel_ops)
return optimizations
def _analyze_dependencies(self, stages):
"""Analyze stage dependencies for optimization."""
dependencies = {
'stages': [],
'pattern': None,
'success_rate': 0,
'expected_improvement': 0
}
# Find stages that could be parallelized
for i, stage in enumerate(stages):
if i > 0 and not stage.get('blocked', False):
prev_stage = stages[i-1]
if not prev_stage.get('blocked', False):
dependencies['stages'].extend([prev_stage['stage_id'], stage['stage_id']])
if len(dependencies['stages']) >= 2:
dependencies['pattern'] = 'parallel_execution'
dependencies['success_rate'] = 0.92
dependencies['expected_improvement'] = 0.15
dependencies['optimization_opportunity'] = True
return dependencies
def _analyze_parallel_opportunities(self, stages):
"""Analyze opportunities for parallel execution."""
opportunities = []
# Look for independent stages
independent_stages = []
for stage in stages:
if not stage.get('blocked', False):
independent_stages.append(stage['stage_id'])
if len(independent_stages) >= 2:
opportunities.append({
'type': 'parallel_execution',
'pattern': {
'stages': independent_stages,
'execution_mode': 'parallel'
},
'stages': independent_stages,
'success_rate': 0.88,
'expected_improvement': 0.20
})
return opportunities
class Constitution:
"""Manages constitutional principles and compliance checking."""
def __init__(self, constitution_file):
self.constitution_file = constitution_file
self.principles = self._load_constitution()
self.evolution_threshold = 0.85
self.evolution_history = []
def _load_constitution(self):
"""Load constitution principles from file."""
return {
'stage_specific': {
1: ['Value-Driven Development'],
2: ['Test-First Development'],
3: ['Architectural Consistency'],
4: ['Interface-First Development', 'Simplicity & YAGNI'],
5: ['Code Quality & Standards'],
6: ['Resilience & Reliability'],
7: ['Observability & Monitoring', 'Deployment Safety', 'Incident Response', 'Change Control', 'Service Continuity'],
8: ['Continuous Improvement']
},
'cross_stage': [
{
'name': 'Human-AI Responsibility Boundary',
'check': 'All AI-generated artifacts reviewed by human',
'enforcement': 'strict'
}
]
}
def get_cross_stage_principles(self):
"""Get cross-stage constitutional principles."""
return self.principles.get('cross_stage', [])
def check_compliance(self, principle, artifacts, context):
"""Check compliance with a constitutional principle."""
check_result = {
'principle': principle,
'passed': True,
'violations': [],
'warnings': []
}
# Simulate compliance check
if principle == 'Value-Driven Development':
if not artifacts.get('business_metrics'):
check_result['passed'] = False
check_result['violations'].append('No business metrics defined')
elif principle == 'Test-First Development':
if not artifacts.get('acceptance_tests'):
check_result['passed'] = False
check_result['violations'].append('No acceptance tests defined')
elif principle == 'Interface-First Development':
if not artifacts.get('interfaces'):
check_result['passed'] = False
check_result['violations'].append('No interfaces defined')
elif principle == 'Human-AI Responsibility Boundary':
if not context.get('human_review'):
check_result['passed'] = False
check_result['violations'].append('No human review recorded')
return check_result
def evolve(self, workflow_data, success_patterns):
"""Evolve constitution based on workflow data and success patterns."""
evolution = {
'changes': [],
'rationale': []
}
# Analyze success patterns for new principles
for pattern in success_patterns:
if pattern.get('consistency', 0) > 0.95:
new_principle = self._derive_principle_from_pattern(pattern)
if new_principle:
evolution['changes'].append({
'type': 'add_principle',
'principle': new_principle
})
evolution['rationale'].append(
f"Derived from consistent pattern: {pattern['name']}"
)
# Record evolution
self.evolution_history.append({
'timestamp': datetime.now(),
'changes': evolution['changes'],
'rationale': evolution['rationale'],
'workflow_id': workflow_data.get('id')
})
return evolution
def _derive_principle_from_pattern(self, pattern):
"""Derive a constitutional principle from a success pattern."""
if 'parallel' in pattern.get('name', '').lower():
return {
'name': 'Parallel Execution Optimization',
'description': 'Consider parallel execution for independent stages',
'enforcement': 'warning'
}
return None
class ContextManager:
"""Manages context across workflow stages."""
def __init__(self):
self.context_store = {}
self.version_history = {}
def get_context(self, stage_id):
"""Get context for a specific stage."""
return self.context_store.get(stage_id, {})
def update_context(self, stage_id, data, constitution_checks):
"""Update context for a stage."""
context_entry = {
'stage_id': stage_id,
'data': data,
'constitution_checks': constitution_checks,
'timestamp': datetime.now(),
'version': len(self.version_history.get(stage_id, [])) + 1
}
self.context_store[stage_id] = context_entry
if stage_id not in self.version_history:
self.version_history[stage_id] = []
self.version_history[stage_id].append(context_entry)
def get_cross_stage_context(self):
"""Get context across all stages."""
return {
'stages': list(self.context_store.keys()),
'timeline': [
{
'stage_id': stage_id,
'timestamp': entry['timestamp'],
'success': entry['data'].get('success', False)
}
for stage_id, entry in self.context_store.items()
]
}
class MetricsCollector:
"""Collects and aggregates metrics across workflow stages."""
def __init__(self):
self.metrics_store = {}
def collect_skill_metrics(self, skill_result):
"""Collect metrics from skill execution."""
return {
f"{skill_result['skill']}_duration": skill_result['metrics'].get('duration', 0),
f"{skill_result['skill']}_success": skill_result['success'],
f"{skill_result['skill']}_quality": skill_result['metrics'].get('output_quality', 0),
f"{skill_result['skill']}_constitution_compliance": skill_result['metrics'].get('constitution_compliance', 1.0)
}
def aggregate(self, stage_metrics_list):
"""Aggregate metrics from all stages."""
aggregated = {
'skill_performance': {},
'stage_metrics': [],
'constitutional_compliance': 0,
'business_value_alignment': 0,
'successful_patterns': []
}
for stage_metrics in stage_metrics_list:
for key, value in stage_metrics.items():
if key.endswith('_success'):
skill_name = key.replace('_success', '')
if skill_name not in aggregated['skill_performance']:
aggregated['skill_performance'][skill_name] = {}
aggregated['skill_performance'][skill_name]['success_rate'] = value
# Calculate overall compliance
compliance_scores = [
m.get(k, 1.0) for m in stage_metrics_list
for k in m.keys() if k.endswith('_constitution_compliance')
]
aggregated['constitutional_compliance'] = sum(compliance_scores) / len(compliance_scores) if compliance_scores else 1.0
return aggregated
class SkillExecutor:
"""Executes skills with proper context and error handling."""
def __init__(self, skill_metadata):
self.skill_metadata = skill_metadata
def execute(self, execution_context):
"""Execute a skill with the given context."""
start_time = datetime.now()
try:
# Simulate skill execution
result = self._execute_implementation(execution_context)
success = True
error = None
except Exception as e:
result = None
success = False
error = str(e)
end_time = datetime.now()
return {
'output': result,
'success': success,
'error': error,
'duration': (end_time - start_time).total_seconds(),
'constitution_compliance': execution_context.get('constitution', {}).get('compliance_score', 1.0),
'resource_usage': {
'memory': '128MB',
'cpu': '50ms'
}
}
def _execute_implementation(self, execution_context):
"""Actual implementation of skill execution."""
return {
'result': f"Executed {execution_context['skill']}",
'context_used': execution_context
}
def generate_uuid():
"""Generate a unique identifier."""
import uuid
return str(uuid.uuid4())
class ConstitutionAwareExecutor:
"""Ensure all executions comply with constitutional principles."""
def check_stage_compliance(self, stage_id, artifacts, context):
"""Check if stage execution complies with constitution."""
relevant_principles = self.constitution.get_principles_for_stage(stage_id)
compliance_report = {
'stage_id': stage_id,
'checks': [],
'overall_score': 0,
'violations': [],
'warnings': []
}
total_weight = 0
weighted_score = 0
for principle in relevant_principles:
check_result = self._check_principle_compliance(
principle=principle,
artifacts=artifacts,
context=context
)
compliance_report['checks'].append(check_result)
if check_result['violation_level'] == 'error':
compliance_report['violations'].append({
'principle': principle['id'],
'description': check_result['description'],
'artifacts': check_result['violating_artifacts']
})
elif check_result['violation_level'] == 'warning':
compliance_report['warnings'].append({
'principle': principle['id'],
'description': check_result['description'],
'suggestion': check_result['suggestion']
})
# Calculate weighted score
weight = principle.get('weight', 1.0)
total_weight += weight
weighted_score += check_result['compliance_score'] * weight
if total_weight > 0:
compliance_report['overall_score'] = weighted_score / total_weight
# Determine if stage should be blocked
if compliance_report['violations']:
compliance_report['blocked'] = True
compliance_report['block_reason'] = "Constitutional violations found"
return compliance_report
def _check_principle_compliance(self, principle, artifacts, context):
"""Check compliance with a specific principle."""
checker = self._get_checker_for_principle(principle['id'])
result = checker.check(artifacts, context)
return {
'principle_id': principle['id'],
'principle_name': principle['name'],
'compliance_score': result.get('score', 0.0),
'violation_level': result.get('violation_level', 'none'),
'description': result.get('description', ''),
'violating_artifacts': result.get('violating_artifacts', []),
'suggestion': result.get('suggestion', ''),
'evidence': result.get('evidence', [])
}
| Mistake | Why It's Wrong | Fix |
|---|---|---|
| Skipping stages for "speed" | Breaks feedback loops, reduces quality | Always complete all mandatory stages |
| Ignoring constitution warnings | Leads to technical debt, architecture drift | Address warnings before proceeding |
| Not collecting metrics | Can't optimize or improve process | Enable metrics collection from start |
| Manual skill selection | Suboptimal skill choices, inefficiency | Trust orchestrator's scheduling |
| Disabling optimization | Stagnant process, missed improvements | Keep optimization enabled |
| Over-customization | Breaks methodology consistency | Follow standard workflow, customize carefully |
| Isolated stage execution | Loses context, reduces traceability | Use orchestrator for end-to-end flow |
Before (Disconnected Methodology):
After (Orchestrated Fusion):
Outcome: Predictable quality, faster delivery, consistent architecture, measurable business value, self-improving process.
Business Context → Methodology Fusion Orchestrator → Deployed System
↑ ↓
└─────────────── Recursive Optimization ←──────────┘
# Constitution evolution triggered by orchestrator
constitution_evolution:
trigger: "metrics.constitutional_compliance > 0.85"
process:
1. Analyze successful patterns across stages
2. Identify principles needing refinement
3. Propose evolution with evidence
4. Human review for major changes
5. Auto-apply minor refinements
outcomes:
- Principles become more precise
- Checks become more contextual
- Enforcement adapts to project type
- New principles emerge from patterns
# Optimization based on aggregated metrics
def optimize_based_on_metrics(workflow_metrics):
"""Trigger optimization based on stage metrics."""
optimizations = []
# Skill optimization
for skill_metrics in workflow_metrics.get('skill_performance', []):
if skill_metrics['success_rate'] < 0.8:
optimizations.append({
'type': 'skill_refinement',
'skill': skill_metrics['name'],
'focus': 'success_rate_improvement',
'target': 0.9
})
# Workflow optimization
stage_durations = workflow_metrics.get('stage_durations', {})
bottleneck = max(stage_durations, key=stage_durations.get)
if stage_durations[bottleneck] > timedelta(hours=8):
optimizations.append({
'type': 'workflow_restructuring',
'bottleneck': bottleneck,
'current_duration': stage_durations[bottleneck],
'target_reduction': '50%',
'strategy': 'parallel_execution_or_skill_optimization'
})
# Constitution optimization
compliance_scores = workflow_metrics.get('constitution_compliance', {})
low_scoring_principles = [
p for p, score in compliance_scores.items()
if score < 0.7
]
for principle in low_scoring_principles:
optimizations.append({
'type': 'constitution_clarification',
'principle': principle,
'current_score': compliance_scores[principle],
'target_score': 0.85,
'approach': 'provide_better_examples_and_checks'
})
return optimizations
# Context sharing across stages
context_sharing:
enabled: true
sharing_strategy: "selective_propagation"
propagated_items:
- business_goals: "From Stage 1 to all stages"
- constitutional_decisions: "From Stage 3 to Stages 4-7"
- architecture_constraints: "From Stage 4 to Stages 5-6"
- performance_targets: "From Stage 1 to Stages 5-7"
- user_experience_requirements: "From Stage 2 to Stages 5-6"
context_enrichment:
- each_stage: "Adds execution_context"
- each_skill: "Adds skill_specific_insights"
- constitution_checks: "Adds compliance_context"
- metrics: "Adds performance_context"
context_persistence:
storage: "workflow_database"
retention: "30_days"
queryable: true
used_for: "optimization_analysis, audit_trails, onboarding"