orchestrata_ultimate.agents.volks.architect.volk

Reference › orchestrata_ultimate › agents › volks › architect › volk

ArchitectVolk (alias U:Ar, role Architect) is the principal designer of the pipeline (specifications 030 and 000). It extends BaseAgent (base) together with two mixins, ArchitectContextMixin (architect_context) and ArchitectProjectMixin (architect_project), and takes its prompt from architect_prompts. It is imported by the package __init__ and the kernel pipeline.

What it receives: an implement packet from the Project Manager (volk) or a ux_complete packet from the UX volk (volk); feedback packets from the Junior volk during design review; and review_edit packets from other volks. What it produces: the design documents (architecture, requirements, file tree, API spec, class diagram) in an implement packet addressed to the CodeDirector (U:Cd), plus design_review packets for J:Junior along the way.

_handle_design loads the structured fields from the Analyst (_load_structured_fields), resolves the project name and directory, builds the prompt from the specification and optional UX sections (_build_spec_section, _build_ux_section) and existing monorepo API sources, initialises the project (_init_project), calls the model through the tool loop with a budget of 16 turns, and parses the sections. When ORCHESTRATA_SKELETON_FLOW is 1 it first validates the Mermaid class diagram with the real skeleton parser (_validate_class_diagram). If every core section comes back blank it retries once, and a second blank result raises TaskExecutionError rather than sending an empty design to the Junior. Otherwise it writes the design documents and sends a review packet (_make_review_packet) to the Junior. _handle_feedback reads the verdict: when the Junior says READY, or the cycle count reaches MAX_DESIGN_CYCLES, _finalize_design writes the docs, commits them to git and escalates to U:Cd; otherwise _iterate_design answers the Junior's questions and loops.

_handle_review_edit is a deterministic floor with no LLM call. It parses the edit fragment with the helpers in _edit_review and returns BLOCK for a relative import, or for direct driver construction outside a factory module, and APPROVE otherwise, in a review_edit_result packet.

At a glance

Class diagram

Dependency diagram

API

Classes

ArchitectVolk (pydantic)

Bases: ArchitectProjectMixin, ArchitectContextMixin, BaseAgent · Decorators: traces(spec='000', story='US-000', reqs=['REQ-000-105']), traces(spec='030', story='US-007', reqs=['REQ-030-045']) · Line 151

Architect volk — designs architecture using deepseek-reasoner.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/agents/volks/architect/volk.py

# mypy: ignore-errors
"""U:Ar — Architect volk using deepseek-reasoner for design and specs."""

import logging
import os
import subprocess
import sys
import tempfile
from pathlib import Path

from pydantic import Field

from orchestrata_ultimate.agents.exceptions import TaskExecutionError
from orchestrata_ultimate.agents.volks._edit_review import (
    extract_edit_fields,
    has_direct_driver_construction,
    has_relative_import,
    parse_fragment,
    regex_has_direct_driver_construction,
    regex_has_relative_import,
)
from orchestrata_ultimate.agents.volks.architect.architect_context import (
    ArchitectContextMixin,
)
from orchestrata_ultimate.agents.volks.architect.architect_project import (
    ArchitectProjectMixin,
)
from orchestrata_ultimate.agents.volks.architect.architect_prompts import (
    ARCHITECT_UPDATE_TEMPLATE,
    MAX_DESIGN_CYCLES,
    get_architect_prompt_template,
)
from orchestrata_ultimate.agents.volks.base import BaseAgent
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph_directives import extract_graph_context
from orchestrata_ultimate.core.infra_config import PROJECT_WORKSPACE_ROOT
from orchestrata_ultimate.core.packet import CXPPacket, WorkStatus

_logger = logging.getLogger(__name__)


@traces(spec="030", story="US-002", reqs=["REQ-030-017"])
def _esc(s: str) -> str:
    """Escape braces so str.format() does not misinterpret content.

    Args:
        s: String that may contain literal curly braces.

    Returns:
        String with {{ and }} replacing { and }.
    """
    return s.replace("{", "{{").replace("}", "}}")


def _repo_root() -> Path:
    """Return the repository root (the directory containing ``scripts/``).

    ``get_project_root()`` stops at the package's bundled ``prolog/``
    directory, which is not the repo root, so this helper derives it from
    the installed ``orchestrata_ultimate`` package location instead.
    """
    import orchestrata_ultimate

    return Path(orchestrata_ultimate.__file__).resolve().parent.parent.parent


