Universal Orchestrator — Event Routing Architecture¶

EVENT_ROUTE_MAP = {
    # Build pipeline
    "ui.build_confirmed":           "DecompositionGroupChat",
    "ui.deploy_clicked":            "DeploymentGroupChat",
    "ui.sandbox_extend":            "e2b.extend_sandbox",          # no AI
    "ui.sandbox_kill":              "e2b.kill_sandbox",            # no AI

    # Settings changes that affect the built app
    "rest.settings.tech_stack":     "DecompositionGroupChat",      # full re-decompose
    "rest.settings.db_type":        "DecompositionGroupChat",      # affects schema tasks
    "rest.settings.app_name":       "PatchAgent",                  # surface rename

    # Account / subscription
    "rest.subscription.upgraded":   "ValueEngineGroupChat",        # suggest expanded features
    "rest.subscription.downgraded": "ValueEngineGroupChat",        # trim features to tier

    # System / lifecycle
    "system.build_completed":       "e2b.run_preview",             # no AI
    "system.sandbox_expired":       "notify.user",                 # no AI
    "webhook.stripe.payment":       "rest.billing_handler",        # no AI
}

CHANGE_TYPE_ROUTE_MAP = {
    "FOUNDATIONAL": "ValueEngineGroupChat",
    "STRUCTURAL":   "DecompositionGroupChat",
    "FEATURE":      "FeedbackGroupChat",
    "SURFACE":      "FeedbackGroupChat",
}

from autogen.agentchat.group.conditions import StringLLMCondition

# One condition per ChangeType. Evaluated highest-severity first.
foundational_condition = StringLLMCondition(
    prompt=(
        "The user submitted this message: '{user_text}'\n"
        "The current process state: {process_context_summary}\n\n"
        "Does this require restarting the entire process from the beginning, "
        "changing the fundamental goal, or invalidating >50% of prior work?\n"
        "Answer only YES or NO."
    )
)

structural_condition = StringLLMCondition(
    prompt=(
        "The user submitted this message: '{user_text}'\n"
        "Current domains / modules: {domains}\n\n"
        "Does this add, remove, or fundamentally rename a major domain or module "
        "without changing the overall goal?\n"
        "Answer only YES or NO."
    )
)

# FEATURE and SURFACE: apply only if FOUNDATIONAL and STRUCTURAL are NO

from autogen.events.base_event import BaseEvent, wrap_event
from autogen.io.base import IOStream

@wrap_event
class DAGTaskCompletedEvent(BaseEvent):
    """Emitted by a task agent tool when it finishes writing its output file."""
    task_id: str
    task_type: str
    domain: str
    output_files: list[str]
    duration_seconds: float

    def print(self, f=None):
        f = resolve_print_callable(f)
        f(f"[DAG] {self.task_type} ({self.domain}) completed in {self.duration_seconds:.1f}s")

# Inside write_output_files() tool that every task agent calls:
IOStream.get_default().send(
    DAGTaskCompletedEvent(
        task_id=task.task_id,
        task_type=task.task_type,
        domain=task.domain,
        output_files=written_files,
        duration_seconds=elapsed,
    )
)

async def _run_layer(self, tasks: list[DAGTask]) -> list[DAGTaskCompletedEvent]:
    """Run all tasks in a dependency layer concurrently."""
    async def consume(task):
        async for event in task_agent.a_run_iter(
            message=build_prompt(task),
            yield_on=[DAGTaskCompletedEvent],
        ):
            if isinstance(event, DAGTaskCompletedEvent):
                # inject output into ContextVariables for dependent tasks
                self.context[event.content.task_id] = event.content
                # forward to WebSocket → build stepper UI updates live
                await self.websocket.send(event.content)
                return event.content
    return await asyncio.gather(*[consume(t) for t in tasks])

# Main execution loop — replaces ~100 lines of current dag_executor.py:
for layer in topological_layers(self.dag):
    await self._run_layer(layer)
# When gather resolves → all deps met → next layer fires automatically

from typing import Any, Callable
from autogen.events.base_event import BaseEvent, wrap_event, resolve_print_callable

