Spaces:

GaryLeckey
/

aureon-os

Running

aureon-os / aureon /code_architect /architect.py

Deploy Aureon OS (deploy/selfhost stack: nginx edge + fusion API x2 + console + HNC organ + redis + postgres; paper mode) (part 3)

1ccf652 verified 1 day ago

Raw History Blame Contribute Delete

19.2 kB

"""
CodeArchitect — the top-level self-authoring skill orchestrator
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

Ties the full pipeline together:

  Observer  →  Writer  →  Validator  →  Library  →  Executor
      │           │          │            │            │
      └ swarm     └ in-house └ Queen      └ JSON      └ VM/local
        motion      AI         meaning     store       dispatch

Usage:
    from aureon.code_architect import get_code_architect

    arch = get_code_architect()
    arch.bootstrap_atomics()                        # seed L0 skills
    arch.teach_compound("click_and_type", ["vm_left_click", "vm_type_text"])
    arch.build_workflow("morning_routine", ["check_mail", "review_calendar"])
    arch.build_role("be_ceo", ["morning_routine", "strategic_planning"])

    arch.execute_skill("be_ceo")                    # runs the whole tree

High-level API:
  - bootstrap_atomics()       : seed all 20 VM primitives as L0 skills
  - observe_and_propose()     : run one observe→propose cycle on the observer
  - learn_from_pattern(p)     : turn an ObservedPattern into a stored Skill
  - teach_compound(n, deps)   : manually compose a compound skill
  - build_workflow(n, tasks)  : manually compose a workflow
  - build_role(n, workflows)  : manually compose a role
  - execute_skill(name)       : run a skill through the full tree
  - get_status()              : full pipeline health
"""

from __future__ import annotations

import logging
import threading
import time
from typing import Any, Dict, List, Optional

from aureon.code_architect.skill import (
    Skill,
    SkillProposal,
    SkillLevel,
    SkillStatus,
)
from aureon.code_architect.skill_library import SkillLibrary, get_skill_library
from aureon.code_architect.primitives import VM_ACTION_NAMES
from aureon.code_architect.observer import ObservationEngine, ObservedPattern
from aureon.code_architect.writer import SkillWriter
from aureon.code_architect.validator import SkillValidator
from aureon.code_architect.executor import SkillExecutor, SkillExecutionResult

logger = logging.getLogger("aureon.code_architect.architect")