def _extract_mermaid_block(text: str) -> str | None:
    """Extract the first fenced `````mermaid classDiagram````` block.

    Args:
        text: The Architect response text.

    Returns:
        The diagram source inside the fence, or ``None`` if absent.
    """
    markers = ("```mermaid classDiagram", "```mermaid\nclassDiagram")
    for marker in markers:
        start = text.find(marker)
        if start != -1:
            after = start + len(marker)
            end = text.find("```", after)
            if end == -1:
                return None
            return text[after:end].strip()
    return None


def _validate_class_diagram(response_text: str) -> None:
    """Validate a Mermaid contract with the real gen-skeleton parser.

    When ``ORCHESTRATA_SKELETON_FLOW=1`` is set, U:Ar's CLASS_DIAGRAM block
    must be parseable by ``scripts/serialize_codebase_ast.py gen-skeleton``.
    A malformed contract is logged as a WARNING and the pipeline falls back
    to the freehand design path — it never crashes the run.

    Args:
        response_text: The raw LLM response containing the CLASS_DIAGRAM section.
    """
    block = _extract_mermaid_block(response_text)
    if block is None:
        _logger.warning(
            "SKELETON_FLOW: no fenced mermaid classDiagram block found; "
            "falling back to freehand design path"
        )
        return

    try:
        repo_root = _repo_root()
    except Exception as exc:  # pragma: no cover - defensive
        _logger.warning("SKELETON_FLOW: cannot locate repo root: %s", exc)
        return

    script_path = repo_root / "scripts" / "serialize_codebase_ast.py"
    with tempfile.TemporaryDirectory() as tmpdir:
        diagram_path = Path(tmpdir) / "contract.mmd"
        output_path = Path(tmpdir) / "skeleton.py"
        diagram_path.write_text(block, encoding="utf-8")
        try:
            result = subprocess.run(
                [
                    sys.executable,
                    str(script_path),
                    "gen-skeleton",
                    "--no-smoke",
                    "--output",
                    str(output_path),
                    str(diagram_path),
                ],
                cwd=str(repo_root),
                capture_output=True,
                text=True,
                timeout=30,
            )
        except Exception as exc:  # pragma: no cover - defensive
            _logger.warning(
                "SKELETON_FLOW: contract validation subprocess failed: %s",
                exc,
            )
            return

        if result.returncode != 0:
            detail = (result.stderr.strip() or result.stdout.strip())[:500]
            _logger.warning(
                "SKELETON_FLOW: contract failed gen-skeleton validation: %s",
                detail,
            )


