orchestrata_ultimate.integrations.cheetahclaws

Reference › orchestrata_ultimate › integrations › cheetahclaws › init

This module (524 lines) implements the CheetahClaws integration of specification 050: UltimateTUI adapts the CheetahClaws terminal UI to Orchestrata's agents. The gateway's terminal-UI handler imports it, and it uses the base agent (base) type for the reviewing volks.

route_command maps a slash command to a handler; handle_edit is the main path. It takes a file path, old and new strings and a strict flag. First GovernanceBridge.can_edit applies the file-operation policy and returns an allow flag with a reason; it rejects files that do not exist, directories, oversized or binary files (_is_binary checks for null bytes), secret-looking names (_contains_secrets) and blocked path patterns. If allowed, MultiAgentCoordinator.dispatch_review sends the edit as CXP packets, in parallel and with a timeout (default 30 seconds), to the code, test and architecture volks, and to the UX volk when the file looks like frontend code (_is_frontend_file). Each agent is lazily created and cached (_get_agent), the per-agent prompt comes from _build_edit_prompt, and each reply is parsed into an AgentFeedback with an AgentVerdict of approve, block, suggest, timeout or error (_parse_agent_response). rejecting_feedback is the fail-closed gate: block, timeout and error all withhold approval. The result is an EditResult with success, diff, bytes changed and the feedback. handle_intent runs the full agent pipeline for an /intent command, and _handle_write, _handle_agents and _handle_model cover the write, agent-list and profile-switch commands.

At a glance

Class diagram

Dependency diagram

API

Classes

AgentVerdict (enum)

Bases: StrEnum · Line 52

AgentFeedback (dataclass)

Decorators: traces(spec='050', story='US-001', reqs=['REQ-050-001', 'REQ-050-FR1-001']), dataclass · Line 62

EditRequest (dataclass)

Decorators: traces(spec='050', story='US-001', reqs=['REQ-050-001', 'REQ-050-FR1-001']), dataclass · Line 73

EditResult (dataclass)

Decorators: traces(spec='050', story='US-001', reqs=['REQ-050-001', 'REQ-050-FR1-001']), dataclass · Line 83

MultiAgentCoordinator

Decorators: traces(spec='050', story='US-001', reqs=['REQ-050-001', 'REQ-050-FR1-001']) · Line 99

Dispatches CXP packets to multiple agents in parallel and collects results.

GovernanceBridge

Decorators: traces(spec='050', story='US-001', reqs=['REQ-050-FR2-001']) · Line 263

PAM++ policy enforcement for CheetahClaws file operations.

UltimateTUI

Decorators: traces(spec='050', story='US-001', reqs=['REQ-050-FR1-001', 'REQ-050-FR2-001', 'REQ-050-FR9-001']) · Line 332

CheetahClaws TUI adapted for Orchestrata multi-agent coordination.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/integrations/cheetahclaws/__init__.py

"""CheetahClaws Integration — UltimateTUI wiring CC commands to Orchestrata agents.

SPEC-050 FR-001 through FR-015.

Architecture:
    CC Slash Command → UltimateTUI → CommandRouter → CXP Dispatcher
                                              │
    ┌─────────────────────────────────────────┤
    │                                         │
    ▼                                         ▼
  Governance (PAM++)              Multi-Agent Dispatch
  (pre-edit policy check)         (parallel Cd, Qa, Ar, Ux)
                                          │
                              ┌───────────┼───────────┐
                              ▼           ▼           ▼           ▼
                           U:Cd        U:Qa        U:Ar        U:Ux
                        (correctness) (testing)  (arch)      (ux)
                              │           │           │           │
                              └───────────┴───────────┴───────────┘
                                          │
                                    ALL approve?
                                    ├── YES → apply edit, create checkpoint
                                    └── NO  → reject with agent feedback
"""

from __future__ import annotations

import asyncio
import logging
import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import StrEnum
from pathlib import Path
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket, WorkStatus

if TYPE_CHECKING:
    from orchestrata_ultimate.agents.volks.base import BaseAgent
    from orchestrata_ultimate.postman.postman import Postman

logger = logging.getLogger(__name__)


# ═══════════════════════════════════════════════════════════════════════
# Data Models (SPEC-050 § Data Models)
# ═══════════════════════════════════════════════════════════════════════


class AgentVerdict(StrEnum):
    APPROVE = "approve"
    BLOCK = "block"
    SUGGEST = "suggest"
    TIMEOUT = "timeout"
    ERROR = "error"