@wrap_event
class DAGTaskCompletedEvent(BaseEvent):
    """Emitted by a task agent's write_output tool when it finishes.
    DAGExecutor yields on this to know when to advance the DAG.
    WebSocket forwards this to the frontend build stepper.
    """
    task_id: str
    task_type: str
    domain: str
    output_files: list[str]
    duration_seconds: float
    context_key: str           # key under which output is stored in ContextVariables

    def print(self, f: Callable[..., Any] | None = None) -> None:
        f = resolve_print_callable(f)
        f(f"[DAG] {self.task_type} ({self.domain}) ✓ {self.duration_seconds:.1f}s", flush=True)


@wrap_event
class DAGLayerCompletedEvent(BaseEvent):
    """Emitted by DAGExecutor when all tasks in a dependency layer finish.
    Not forwarded to WebSocket — internal signal only.
    """
    layer_index: int
    task_ids: list[str]

    def print(self, f: Callable[..., Any] | None = None) -> None:
        f = resolve_print_callable(f)
        f(f"[DAG] Layer {self.layer_index} complete ({len(self.task_ids)} tasks)", flush=True)


@wrap_event
class BuildCompletedEvent(BaseEvent):
    """Emitted by runner.py when the full DAG finishes.
    Triggers E2B sandbox provisioning automatically.
    """
    enterprise_id: str
    output_dir: str
    total_tasks: int
    total_duration_seconds: float

    def print(self, f: Callable[..., Any] | None = None) -> None:
        f = resolve_print_callable(f)
        f(f"[Build] Complete — {self.total_tasks} tasks in {self.total_duration_seconds:.0f}s", flush=True)

from autogen.agentchat.group import ContextVariables
from build_pipeline.core.events import DAGTaskCompletedEvent, DAGLayerCompletedEvent
from autogen.io.base import IOStream

class DAGExecutor:
    def __init__(self, dag: BuildDAG, llm_config, output_dir, websocket=None):
        self.dag = dag
        self.llm_config = llm_config
        self.output_dir = output_dir
        self.websocket = websocket
        # AG2 ContextVariables replaces our hand-rolled ContextStore
        self.context = ContextVariables(data={})

    async def run(self) -> BuildRunResult:
        for layer_index, layer in enumerate(topological_layers(self.dag)):
            completed = await self._run_layer(layer_index, layer)
        return BuildRunResult(output_dir=self.output_dir, ...)

    async def _run_layer(self, layer_index: int, tasks: list[DAGTask]):
        async def consume(task: DAGTask):
            agent = self._build_agent(task)   # ConversableAgent with UpdateSystemMessage
            async for event in agent.a_run_iter(
                message=self._build_prompt(task),
                yield_on=[DAGTaskCompletedEvent],
            ):
                if isinstance(event, DAGTaskCompletedEvent):
                    # Store output for downstream tasks
                    self.context[event.content.context_key] = event.content
                    # Forward to WebSocket → build stepper UI
                    if self.websocket:
                        await self.websocket.send_json(event.content.__dict__)
                    return event.content

        results = await asyncio.gather(*[consume(t) for t in tasks])
        return results

    def _build_agent(self, task: DAGTask) -> ConversableAgent:
        role = _role_registry.get(task.task_type)
        return ConversableAgent(
            name=f"{task.task_type}_{task.domain}",
            llm_config=self.llm_config,
            # UpdateSystemMessage injects prior task outputs automatically
            update_agent_state_before_reply=[
                UpdateSystemMessage(role.base_prompt + self._context_template(task))
            ],
        )

    def _context_template(self, task: DAGTask) -> str:
        # Builds {context_key} slots from task.consumes_context
        # AG2 substitutes actual values from self.context before each reply
        return "\n".join(
            f"Prior output ({key}): {{{key}}}"
            for key in (task.consumes_context or [])
        )

# When user closes browser mid-build:
manager_state = groupchat_manager.export_state()   # AG2 built-in
await mongodb.save(enterprise_id, "session_state", manager_state)

# When user reopens:
state = await mongodb.load(enterprise_id, "session_state")
groupchat_manager.resume(messages=state)           # AG2 built-in — continues exactly where left off