@traces(spec="000", story="US-000", reqs=["REQ-000-105"])
@traces(spec="030", story="US-007", reqs=["REQ-030-045"])
class ArchitectVolk(ArchitectProjectMixin, ArchitectContextMixin, BaseAgent):
    """Architect volk — designs architecture using deepseek-reasoner."""

    alias: str = Field(default="U:Ar")
    role: str = "Architect"

    def __init__(self):
        super().__init__()
        self.escalation_target = "U:Cd"
        self.design_docs = {
            "architecture": "",
            "requirements": "",
            "file_tree": "",
            "api_spec": "",
            "class_diagram": "",
        }
        self.design_cycles = 0
        self.project_name = ""
        self.project_dir = ""
        self.task_summary = ""
        self._monorepo_src_section = ""

    @traces(
        spec="030",
        story="US-001",
        reqs=["REQ-030-001", "REQ-030-002"],
    )
    async def process(self, packet: CXPPacket) -> CXPPacket | None:
        """Route incoming CXP packets to the appropriate handler.

        Args:
            packet: Incoming CXP packet to process.

        Returns:
            Response CXP packet, or None if the subject is unhandled.
        """
        self._graph_context = extract_graph_context(packet.payload)
        if packet.subject in ("implement", "scope_validated", "ux_complete"):
            return await self._handle_design(packet)
        elif packet.subject == "design_feedback":
            return await self._handle_feedback(packet)
        elif packet.subject == "review_edit":
            return await self._handle_review_edit(packet)
        return None

    @traces(
        spec="030",
        story="US-001",
        reqs=[
            "REQ-030-003",
            "REQ-030-004",
            "REQ-030-005",
            "REQ-030-007",
            "REQ-030-009",
        ],
    )
    async def _handle_design(self, packet: CXPPacket) -> CXPPacket:
        """Create initial architecture from requirements."""
        self.status = WorkStatus.IN_PROGRESS
        self.current_thread = packet.thread_id

        requirements = packet.payload.get("requirements", "")
        # Retained (not just a local var) so _iterate_design can re-ground
        # later design-update calls in the ORIGINAL BA requirements text --
        # dogfood run-2 F3 defect 3: without this, _iterate_design only had
        # its OWN prior (possibly blank) parsed requirements section to work
        # from.
        self._requirements = requirements
        self.task_summary = packet.payload.get("task_summary", "")
        self._original_intent = packet.payload.get("original_intent", "")
        self._ux_design = packet.payload.get("ux_design", "")
        self._scope_data = packet.payload.get("scope_data", {})
        self._load_structured_fields(packet)
        self._resolve_project_name(packet)

        self.summary = "Designing architecture"
        print(
            f"  [U:Ar] Designing architecture for: {self.task_summary}",
            flush=True,
        )
        print(f"  [U:Ar] Project dir: {self.project_dir}", flush=True)
        self._init_project()

        original_spec_section = _build_spec_section(self._original_intent)
        full_monorepo_context = await self._build_full_monorepo_context(requirements)

        prompt = get_architect_prompt_template().format(
            task_summary=self.task_summary,
            requirements=requirements + _build_ux_section(self._ux_design),
            original_spec_section=original_spec_section,
            monorepo_src_section=full_monorepo_context,
        )
        from orchestrata_ultimate.agents.volks.persona import (  # noqa: PLC0415
            compose_persona,
        )

        # W4.3a: route the primary design-generation call through the
        # tool-calling loop (SPEC-103) with U:Ar's registered tools
        # (read_file/search_code/ast_query/list_dir --
        # tool_registry.AGENT_TOOLS["U:Ar"]) so the Architect can inspect the
        # real codebase while designing instead of reasoning from the prompt
        # text alone. This replaces the direct ``call_reasoner`` call (still
        # used by ``_iterate_design`` below), which has no tool support.
        #
        # ``call_reasoner`` hardcodes its own system prompt and puts the
        # whole persona-composed prompt in the user turn (see the removed
        # comment this replaces); ``_call_llm_with_tools`` takes system_prompt
        # and user_message explicitly, so that same split is reproduced here.
        # It also takes no ``graph_context=`` kwarg -- unlike ``call_reasoner``
        # /``call_llm``, the tool loop only forwards temperature/model to
        # ``call_with_tools`` -- so the graph context is folded directly into
        # the user turn, matching the injection ``call_llm`` performs
        # internally (``graph_context + "\n\n---\n\n" + user_message``).
        user_message = compose_persona(prompt)
        if self._graph_context:
            user_message = self._graph_context + "\n\n---\n\n" + user_message
        # Dogfood run-4/F13: grounded design work against a real codebase
        # routinely burns 6-9 exploration turns before the model is ready to
        # compose, so the default ``max_turns=10`` budget was found too tight
        # -- the finalization call would fire mid-exploration and return a
        # truncated mid-thought fragment instead of a real design. Only the
        # primary design-generation call gets the larger budget; the retry
        # path in ``_iterate_design`` below is unaffected.
        response = await self._call_llm_with_tools(
            "You are a deep reasoning architect. Think step by step.",
            user_message,
            max_turns=16,
            temperature=1.0,
        )
        self.output_artifacts.append(f"=== INITIAL ARCHITECTURE ===\n{response}")

        if os.environ.get("ORCHESTRATA_SKELETON_FLOW", "") == "1":
            _validate_class_diagram(response)

        self._parse_design(response)
        if self._core_sections_blank():
            # Dogfood run-2 F3 defect 2: a blank response was previously
            # written straight to design/ and shipped to J:Junior for review
            # (Junior's own review found "Architecture, Requirements, File
            # Tree, and API Specification sections ... are all blank"). Retry
            # ONCE through the same tool-loop path before failing loud --
            # mirrors the fail-loud doctrine already applied to J:Junior
            # codegen failures (CodeDirectorVolk._handle_junior_complete).
            _logger.warning(
                "U:Ar: initial design response had all core sections blank "
                "-- retrying once"
            )
            response = await self._call_llm_with_tools(
                "You are a deep reasoning architect. Think step by step.",
                user_message,
                temperature=1.0,
            )
            self.output_artifacts.append(
                f"=== INITIAL ARCHITECTURE (retry) ===\n{response}"
            )
            self._parse_design(response)
            if self._core_sections_blank():
                raise TaskExecutionError(
                    "U:Ar: design response had all core sections "
                    "(architecture/requirements/file_tree/api_spec) blank "
                    "after retry -- refusing to send a blank design to "
                    "J:Junior for review",
                    task_id=packet.thread_id,
                )
        self._write_design_docs()
        self.design_cycles = 1
        print(
            "  [U:Ar] Initial design complete. Sending to J:Junior for review...",
            flush=True,
        )
        return self._make_review_packet(packet.thread_id)

    @traces(
        spec="030",
        story="US-002",
        reqs=["REQ-030-010", "REQ-030-011", "REQ-030-012", "REQ-030-013"],
    )
    async def _handle_feedback(self, packet: CXPPacket) -> CXPPacket:
        """Handle design feedback from J:Junior."""
        verdict = packet.payload.get("verdict", "")
        questions = packet.payload.get("questions", [])
        cycle = packet.payload.get("design_cycle", self.design_cycles)

        if verdict == "READY" or cycle >= MAX_DESIGN_CYCLES:
            return self._finalize_design(packet, cycle, verdict)
        return await self._iterate_design(packet, cycle, questions)

    @traces(
        spec="030",
        story="US-002",
        reqs=["REQ-030-014", "REQ-030-015", "REQ-030-016"],
    )
    def _finalize_design(
        self, packet: CXPPacket, cycle: int, verdict: str
    ) -> CXPPacket:
        """Finalize and escalate approved design to U:Cd."""
        if cycle >= MAX_DESIGN_CYCLES and verdict != "READY":
            print(
                f"  [U:Ar] Max design cycles ({MAX_DESIGN_CYCLES}) reached -- proceeding",
                flush=True,
            )
        else:
            print(f"  [U:Ar] Junior is READY after {cycle} cycle(s)", flush=True)

        self.summary = f"Design complete ({cycle} cycles)"
        self._write_design_docs()
        self._git_commit(f"Design finalized after {cycle} review cycle(s)")
        return CXPPacket(
            from_agent="U:Ar",
            to_agent="U:Cd",
            thread_id=packet.thread_id,
            subject="implement",
            payload={
                "architecture": self.design_docs["architecture"],
                "requirements": self.design_docs["requirements"],
                "file_tree": self.design_docs["file_tree"],
                "api_spec": self.design_docs["api_spec"],
                "class_diagram": self.design_docs["class_diagram"],
                "original_intent": self._original_intent,
                "project_dir": self.project_dir,
                "task_summary": self.task_summary,
                "structured_requirements": self._structured_requirements,
                "user_stories": self._user_stories,
                # Forward monorepo API sources to U:Cd so J:Aider sees real constructors
                "monorepo_src_section": self._monorepo_src_section,
                "nfrs": self._nfrs,
                "spec_metadata": self._spec_metadata,
                # Capability-routed hop: the Architect needs *implementation*
                # next. The kernel resolves this to whichever volk provides it
                # (U:Cd by default), so ``to_agent`` above is the fallback, not a
                # hardcoded destination. See PipelineEscalationMixin.
                "needs_capability": "implementation",
            },
            work_status=WorkStatus.IN_PROGRESS,
        )

    @traces(
        spec="030",
        story="US-002",
        reqs=["REQ-030-018", "REQ-030-019"],
    )
    async def _iterate_design(
        self, packet: CXPPacket, cycle: int, questions: list
    ) -> CXPPacket:
        """Update design in response to junior's questions.

        Args:
            packet: The incoming feedback packet.
            cycle: Current design cycle number.
            questions: List of blocking questions from J:Junior.

        Returns:
            Updated design packet addressed back to J:Junior.
        """
        self.design_cycles = cycle + 1
        self.summary = (
            f"Updating design (cycle {self.design_cycles})"
            f" -- {len(questions)} questions"
        )
        print(
            f"  [U:Ar] Updating design for {len(questions)} questions "
            f"(cycle {self.design_cycles})...",
            flush=True,
        )
        questions_text = "\n".join(f"{i + 1}. {q}" for i, q in enumerate(questions))
        prompt = ARCHITECT_UPDATE_TEMPLATE.format(
            task_summary=self.task_summary,
            requirements=_esc(getattr(self, "_requirements", "")),
            architecture=_esc(self.design_docs["architecture"]),
            requirements_doc=_esc(self.design_docs["requirements"]),
            file_tree=_esc(self.design_docs["file_tree"]),
            api_spec=_esc(self.design_docs["api_spec"]),
            class_diagram=_esc(self.design_docs["class_diagram"]),
            questions=questions_text,
            monorepo_src_section=_esc(getattr(self, "_monorepo_src_section", "")),
        )
        from orchestrata_ultimate.agents.volks.persona import (  # noqa: PLC0415
            compose_persona,
        )

        # Dogfood run-2 F3 defect 3: plain ``call_reasoner`` has no tool
        # support and, when the prior design was blank/truncated, nothing
        # left to ground it in the real task -- on run 2 it hallucinated an
        # unrelated ("books API") design in response to "your design is
        # blank" questions. Mirrors the W4.3a routing in _handle_design
        # above: ``_call_llm_with_tools`` takes system_prompt/user_message
        # explicitly (unlike ``call_reasoner``, which hardcodes its own
        # system prompt and puts everything in the user turn) and takes no
        # ``graph_context=`` kwarg, so the graph context is folded directly
        # into the user turn here too, matching ``call_reasoner``'s internal
        # injection (``graph_context + "\n\n---\n\n" + user_message``). The
        # template above also now carries the ORIGINAL task_summary + BA
        # requirements text (not just this volk's own prior, possibly blank,
        # parsed design) so the iteration stays grounded even when the prior
        # design was empty.
        user_message = compose_persona(prompt)
        if self._graph_context:
            user_message = self._graph_context + "\n\n---\n\n" + user_message
        response = await self._call_llm_with_tools(
            "You are a deep reasoning architect. Think step by step.",
            user_message,
            temperature=1.0,
        )
        self.output_artifacts.append(
            f"=== DESIGN UPDATE (cycle {self.design_cycles}) ===\n{response}"
        )
        self._parse_design(response)
        self._write_design_docs()
        return self._make_review_packet(packet.thread_id)

    @traces(
        spec="030",
        story="US-003",
        reqs=["REQ-030-020", "REQ-030-021"],
    )
    def _make_review_packet(self, thread_id: str) -> CXPPacket:
        """Build a design_review CXP packet for J:Junior.

        Args:
            thread_id: Thread ID to attach to the packet.

        Returns:
            CXP packet ready to send to J:Junior.
        """
        return CXPPacket(
            from_agent="U:Ar",
            to_agent="J:Junior",
            thread_id=thread_id,
            subject="design_review",
            payload={
                "architecture": self.design_docs["architecture"],
                "requirements": self.design_docs["requirements"],
                "file_tree": self.design_docs["file_tree"],
                "api_spec": self.design_docs["api_spec"],
                "class_diagram": self.design_docs["class_diagram"],
                "design_cycle": self.design_cycles,
                "project_dir": self.project_dir,
                "task_summary": self.task_summary,
            },
            work_status=WorkStatus.IN_PROGRESS,
        )

    # --- Private helpers: prompt-section assembly ---

    @traces(
        spec="030",
        story="US-004",
        reqs=["REQ-030-025", "REQ-030-026"],
    )
    def _load_structured_fields(self, packet: CXPPacket) -> None:
        """Extract structured BA fields from packet payload.

        Args:
            packet: CXP packet whose payload contains BA analysis output.
        """
        impact_report = packet.payload.get("parameters", {}).get("_impact_report", {})
        if impact_report:
            print(
                f"  [U:Ar] Impact analysis: {impact_report.get('classification')} - "
                f"{impact_report.get('severity')} severity, "
                f"{len(impact_report.get('affected_files', []))} affected files",
                flush=True,
            )
        self._structured_requirements = packet.payload.get(
            "structured_requirements", []
        )
        self._user_stories = packet.payload.get("user_stories", [])
        self._nfrs = packet.payload.get("nfrs", [])
        self._spec_metadata = packet.payload.get("spec_metadata", {})
        self._analyst_output = packet.payload.get("analyst_output", "")

    @traces(
        spec="030",
        story="US-004",
        reqs=["REQ-030-027", "REQ-030-028"],
    )
    def _resolve_project_name(self, packet: CXPPacket) -> None:
        """Set project_name and project_dir from packet or task summary.

        Args:
            packet: CXP packet that may carry a concierge-supplied project name.
        """
        concierge_name = packet.payload.get("project_name")
        if concierge_name:
            self.project_name = self._slugify(concierge_name)
        else:
            self.project_name = self._slugify(self.task_summary)
        self.project_dir = os.path.join(PROJECT_WORKSPACE_ROOT, self.project_name)

    @traces(spec="050", story="US-001", reqs=["REQ-050-FR1-001"])
    async def _handle_review_edit(self, packet: CXPPacket) -> CXPPacket:
        """SPEC-050 P3.1 fail-closed FLOOR for architectural compliance.

        Deterministic, no LLM call — not a correctness review. Approves
        unless the new code violates one of two hard architecture rules
        (see CLAUDE.md § Key Rules / scripts/architecture_checker.py):
          - absolute-imports-only: no ``from .`` / ``from ..`` relative
            imports.
          - no direct driver construction outside a factory module: no
            ``GraphDatabase.driver(``/``redis.Redis(`` calls unless the
            target file path itself looks like a factory module.

        Checks run against the parsed AST when ``new_code`` parses (via
        ``_edit_review.parse_fragment``, which tolerates indented
        fragments) so a docstring/comment merely mentioning ``from .foo``
        or ``GraphDatabase.driver(`` never triggers a false BLOCK; the
        regex fallback only fires for genuinely unparseable fragments.
        """
        file_path, _old_code, new_code = _extract_edit_fields(packet.payload)
        verdict = "APPROVE"
        reason = "No architecture violations detected."

        new_tree, _parse_error = parse_fragment(new_code)
        if new_tree is not None:
            relative_import = has_relative_import(new_tree)
            direct_driver = has_direct_driver_construction(new_tree)
        else:
            relative_import = regex_has_relative_import(new_code)
            direct_driver = regex_has_direct_driver_construction(new_code)

        if relative_import:
            verdict = "BLOCK"
            reason = "Edit introduces a relative import (absolute imports only)."
        elif direct_driver and "factory" not in file_path.lower():
            verdict = "BLOCK"
            reason = "Edit constructs a driver directly outside a factory module."

        return CXPPacket(
            from_agent=self.alias,
            to_agent=packet.from_agent,
            thread_id=packet.thread_id,
            subject="review_edit_result",
            payload={
                "verdict": verdict,
                "reason": reason,
                "content": reason,
                "file_path": file_path,
            },
            work_status=WorkStatus.RESOLVED,
        )