@traces(spec="050", story="US-001", reqs=["REQ-050-001", "REQ-050-FR1-001"])
@dataclass
class AgentFeedback:
    agent_id: str
    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
    verdict: AgentVerdict = AgentVerdict.APPROVE
    content: str = ""
    tools_used: list[str] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)


@traces(spec="050", story="US-001", reqs=["REQ-050-001", "REQ-050-FR1-001"])
@dataclass
class EditRequest:
    file_path: str
    old_string: str
    new_string: str
    strict: bool = True
    thread_id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])


@traces(spec="050", story="US-001", reqs=["REQ-050-001", "REQ-050-FR1-001"])
@dataclass
class EditResult:
    success: bool
    file_path: str
    diff: str = ""
    bytes_changed: int = 0
    agents_responded: list[str] = field(default_factory=list)
    feedback: list[AgentFeedback] = field(default_factory=list)
    error: str = ""


# ═══════════════════════════════════════════════════════════════════════
# Multi-Agent Dispatch (SPEC-050 FR-001)
# ═══════════════════════════════════════════════════════════════════════


@traces(spec="050", story="US-001", reqs=["REQ-050-001", "REQ-050-FR1-001"])
class MultiAgentCoordinator:
    """Dispatches CXP packets to multiple agents in parallel and collects results.

    SPEC-050 FR-001: Coordinate file edits across Cd, Qa, Ar, and Ux agents
    in parallel, requiring approval from all before committing changes.
    """

    # Agents to dispatch for each command type
    EDIT_AGENTS = frozenset({"U:Cd", "U:Qa", "U:Ar"})
    EDIT_AGENTS_WITH_UX = frozenset({"U:Cd", "U:Qa", "U:Ar", "U:Ux"})
    INTENT_AGENTS = frozenset({"U:Cx"})
    REVIEW_AGENTS = frozenset({"U:Qa", "U:Ar", "U:Cd"})
    WRITE_AGENTS = frozenset({"U:Cd", "U:Qa"})

    # Verdicts that do NOT constitute reviewer approval (SPEC-050 FR-001:
    # "requiring approval from all agents before committing changes").
    # SUGGEST is non-blocking feedback alongside APPROVE. Everything else —
    # including TIMEOUT/ERROR — withholds approval, so a "review_edit"
    # dispatch that no volk actually handles (no explicit APPROVE) fails
    # CLOSED instead of silently letting the edit through.
    NON_APPROVING_VERDICTS = frozenset(
        {
            AgentVerdict.BLOCK,
            AgentVerdict.TIMEOUT,
            AgentVerdict.ERROR,
        }
    )

    def __init__(self, postman: Postman | None = None):
        self._postman = postman
        self._agent_instances: dict[str, Any] = {}

    @traces(spec="050", story="US-001", reqs=["REQ-050-FR1-001"])
    def _get_agent(self, alias: str) -> BaseAgent:
        """Lazy-load and cache agent instances."""
        if alias not in self._agent_instances:
            from orchestrata_ultimate.orchestrator.pipeline_router import PipelineRouter

            router = PipelineRouter()
            cls = router.get_agent_class(alias)
            self._agent_instances[alias] = cls()
        agent: BaseAgent = self._agent_instances[alias]
        return agent

    @traces(spec="050", story="US-001", reqs=["REQ-050-FR1-001"])
    async def dispatch_review(
        self,
        edit: EditRequest,
        agents: frozenset[str] | None = None,
        timeout: float = 30.0,
    ) -> list[AgentFeedback]:
        """Dispatch edit review to multiple agents in parallel.

        Each agent receives a CXP packet with the edit context and returns
        a verdict: APPROVE, BLOCK, SUGGEST, TIMEOUT, or ERROR.

        Args:
            edit: The edit request to review.
            agents: Set of agent aliases (default: EDIT_AGENTS).
            timeout: Max seconds to wait for each agent.

        Returns:
            List of AgentFeedback — one per dispatched agent.
        """
        agents = agents or self.EDIT_AGENTS
        prompt = self._build_edit_prompt(edit)

        async def _ask(alias: str) -> AgentFeedback:
            try:
                agent = self._get_agent(alias)
                packet = CXPPacket(
                    from_agent="CC-Agent",
                    to_agent=alias,
                    thread_id=edit.thread_id,
                    subject="review_edit",
                    payload={"edit": edit.__dict__, "prompt": prompt[alias]},
                    work_status=WorkStatus.PENDING,
                )
                response = await asyncio.wait_for(
                    agent.process(packet), timeout=timeout
                )
                if response is None:
                    return AgentFeedback(agent_id=alias, verdict=AgentVerdict.TIMEOUT)
                return self._parse_agent_response(alias, response)
            except TimeoutError:
                return AgentFeedback(agent_id=alias, verdict=AgentVerdict.TIMEOUT)
            except Exception as e:
                return AgentFeedback(
                    agent_id=alias, verdict=AgentVerdict.ERROR, errors=[str(e)]
                )

        tasks = [_ask(a) for a in agents]
        return await asyncio.gather(*tasks)

    @traces(spec="050", story="US-001", reqs=["REQ-050-FR1-001"])
    def _build_edit_prompt(self, edit: EditRequest) -> dict[str, str]:
        """Build per-agent review prompts."""
        base = (
            f"FILE: {edit.file_path}\n"
            f"REPLACE:\n  -{edit.old_string}\n  +{edit.new_string}\n"
        )
        return {
            "U:Cd": base
            + "Review for code correctness, patterns, standards. Reply: APPROVE/BLOCK/SUGGEST + reason.",
            "U:Qa": base
            + "Review for test coverage, error handling, edge cases. Reply: APPROVE/BLOCK/SUGGEST + reason.",
            "U:Ar": base
            + "Review for architecture fit, module boundaries, dependency direction. Reply: APPROVE/BLOCK/SUGGEST + reason.",
            "U:Ux": base
            + "Review for UI/UX consistency, accessibility, design tokens. Reply: APPROVE/BLOCK/SUGGEST + reason.",
        }

    @traces(spec="050", story="US-001", reqs=["REQ-050-FR1-001"])
    def _parse_agent_response(self, alias: str, packet: CXPPacket) -> AgentFeedback:
        """Parse agent's CXP response into structured feedback."""
        payload = packet.payload
        content = payload.get("content", payload.get("review_result", str(payload)))
        verdict_str = payload.get("verdict", "").upper()

        if "BLOCK" in verdict_str or "BLOCK" in content[:50].upper():
            verdict = AgentVerdict.BLOCK
        elif "SUGGEST" in verdict_str:
            verdict = AgentVerdict.SUGGEST
        else:
            verdict = AgentVerdict.APPROVE

        return AgentFeedback(
            agent_id=alias,
            verdict=verdict,
            content=content[:2000],
            tools_used=payload.get("tools_used", []),
        )

    @traces(spec="050", story="US-001", reqs=["REQ-050-FR1-001"])
    def rejecting_feedback(self, feedback: list[AgentFeedback]) -> list[AgentFeedback]:
        """FAIL-CLOSED edit-approval gate (SPEC-050 FR-001).

        An edit may only be committed once every dispatched reviewer has
        actually approved it. This returns the feedback entries that
        withhold approval; an empty result means every agent APPROVEd (or
        merely SUGGESTed, which is non-blocking).

        Critically, a "review_edit" dispatch that no volk handles comes
        back as TIMEOUT (see ``dispatch_review``'s ``response is None``
        branch) — that is NOT an approval, so TIMEOUT (and ERROR) verdicts
        are treated as withholding approval exactly like an explicit BLOCK.
        Without this, an unhandled/failed review would silently let every
        edit through (approval theater).

        Args:
            feedback: Feedback collected from ``dispatch_review``.

        Returns:
            The subset of feedback whose verdict withholds approval.
        """
        return [f for f in feedback if f.verdict in self.NON_APPROVING_VERDICTS]


