orchestrata_ultimate.gateway.handlers.ultimatetui
Reference › orchestrata_ultimate › gateway › handlers › ultimatetui
UltimateTUIHandler bridges the VizInt UI and the terminal UI to the CheetahClaws agent integration (integrations.cheetahclaws) and the model settings in llm.config. It serves the tui.edit, tui.intent, tui.agents, tui.models, tui.model_switch and tui.read RPCs. rpc is its importer, and it raises errors from exceptions. See gateway.
The constructor takes a workspace path. handle_edit needs file_path and old_string (code -32602 if missing), runs a multi-agent file edit with new_string and a strict flag, and returns whether it succeeded, the diff, the bytes changed, which agents responded and each agent's verdict, truncated content, tools used and errors. handle_intent needs text and runs the full agent pipeline, returning a thread id, status, stages and artifact count. handle_agents and handle_models list agents and models, handle_model_switch changes the active model, and handle_read reads a file through the integration.
At a glance
| Source | src/orchestrata_ultimate/gateway/handlers/ultimatetui.py |
| Lines of code | 240 |
| Classes | 1 |
| Functions | 0 |
| Imports (internal) | orchestrata_ultimate.core.base.traces, orchestrata_ultimate.gateway.exceptions, orchestrata_ultimate.integrations.cheetahclaws, orchestrata_ultimate.llm.config |
| Imported by | orchestrata_ultimate.gateway.rpc |
| Tests that import it | tests/gateway/handlers/test_ultimatetui.py, tests/generated_gateway/test_handlers_ultimatetui_generated.py |
Class diagram
Dependency diagram
API
Classes
UltimateTUIHandler
Decorators: traces(spec='050', story='US-001', reqs=['REQ-050-FR9-001']) · Line 42
Gateway handler that bridges VizInt UI to CheetahClaws agent dispatch.
| Method | Signature | Summary |
|---|---|---|
__init__ |
__init__(workspace: str | None = None) |
|
handle_edit |
async handle_edit(connection_id: str, params: dict[str, Any]) -> dict[str, Any] |
Handle tui.edit — multi-agent file edit through CXP. |
handle_intent |
async handle_intent(connection_id: str, params: dict[str, Any]) -> dict[str, Any] |
Handle tui.intent — run full agent pipeline. |
handle_agents |
async handle_agents(connection_id: str, params: dict[str, Any]) -> dict[str, Any] |
Handle tui.agents — get agent roster with provider/model assignments. |
handle_models |
async handle_models(connection_id: str, params: dict[str, Any]) -> dict[str, Any] |
Handle tui.models — list available LLM profiles. |
handle_model_switch |
async handle_model_switch(connection_id: str, params: dict[str, Any]) -> dict[str, Any] |
Handle tui.model_switch — switch to a different LLM profile. |
handle_read |
async handle_read(connection_id: str, params: dict[str, Any]) -> dict[str, Any] |
Handle tui.read — read file contents. |
Source
"""UltimateTUI gateway handler — bridges VizInt UI to CheetahClaws agents.
Exposes CC slash commands as JSON-RPC 2.0 endpoints so the VizInt browser
can call `UltimateTUI.handle_edit()`, `handle_intent()`, etc. through the
WebSocket gateway.
RPC Methods:
tui.edit { file_path, old_string, new_string, strict? } → EditResult
tui.intent { text } → PipelineResult
tui.agents {} → agent roster
tui.models {} → available profiles
tui.model_switch { profile } → confirmation
tui.read { file_path } → file contents
Architecture:
Browser → WebSocket → Gateway → UltimateTUIHandler
│
UltimateTUI
│
┌──────────┼──────────┐
▼ ▼ ▼
Governance MultiAgent PipelineRunner
Bridge Coordinator
"""
from __future__ import annotations
import logging
from pathlib import Path
from typing import Any
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.gateway.exceptions import RPCError
from orchestrata_ultimate.integrations.cheetahclaws import UltimateTUI
logger = logging.getLogger(__name__)
RPC_PREFIX = "tui"
@traces(spec="050", story="US-001", reqs=["REQ-050-FR9-001"])
class UltimateTUIHandler:
"""Gateway handler that bridges VizInt UI to CheetahClaws agent dispatch.
Every method corresponds to a ``tui.*`` RPC call the browser can make.
"""
def __init__(self, workspace: str | None = None):
self._tui = UltimateTUI(workspace=Path(workspace) if workspace else None)
# ── tui.edit ────────────────────────────────────────────────────
@traces(spec="050", story="US-001", reqs=["REQ-050-FR1-001"])
async def handle_edit(
self, connection_id: str, params: dict[str, Any]
) -> dict[str, Any]:
"""Handle tui.edit — multi-agent file edit through CXP.
Params:
file_path: Path to file.
old_string: Text to replace.
new_string: Replacement text.
strict: Whether old_string must match exactly (default: true).
Returns:
EditResult as dict.
"""
file_path = params.get("file_path", "")
old_string = params.get("old_string", "")
new_string = params.get("new_string", "")
strict = params.get("strict", True)
if not all([file_path, old_string is not None]):
raise RPCError(
"Missing required params: file_path, old_string", rpc_code=-32602
)
logger.info(
"tui.edit: %s (%d chars → %d chars)",
file_path,
len(old_string),
len(new_string),
)
result = await self._tui.handle_edit(
file_path=file_path,
old_string=old_string,
new_string=new_string,
strict=strict,
)
return {
"success": result.success,
"file_path": result.file_path,
"diff": result.diff,
"bytes_changed": result.bytes_changed,
"agents_responded": result.agents_responded,
"feedback": [
{
"agent_id": f.agent_id,
"verdict": f.verdict.value,
"content": f.content[:500],
"tools_used": f.tools_used,
"errors": f.errors,
}
for f in result.feedback
],
"error": result.error,
}
# ── tui.intent ──────────────────────────────────────────────────
@traces(spec="050", story="US-001", reqs=["REQ-050-FR9-001"])
async def handle_intent(
self, connection_id: str, params: dict[str, Any]
) -> dict[str, Any]:
"""Handle tui.intent — run full agent pipeline.
Params:
text: Human intent text.
Returns:
Dict with thread_id, status, stages, artifacts_count.
"""
text = params.get("text", "")
if not text:
raise RPCError("Missing required param: text", rpc_code=-32602)
logger.info("tui.intent: %s", text[:100])
result = await self._tui.handle_intent(text)
return dict(result)
# ── tui.agents ──────────────────────────────────────────────────
async def handle_agents(
self, connection_id: str, params: dict[str, Any]
) -> dict[str, Any]:
"""Handle tui.agents — get agent roster with provider/model assignments."""
from orchestrata_ultimate.llm.config import _get_config
cfg = _get_config()
agents = {}
for alias in sorted(cfg.known_agents):
agents[alias] = {
"provider": cfg.get_provider(alias),
"model": cfg.get_model(alias),
}
return {
"profile": cfg.profile_name,
"agents": agents,
}
# ── tui.models ──────────────────────────────────────────────────
async def handle_models(
self, connection_id: str, params: dict[str, Any]
) -> dict[str, Any]:
"""Handle tui.models — list available LLM profiles."""
from orchestrata_ultimate.llm.config import _get_config, list_profiles
return {
"active": _get_config().profile_name,
"profiles": list_profiles(),
}
# ── tui.model_switch ────────────────────────────────────────────
async def handle_model_switch(
self, connection_id: str, params: dict[str, Any]
) -> dict[str, Any]:
"""Handle tui.model_switch — switch to a different LLM profile.
Params:
profile: Profile name (default, claude, kimi, qwen, budget, mix).
Returns:
Confirmation with active profile.
"""
profile = params.get("profile", "")
if not profile:
raise RPCError("Missing required param: profile", rpc_code=-32602)
from orchestrata_ultimate.llm.config import (
_get_config,
list_profiles,
select_profile,
)
if profile not in list_profiles():
raise RPCError(
f"Unknown profile: {profile}. Available: {list_profiles()}",
rpc_code=-32602,
)
select_profile(profile)
logger.info("tui.model_switch: %s", profile)
return {
"active": _get_config().profile_name,
"switched": True,
}
# ── tui.read ────────────────────────────────────────────────────
async def handle_read(
self, connection_id: str, params: dict[str, Any]
) -> dict[str, Any]:
"""Handle tui.read — read file contents.
Params:
file_path: Path to file.
Returns:
Dict with content and metadata.
"""
file_path = params.get("file_path", "")
if not file_path:
raise RPCError("Missing required param: file_path", rpc_code=-32602)
# Workspace containment (audit 2026-07-06): this RPC previously read
# ANY host path — /etc/shadow, ~/.ssh keys, the gateway's own .env —
# for any connected client. resolve() collapses ../ and symlinks
# BEFORE the containment check.
workspace = Path.cwd().resolve()
raw = Path(file_path)
p = (raw if raw.is_absolute() else workspace / raw).resolve()
if not p.is_relative_to(workspace):
raise RPCError(f"Path outside workspace: {file_path}", rpc_code=-32000)
if not p.exists():
raise RPCError(f"File not found: {file_path}", rpc_code=-32000)
content = p.read_text(encoding="utf-8")
return {
"file_path": file_path,
"content": content[:100000], # Cap at 100KB
"size": len(content),
"lines": content.count("\n") + 1,
}