def _extract_edit_fields(payload: dict) -> tuple[str, str, str]:
    """Extract (file_path, old_code, new_code) from a review_edit payload.

    Thin re-export of the shared helper (see
    ``orchestrata_ultimate.agents.volks._edit_review.extract_edit_fields``)
    kept under this name for backward compatibility with existing call
    sites/tests in this module.
    """
    return extract_edit_fields(payload)


@traces(spec="030", story="US-001", reqs=["REQ-030-006"])
def _build_spec_section(original_intent: str) -> str:
    """Build the ORIGINAL_SPEC section string for the design prompt.

    Args:
        original_intent: The original intent text to include in the prompt.

    Returns:
        Formatted spec section with brace-escaped content, or empty string.
    """
    if not original_intent:
        return ""
    escaped = original_intent.replace("{", "{{").replace("}", "}}")
    return "\n\nORIGINAL_SPEC (from human -- use these exact signatures):\n" + escaped


@traces(spec="030", story="US-001", reqs=["REQ-030-006"])
def _build_ux_section(ux_design: str) -> str:
    """Build the UX DESIGN section string for the design prompt.

    Args:
        ux_design: The UX design text to incorporate.

    Returns:
        Formatted UX section with brace-escaped content, or empty string.
    """
    if not ux_design:
        return ""
    escaped = ux_design.replace("{", "{{").replace("}", "}}")
    print("  [U:Ar] Incorporating UX design into architecture", flush=True)
    return (
        "\n\nUX DESIGN (from U:Ux -- incorporate these workflows, "
        "components, wireframes, and design tokens into the "
        "architecture):\n" + escaped
    )