class CodeArchitect:
    """The top-level orchestrator of the self-authoring skill pipeline."""

    def __init__(
        self,
        library: Optional[SkillLibrary] = None,
        adapter: Any = None,
        dispatcher: Any = None,
        queen_bridge: Any = None,
        pillar_alignment: Any = None,
        thought_bus: Any = None,
        auto_wire: bool = True,
    ):
        # Use `is not None` rather than `or` because an empty library is
        # falsy (SkillLibrary.__len__ == 0 initially) and would otherwise
        # silently fall through to the singleton.
        self.library = library if library is not None else get_skill_library()
        self.writer = SkillWriter(adapter=adapter)
        self.validator = SkillValidator(
            queen_bridge=queen_bridge,
            pillar_alignment=pillar_alignment,
        )
        self.observer = ObservationEngine(window_size=3, min_occurrences=2)
        self.executor = SkillExecutor(
            library=self.library,
            dispatcher=dispatcher,
            validator=self.validator,
            thought_bus=thought_bus,
        )

        self._lock = threading.RLock()
        self._created_at = time.time()

        if auto_wire:
            self._auto_wire()

    def _auto_wire(self) -> None:
        # VM dispatcher
        if self.executor.dispatcher is None:
            try:
                from aureon.autonomous.vm_control import get_vm_dispatcher
                self.executor.dispatcher = get_vm_dispatcher()
            except Exception:
                pass

        # ThoughtBus
        if self.executor.thought_bus is None:
            try:
                from aureon.core.aureon_thought_bus import ThoughtBus
                self.executor.thought_bus = ThoughtBus()
            except Exception:
                pass

        # Load adapter for the writer (if not already)
        if self.writer.adapter is None:
            try:
                from aureon.inhouse_ai.llm_adapter import AureonBrainAdapter
                self.writer.adapter = AureonBrainAdapter()
            except Exception:
                pass

        # S03: Eager-wire validator's Queen bridge + pillar alignment.
        # Without this, the validator only lazy-loads them on first
        # validate() call, which means an audit right after __init__
        # sees them as None.
        try:
            self.validator._load_subsystems()
        except Exception:
            pass

    # ─────────────────────────────────────────────────────────────────────
    # Bootstrapping — seed the L0 atomic skills
    # ─────────────────────────────────────────────────────────────────────

    def bootstrap_atomics(self, overwrite: bool = False) -> List[Skill]:
        """
        Seed the library with L0 atomic skills for every VM primitive.

        Returns the list of skills created/verified.
        """
        created: List[Skill] = []
        for action in VM_ACTION_NAMES:
            skill_name = action  # keep name matching the primitive (e.g. "left_click")
            if not overwrite and self.library.contains(skill_name):
                created.append(self.library.get(skill_name))
                continue

            proposal = self.writer.propose_atomic(
                action=action,
                name=skill_name,
                description=f"Atomic VM primitive: {action}",
                category="atomic",
                target="vm",
            )
            skill = self._store_if_valid(proposal)
            if skill:
                created.append(skill)

        self.library.save()
        logger.info("Bootstrapped %d atomic skills", len(created))
        return created

    def _store_if_valid(self, proposal: SkillProposal) -> Optional[Skill]:
        """Validate a proposal and store it in the library if it passes."""
        validation = self.validator.validate(proposal)
        skill = Skill.from_proposal(proposal)

        # Fold validation verdict into the skill
        skill.queen_verdict = validation.queen_verdict
        skill.queen_confidence = validation.queen_confidence
        skill.pillar_alignment_score = validation.pillar_alignment_score
        skill.pillar_lighthouse = validation.pillar_lighthouse
        skill.harmonic_signature = validation.harmonic_signature

        if not validation.validated:
            skill.status = SkillStatus.BLOCKED
            skill.last_error = validation.reasoning
            logger.warning("Skill %s blocked: %s", skill.name, validation.reasoning)
            # Still store blocked skills so the user can inspect them
            self.library.add(skill, persist=False)
            return None

        if validation.approved:
            skill.status = SkillStatus.APPROVED
        else:
            skill.status = SkillStatus.VALIDATED

        self.library.add(skill, persist=False)
        return skill

    # ─────────────────────────────────────────────────────────────────────
    # Observation → pattern → skill
    # ─────────────────────────────────────────────────────────────────────

    def observe_and_propose(self) -> List[Skill]:
        """
        Pull all pending patterns from the observer and turn each into a
        validated, stored skill. Returns the list of new skills.
        """
        patterns = self.observer.get_pending_patterns(clear=True)
        created: List[Skill] = []
        for pattern in patterns:
            skill = self.learn_from_pattern(pattern)
            if skill:
                created.append(skill)
        if created:
            self.library.save()
        return created

    def learn_from_pattern(self, pattern: ObservedPattern) -> Optional[Skill]:
        """Generate + validate + store a skill from an ObservedPattern."""
        proposal = self.writer.propose_from_pattern(pattern)
        if self.library.contains(proposal.name):
            # Append a suffix to avoid collisions
            base = proposal.name
            suffix = 1
            while self.library.contains(f"{base}_{suffix}"):
                suffix += 1
            proposal.name = f"{base}_{suffix}"
            proposal.entry_function = proposal.name
            # Regenerate code with the new name
            proposal.code = proposal.code.replace(f"def {base}(", f"def {proposal.name}(", 1)
        return self._store_if_valid(proposal)

    # ─────────────────────────────────────────────────────────────────────
    # Manual composition — teach compound/workflow/role skills
    # ─────────────────────────────────────────────────────────────────────

    def teach_compound(
        self,
        name: str,
        dependency_skill_names: List[str],
        description: str = "",
    ) -> Optional[Skill]:
        """Teach a compound (L1/L2) skill composed of existing skills."""
        proposal = self.writer.propose_compound(
            name=name,
            dependency_skill_names=dependency_skill_names,
            description=description,
            level=SkillLevel.COMPOUND,
            category="compound",
        )
        skill = self._store_if_valid(proposal)
        self.library.save()
        return skill

    def build_task(
        self,
        name: str,
        dependency_skill_names: List[str],
        description: str = "",
    ) -> Optional[Skill]:
        """Build an L2 task skill."""
        proposal = self.writer.propose_compound(
            name=name,
            dependency_skill_names=dependency_skill_names,
            description=description,
            level=SkillLevel.TASK,
            category="task",
        )
        skill = self._store_if_valid(proposal)
        self.library.save()
        return skill

    def build_workflow(
        self,
        name: str,
        task_skill_names: List[str],
        description: str = "",
    ) -> Optional[Skill]:
        """Build an L3 workflow."""
        proposal = self.writer.propose_workflow(
            name=name,
            task_skill_names=task_skill_names,
            description=description,
        )
        skill = self._store_if_valid(proposal)
        self.library.save()
        return skill

    def build_role(
        self,
        name: str,
        workflow_skill_names: List[str],
        description: str = "",
    ) -> Optional[Skill]:
        """
        Build an L4 role (persona) — e.g. be_ceo, be_engineer, be_scout.

        The role's generated code emits 'role.enter' / 'role.exit' events
        and calls each workflow in sequence via call_skill().
        """
        proposal = self.writer.propose_role(
            name=name,
            workflow_skill_names=workflow_skill_names,
            description=description,
        )
        skill = self._store_if_valid(proposal)
        self.library.save()
        return skill

    # ─────────────────────────────────────────────────────────────────────
    # Execution
    # ─────────────────────────────────────────────────────────────────────

    def execute_skill(
        self,
        skill_name: str,
        params: Optional[Dict[str, Any]] = None,
        resolve_deps: bool = True,
    ) -> SkillExecutionResult:
        """
        Execute a skill by name.

        If resolve_deps=True, pre-compiles the full dependency tree first.
        """
        if resolve_deps:
            return self.executor.execute_with_dependencies(skill_name, params=params)
        return self.executor.execute_by_name(skill_name, params=params)

    # ─────────────────────────────────────────────────────────────────────
    # High-level: build the "be a CEO" persona from scratch
    # ─────────────────────────────────────────────────────────────────────

    def demo_build_ceo_persona(self) -> Dict[str, Any]:
        """
        Full demo: build a complete skill tree from L0 atomics up to a L4
        "be_ceo" role, using only the VM primitives that already exist.

        This proves the full hierarchical composition pipeline works
        without any real LLM — purely through template-mode composition
        and validated execution.

        Returns a summary dict.
        """
        # 1. Seed L0 atomics
        atomics = self.bootstrap_atomics()

        # 2. Build a few L1 compound skills from existing primitives
        compound_specs = [
            ("click_somewhere", ["mouse_move", "left_click"]),
            ("type_a_sentence", ["type_text", "press_key"]),
            ("inspect_screen",  ["screenshot", "list_windows"]),
            ("run_command",     ["execute_shell"]),
            ("read_powershell", ["execute_powershell"]),
        ]
        compounds: List[Optional[Skill]] = []
        for name, deps in compound_specs:
            s = self.teach_compound(name=name, dependency_skill_names=deps,
                                    description=f"Compound: {name.replace('_', ' ')}")
            compounds.append(s)

        # 3. Build L2 tasks from compounds
        task_specs = [
            ("check_mail", ["inspect_screen", "click_somewhere", "type_a_sentence"]),
            ("review_calendar", ["inspect_screen", "click_somewhere"]),
            ("analyse_market", ["run_command", "inspect_screen"]),
            ("make_a_decision", ["read_powershell", "type_a_sentence"]),
            ("delegate_tasks", ["type_a_sentence", "click_somewhere"]),
            ("strategic_thinking", ["inspect_screen", "read_powershell", "type_a_sentence"]),
        ]
        tasks: List[Optional[Skill]] = []
        for name, deps in task_specs:
            t = self.build_task(name=name, dependency_skill_names=deps,
                                description=f"Task: {name.replace('_', ' ')}")
            tasks.append(t)

        # 4. Build L3 workflows from tasks
        workflow_specs = [
            ("morning_routine", ["check_mail", "review_calendar", "analyse_market"]),
            ("strategic_planning", ["analyse_market", "strategic_thinking", "make_a_decision"]),
            ("leadership_cycle", ["delegate_tasks", "strategic_thinking"]),
            ("end_of_day", ["review_calendar", "delegate_tasks"]),
        ]
        workflows: List[Optional[Skill]] = []
        for name, deps in workflow_specs:
            w = self.build_workflow(name=name, task_skill_names=deps,
                                    description=f"Workflow: {name.replace('_', ' ')}")
            workflows.append(w)

        # 5. Build the L4 role — be_ceo
        ceo = self.build_role(
            name="be_ceo",
            workflow_skill_names=[
                "morning_routine",
                "strategic_planning",
                "leadership_cycle",
                "end_of_day",
            ],
            description="Persona: act as a CEO — orchestrate the full day cycle",
        )

        # 6. Summary
        return {
            "atomics_count": len([s for s in atomics if s is not None]),
            "compound_count": len([s for s in compounds if s is not None]),
            "task_count": len([s for s in tasks if s is not None]),
            "workflow_count": len([s for s in workflows if s is not None]),
            "role_built": ceo is not None,
            "ceo_skill_id": ceo.skill_id if ceo else None,
            "ceo_status": ceo.status.value if ceo else None,
            "ceo_dependencies": ceo.dependencies if ceo else [],
            "library_size": len(self.library),
        }

    # ─────────────────────────────────────────────────────────────────────
    # Status
    # ─────────────────────────────────────────────────────────────────────

    def get_status(self) -> Dict[str, Any]:
        return {
            "uptime_s": round(time.time() - self._created_at, 2),
            "library": self.library.get_stats(),
            "writer": self.writer.get_stats(),
            "validator": self.validator.get_stats(),
            "executor": self.executor.get_status(),
            "observer": self.observer.get_status(),
        }