# ═══════════════════════════════════════════════════════════════════════
# Governance Bridge (SPEC-050 FR-002)
# ═══════════════════════════════════════════════════════════════════════


@traces(spec="050", story="US-001", reqs=["REQ-050-FR2-001"])
class GovernanceBridge:
    """PAM++ policy enforcement for CheetahClaws file operations.

    SPEC-050 FR-002: Enforce governance policies before any file operation.
    """

    BLOCKED_PATTERNS = frozenset(
        {
            ".env",
            ".secrets",
            "credentials",
            "private_key",
            "id_rsa",
            "node_modules/",
            ".git/",
            "__pycache__/",
        }
    )

    @traces(spec="050", story="US-001", reqs=["REQ-050-FR2-001"])
    async def can_edit(self, file_path: str) -> tuple[bool, str]:
        """Check if a file can be edited under current governance policies.

        Returns (allowed: bool, reason: str).
        """
        path = Path(file_path)
        if not path.exists():
            return False, f"File does not exist: {file_path}"
        if path.is_dir():
            return False, f"Cannot edit a directory: {file_path}"
        if path.stat().st_size > 10 * 1024 * 1024:  # 10MB
            return False, f"File too large: {path.stat().st_size} bytes (max 10MB)"
        if self._is_binary(file_path):
            return False, f"Binary file detected: {file_path}"
        if self._contains_secrets(file_path):
            return False, f"File contains potential secrets: {file_path}"
        for blocked in self.BLOCKED_PATTERNS:
            if blocked in file_path:
                return False, f"Path blocked by governance: {blocked}"
        return True, ""

    @traces(spec="050", story="US-001", reqs=["REQ-050-FR2-001"])
    def _is_binary(self, file_path: str) -> bool:
        """Check if file contains null bytes (binary indicator)."""
        try:
            with open(file_path, "rb") as f:
                chunk = f.read(8192)
            return b"\x00" in chunk
        except Exception:
            return True

    @traces(spec="050", story="US-001", reqs=["REQ-050-FR2-001"])
    def _contains_secrets(self, file_path: str) -> bool:
        """Check filename against known secret patterns."""
        name = Path(file_path).name.lower()
        secret_markers = {".env", "secret", "credential", "private", "token", "key"}
        return any(m in name for m in secret_markers)


# ═══════════════════════════════════════════════════════════════════════
# UltimateTUI — Command Handler (SPEC-050 FR-009)
# ═══════════════════════════════════════════════════════════════════════


@traces(
    spec="050",
    story="US-001",
    reqs=["REQ-050-FR1-001", "REQ-050-FR2-001", "REQ-050-FR9-001"],
)
class UltimateTUI:
    """CheetahClaws TUI adapted for Orchestrata multi-agent coordination.

    Wires 40+ CC slash commands to Orchestrata kernel + agents through CXP.

    SPEC-050 FR-009: Route all CheetahClaws commands through Orchestrata
    kernel with appropriate agent dispatch.
    """

    def __init__(self, workspace: Path | None = None):
        self.workspace = workspace or Path.cwd()
        self.coordinator = MultiAgentCoordinator()
        self.governance = GovernanceBridge()
        self.session_id: str | None = None

    # ── Edit Command (FR-001) ──────────────────────────────────────

    @traces(spec="050", story="US-001", reqs=["REQ-050-FR1-001"])
    async def handle_edit(
        self,
        file_path: str,
        old_string: str,
        new_string: str,
        strict: bool = True,
    ) -> EditResult:
        """Handle /edit command with multi-agent coordination.

        SPEC-050 FR-001: Dispatch to Cd, Qa, Ar, Ux in parallel.
        All must approve — FAIL-CLOSED: a BLOCK, an unhandled/TIMEOUT
        review, or an ERROR all withhold approval and reject the edit
        (see ``MultiAgentCoordinator.rejecting_feedback``).

        Args:
            file_path: Path to file to edit.
            old_string: Text to replace.
            new_string: Replacement text.
            strict: Whether old_string must match exactly.

        Returns:
            EditResult with success status, diff, and agent feedback.
        """
        edit = EditRequest(file_path, old_string, new_string, strict)

        # 1. Governance check
        allowed, reason = await self.governance.can_edit(file_path)
        if not allowed:
            return EditResult(success=False, file_path=file_path, error=reason)

        # 2. Determine which agents to involve
        is_frontend = self._is_frontend_file(file_path)
        agents = (
            self.coordinator.EDIT_AGENTS_WITH_UX
            if is_frontend
            else self.coordinator.EDIT_AGENTS
        )

        # 3. Dispatch to agents in parallel
        feedback = await self.coordinator.dispatch_review(edit, agents=agents)

        # 4. Check for blockers — FAIL-CLOSED (SPEC-050 FR-001): approval
        # must be explicit. BLOCK, TIMEOUT (unhandled review), and ERROR
        # all withhold approval; only APPROVE/SUGGEST let the edit proceed.
        blockers = self.coordinator.rejecting_feedback(feedback)
        if blockers:
            return EditResult(
                success=False,
                file_path=file_path,
                feedback=feedback,
                error=f"Edit blocked by {len(blockers)} agent(s): "
                f"{', '.join(f'{b.agent_id} ({b.verdict.value})' for b in blockers)}",
            )

        # 5. Apply the edit
        try:
            original = Path(file_path).read_text(encoding="utf-8")
            if strict and old_string not in original:
                return EditResult(
                    success=False,
                    file_path=file_path,
                    error="old_string not found in file (strict mode)",
                )
            new_content = original.replace(old_string, new_string, 1)
            Path(file_path).write_text(new_content, encoding="utf-8")
            bytes_changed = len(new_content) - len(original)
        except Exception as e:
            return EditResult(success=False, file_path=file_path, error=str(e))

        return EditResult(
            success=True,
            file_path=file_path,
            diff=f"-{old_string}\n+{new_string}",
            bytes_changed=bytes_changed,
            agents_responded=[f.agent_id for f in feedback],
            feedback=feedback,
        )

    # ── Intent Command (Concierge Pipeline) ────────────────────────

    @traces(spec="050", story="US-001", reqs=["REQ-050-FR9-001"])
    async def handle_intent(self, text: str) -> dict[str, Any]:
        """Handle /intent command — run full agent pipeline.

        Routes through Concierge → BA → PM → Ar → Cd → QA → Cp.

        Args:
            text: User's natural language intent.

        Returns:
            Dict with thread_id, pipeline_result, and artifacts.
        """
        from orchestrata_ultimate.orchestrator.pipeline_runner import PipelineRunner

        runner = PipelineRunner()
        result = await runner.run(intent=text, work_type="feature")

        return {
            "thread_id": result.thread_id,
            "status": result.overall_status.value,
            "stages": list(result.stages.keys()),
            "artifacts_count": len(result.artifacts),
        }

    # ── Command Routing ────────────────────────────────────────────

    @traces(spec="050", story="US-001", reqs=["REQ-050-FR9-001"])
    async def route_command(self, command: str, args: list[str]) -> Any:
        """Route a CC slash command to the appropriate handler.

        Maps 40+ CheetahClaws commands to Orchestrata actions.
        """
        handlers = {
            "/edit": lambda: self.handle_edit(
                args[0], args[1], args[3] if len(args) > 3 else ""
            ),
            "/intent": lambda: self.handle_intent(" ".join(args)),
            "/read": lambda: Path(args[0]).read_text(encoding="utf-8") if args else "",
            "/write": lambda: self._handle_write(args[0], " ".join(args[1:])),
            "/agents": lambda: self._handle_agents(),
            "/model": lambda: self._handle_model(args[0] if args else ""),
        }

        handler = handlers.get(command)
        if handler is None:
            logger.warning("Unknown CC command: %s", command)
            return f"Unknown command: {command}"
        result = handler()
        return await result if asyncio.iscoroutine(result) else result

    # ── Helpers ─────────────────────────────────────────────────────

    @traces(spec="050", story="US-001", reqs=["REQ-050-FR1-001"])
    def _is_frontend_file(self, path: str) -> bool:
        frontend_exts = {
            ".js",
            ".jsx",
            ".ts",
            ".tsx",
            ".css",
            ".html",
            ".vue",
            ".svelte",
        }
        return Path(path).suffix in frontend_exts

    @traces(spec="050", story="US-001", reqs=["REQ-050-FR9-001"])
    async def _handle_write(self, path: str, content: str) -> str:
        Path(path).write_text(content, encoding="utf-8")
        return f"Wrote {len(content)} bytes to {path}"

    @traces(spec="050", story="US-001", reqs=["REQ-050-FR9-001"])
    async def _handle_agents(self) -> str:
        from orchestrata_ultimate.llm.config import _get_config

        cfg = _get_config()
        lines = [f"Profile: {cfg.profile_name}"]
        for alias in sorted(cfg.known_agents):
            lines.append(
                f"  {alias}: {cfg.get_provider(alias)} / {cfg.get_model(alias)}"
            )
        return "\n".join(lines)

    @traces(spec="050", story="US-001", reqs=["REQ-050-FR9-001"])
    async def _handle_model(self, profile: str) -> str:
        from orchestrata_ultimate.llm.config import (
            get_active_profile,
            list_profiles,
            select_profile,
        )

        if profile in list_profiles():
            select_profile(profile)
            return f"Switched to {profile} profile. Active: {get_active_profile()}"
        return f"Unknown profile: {profile}. Available: {list_profiles()}"