# ─────────────────────────────────────────────────────────────────────────────
# Singleton
# ─────────────────────────────────────────────────────────────────────────────


_architect_instance: Optional[CodeArchitect] = None
_architect_lock = threading.Lock()


def get_code_architect(**kwargs) -> CodeArchitect:
    """Get or create the singleton CodeArchitect."""
    global _architect_instance
    with _architect_lock:
        if _architect_instance is None:
            _architect_instance = CodeArchitect(**kwargs)
        return _architect_instance

	"""
	CodeArchitect — the top-level self-authoring skill orchestrator
	━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

	Ties the full pipeline together:

	Observer → Writer → Validator → Library → Executor
	│ │ │ │ │
	└ swarm └ in-house └ Queen └ JSON └ VM/local
	motion AI meaning store dispatch

	Usage:
	from aureon.code_architect import get_code_architect

	arch = get_code_architect()
	arch.bootstrap_atomics() # seed L0 skills
	arch.teach_compound("click_and_type", ["vm_left_click", "vm_type_text"])
	arch.build_workflow("morning_routine", ["check_mail", "review_calendar"])
	arch.build_role("be_ceo", ["morning_routine", "strategic_planning"])

	arch.execute_skill("be_ceo") # runs the whole tree

	High-level API:
	- bootstrap_atomics() : seed all 20 VM primitives as L0 skills
	- observe_and_propose() : run one observe→propose cycle on the observer
	- learn_from_pattern(p) : turn an ObservedPattern into a stored Skill
	- teach_compound(n, deps) : manually compose a compound skill
	- build_workflow(n, tasks) : manually compose a workflow
	- build_role(n, workflows) : manually compose a role
	- execute_skill(name) : run a skill through the full tree
	- get_status() : full pipeline health
	"""

	from __future__ import annotations

	import logging
	import threading
	import time
	from typing import Any, Dict, List, Optional

	from aureon.code_architect.skill import (
	Skill,
	SkillProposal,
	SkillLevel,
	SkillStatus,
	)
	from aureon.code_architect.skill_library import SkillLibrary, get_skill_library
	from aureon.code_architect.primitives import VM_ACTION_NAMES
	from aureon.code_architect.observer import ObservationEngine, ObservedPattern
	from aureon.code_architect.writer import SkillWriter
	from aureon.code_architect.validator import SkillValidator
	from aureon.code_architect.executor import SkillExecutor, SkillExecutionResult

	logger = logging.getLogger("aureon.code_architect.architect")


	class CodeArchitect:
	"""The top-level orchestrator of the self-authoring skill pipeline."""

	def __init__(
	self,
	library: Optional[SkillLibrary] = None,
	adapter: Any = None,
	dispatcher: Any = None,
	queen_bridge: Any = None,
	pillar_alignment: Any = None,
	thought_bus: Any = None,
	auto_wire: bool = True,
	):
	# Use `is not None` rather than `or` because an empty library is
	# falsy (SkillLibrary.__len__ == 0 initially) and would otherwise
	# silently fall through to the singleton.
	self.library = library if library is not None else get_skill_library()
	self.writer = SkillWriter(adapter=adapter)
	self.validator = SkillValidator(
	queen_bridge=queen_bridge,
	pillar_alignment=pillar_alignment,
	)
	self.observer = ObservationEngine(window_size=3, min_occurrences=2)
	self.executor = SkillExecutor(
	library=self.library,
	dispatcher=dispatcher,
	validator=self.validator,
	thought_bus=thought_bus,
	)

	self._lock = threading.RLock()
	self._created_at = time.time()

	if auto_wire:
	self._auto_wire()

	def _auto_wire(self) -> None:
	# VM dispatcher
	if self.executor.dispatcher is None:
	try:
	from aureon.autonomous.vm_control import get_vm_dispatcher
	self.executor.dispatcher = get_vm_dispatcher()
	except Exception:
	pass

	# ThoughtBus
	if self.executor.thought_bus is None:
	try:
	from aureon.core.aureon_thought_bus import ThoughtBus
	self.executor.thought_bus = ThoughtBus()
	except Exception:
	pass

	# Load adapter for the writer (if not already)
	if self.writer.adapter is None:
	try:
	from aureon.inhouse_ai.llm_adapter import AureonBrainAdapter
	self.writer.adapter = AureonBrainAdapter()
	except Exception:
	pass

	# S03: Eager-wire validator's Queen bridge + pillar alignment.
	# Without this, the validator only lazy-loads them on first
	# validate() call, which means an audit right after __init__
	# sees them as None.
	try:
	self.validator._load_subsystems()
	except Exception:
	pass

	# ─────────────────────────────────────────────────────────────────────
	# Bootstrapping — seed the L0 atomic skills
	# ─────────────────────────────────────────────────────────────────────

	def bootstrap_atomics(self, overwrite: bool = False) -> List[Skill]:
	"""
	Seed the library with L0 atomic skills for every VM primitive.

	Returns the list of skills created/verified.
	"""
	created: List[Skill] = []
	for action in VM_ACTION_NAMES:
	skill_name = action # keep name matching the primitive (e.g. "left_click")
	if not overwrite and self.library.contains(skill_name):
	created.append(self.library.get(skill_name))
	continue

	proposal = self.writer.propose_atomic(
	action=action,
	name=skill_name,
	description=f"Atomic VM primitive: {action}",
	category="atomic",
	target="vm",
	)
	skill = self._store_if_valid(proposal)
	if skill:
	created.append(skill)

	self.library.save()
	logger.info("Bootstrapped %d atomic skills", len(created))
	return created

	def _store_if_valid(self, proposal: SkillProposal) -> Optional[Skill]:
	"""Validate a proposal and store it in the library if it passes."""
	validation = self.validator.validate(proposal)
	skill = Skill.from_proposal(proposal)

	# Fold validation verdict into the skill
	skill.queen_verdict = validation.queen_verdict
	skill.queen_confidence = validation.queen_confidence
	skill.pillar_alignment_score = validation.pillar_alignment_score
	skill.pillar_lighthouse = validation.pillar_lighthouse
	skill.harmonic_signature = validation.harmonic_signature

	if not validation.validated:
	skill.status = SkillStatus.BLOCKED
	skill.last_error = validation.reasoning
	logger.warning("Skill %s blocked: %s", skill.name, validation.reasoning)
	# Still store blocked skills so the user can inspect them
	self.library.add(skill, persist=False)
	return None

	if validation.approved:
	skill.status = SkillStatus.APPROVED
	else:
	skill.status = SkillStatus.VALIDATED

	self.library.add(skill, persist=False)
	return skill

	# ─────────────────────────────────────────────────────────────────────
	# Observation → pattern → skill
	# ─────────────────────────────────────────────────────────────────────

	def observe_and_propose(self) -> List[Skill]:
	"""
	Pull all pending patterns from the observer and turn each into a
	validated, stored skill. Returns the list of new skills.
	"""
	patterns = self.observer.get_pending_patterns(clear=True)
	created: List[Skill] = []
	for pattern in patterns:
	skill = self.learn_from_pattern(pattern)
	if skill:
	created.append(skill)
	if created:
	self.library.save()
	return created

	def learn_from_pattern(self, pattern: ObservedPattern) -> Optional[Skill]:
	"""Generate + validate + store a skill from an ObservedPattern."""
	proposal = self.writer.propose_from_pattern(pattern)
	if self.library.contains(proposal.name):
	# Append a suffix to avoid collisions
	base = proposal.name
	suffix = 1
	while self.library.contains(f"{base}_{suffix}"):
	suffix += 1
	proposal.name = f"{base}_{suffix}"
	proposal.entry_function = proposal.name
	# Regenerate code with the new name
	proposal.code = proposal.code.replace(f"def {base}(", f"def {proposal.name}(", 1)
	return self._store_if_valid(proposal)

	# ─────────────────────────────────────────────────────────────────────
	# Manual composition — teach compound/workflow/role skills
	# ─────────────────────────────────────────────────────────────────────

	def teach_compound(
	self,
	name: str,
	dependency_skill_names: List[str],
	description: str = "",
	) -> Optional[Skill]:
	"""Teach a compound (L1/L2) skill composed of existing skills."""
	proposal = self.writer.propose_compound(
	name=name,
	dependency_skill_names=dependency_skill_names,
	description=description,
	level=SkillLevel.COMPOUND,
	category="compound",
	)
	skill = self._store_if_valid(proposal)
	self.library.save()
	return skill

	def build_task(
	self,
	name: str,
	dependency_skill_names: List[str],
	description: str = "",
	) -> Optional[Skill]:
	"""Build an L2 task skill."""
	proposal = self.writer.propose_compound(
	name=name,
	dependency_skill_names=dependency_skill_names,
	description=description,
	level=SkillLevel.TASK,
	category="task",
	)
	skill = self._store_if_valid(proposal)
	self.library.save()
	return skill

	def build_workflow(
	self,
	name: str,
	task_skill_names: List[str],
	description: str = "",
	) -> Optional[Skill]:
	"""Build an L3 workflow."""
	proposal = self.writer.propose_workflow(
	name=name,
	task_skill_names=task_skill_names,
	description=description,
	)
	skill = self._store_if_valid(proposal)
	self.library.save()
	return skill

	def build_role(
	self,
	name: str,
	workflow_skill_names: List[str],
	description: str = "",
	) -> Optional[Skill]:
	"""
	Build an L4 role (persona) — e.g. be_ceo, be_engineer, be_scout.

	The role's generated code emits 'role.enter' / 'role.exit' events
	and calls each workflow in sequence via call_skill().
	"""
	proposal = self.writer.propose_role(
	name=name,
	workflow_skill_names=workflow_skill_names,
	description=description,
	)
	skill = self._store_if_valid(proposal)
	self.library.save()
	return skill

	# ─────────────────────────────────────────────────────────────────────
	# Execution
	# ─────────────────────────────────────────────────────────────────────

	def execute_skill(
	self,
	skill_name: str,
	params: Optional[Dict[str, Any]] = None,
	resolve_deps: bool = True,
	) -> SkillExecutionResult:
	"""
	Execute a skill by name.

	If resolve_deps=True, pre-compiles the full dependency tree first.
	"""
	if resolve_deps:
	return self.executor.execute_with_dependencies(skill_name, params=params)
	return self.executor.execute_by_name(skill_name, params=params)

	# ─────────────────────────────────────────────────────────────────────
	# High-level: build the "be a CEO" persona from scratch
	# ─────────────────────────────────────────────────────────────────────

	def demo_build_ceo_persona(self) -> Dict[str, Any]:
	"""
	Full demo: build a complete skill tree from L0 atomics up to a L4
	"be_ceo" role, using only the VM primitives that already exist.

	This proves the full hierarchical composition pipeline works
	without any real LLM — purely through template-mode composition
	and validated execution.

	Returns a summary dict.
	"""
	# 1. Seed L0 atomics
	atomics = self.bootstrap_atomics()

	# 2. Build a few L1 compound skills from existing primitives
	compound_specs = [
	("click_somewhere", ["mouse_move", "left_click"]),
	("type_a_sentence", ["type_text", "press_key"]),
	("inspect_screen", ["screenshot", "list_windows"]),
	("run_command", ["execute_shell"]),
	("read_powershell", ["execute_powershell"]),
	]
	compounds: List[Optional[Skill]] = []
	for name, deps in compound_specs:
	s = self.teach_compound(name=name, dependency_skill_names=deps,
	description=f"Compound: {name.replace('_', ' ')}")
	compounds.append(s)

	# 3. Build L2 tasks from compounds
	task_specs = [
	("check_mail", ["inspect_screen", "click_somewhere", "type_a_sentence"]),
	("review_calendar", ["inspect_screen", "click_somewhere"]),
	("analyse_market", ["run_command", "inspect_screen"]),
	("make_a_decision", ["read_powershell", "type_a_sentence"]),
	("delegate_tasks", ["type_a_sentence", "click_somewhere"]),
	("strategic_thinking", ["inspect_screen", "read_powershell", "type_a_sentence"]),
	]
	tasks: List[Optional[Skill]] = []
	for name, deps in task_specs:
	t = self.build_task(name=name, dependency_skill_names=deps,
	description=f"Task: {name.replace('_', ' ')}")
	tasks.append(t)

	# 4. Build L3 workflows from tasks
	workflow_specs = [
	("morning_routine", ["check_mail", "review_calendar", "analyse_market"]),
	("strategic_planning", ["analyse_market", "strategic_thinking", "make_a_decision"]),
	("leadership_cycle", ["delegate_tasks", "strategic_thinking"]),
	("end_of_day", ["review_calendar", "delegate_tasks"]),
	]
	workflows: List[Optional[Skill]] = []
	for name, deps in workflow_specs:
	w = self.build_workflow(name=name, task_skill_names=deps,
	description=f"Workflow: {name.replace('_', ' ')}")
	workflows.append(w)

	# 5. Build the L4 role — be_ceo
	ceo = self.build_role(
	name="be_ceo",
	workflow_skill_names=[
	"morning_routine",
	"strategic_planning",
	"leadership_cycle",
	"end_of_day",
	],
	description="Persona: act as a CEO — orchestrate the full day cycle",
	)

	# 6. Summary
	return {
	"atomics_count": len([s for s in atomics if s is not None]),
	"compound_count": len([s for s in compounds if s is not None]),
	"task_count": len([s for s in tasks if s is not None]),
	"workflow_count": len([s for s in workflows if s is not None]),
	"role_built": ceo is not None,
	"ceo_skill_id": ceo.skill_id if ceo else None,
	"ceo_status": ceo.status.value if ceo else None,
	"ceo_dependencies": ceo.dependencies if ceo else [],
	"library_size": len(self.library),
	}

	# ─────────────────────────────────────────────────────────────────────
	# Status
	# ─────────────────────────────────────────────────────────────────────

	def get_status(self) -> Dict[str, Any]:
	return {
	"uptime_s": round(time.time() - self._created_at, 2),
	"library": self.library.get_stats(),
	"writer": self.writer.get_stats(),
	"validator": self.validator.get_stats(),
	"executor": self.executor.get_status(),
	"observer": self.observer.get_status(),
	}


	# ─────────────────────────────────────────────────────────────────────────────
	# Singleton
	# ─────────────────────────────────────────────────────────────────────────────


	_architect_instance: Optional[CodeArchitect] = None
	_architect_lock = threading.Lock()


	def get_code_architect(**kwargs) -> CodeArchitect:
	"""Get or create the singleton CodeArchitect."""
	global _architect_instance
	with _architect_lock:
	if _architect_instance is None:
	_architect_instance = CodeArchitect(**kwargs)
	return _architect_instance