wehub-resource-sync
114 lines
3.9 KiB
Python
114 lines
3.9 KiB
Python
"""Executor that bridges Omnigent web-chat turns into the native Kiro TUI."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
from collections.abc import AsyncIterator
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from omnigent.inner.executor import (
|
|
Executor,
|
|
ExecutorConfig,
|
|
ExecutorError,
|
|
ExecutorEvent,
|
|
Message,
|
|
ToolSpec,
|
|
TurnComplete,
|
|
)
|
|
from omnigent.kiro_native_bridge import KIRO_NATIVE_BRIDGE_DIR_ENV_VAR, inject_user_message
|
|
|
|
|
|
class KiroNativeExecutor(Executor):
|
|
"""Harness-side executor for ``omnigent kiro`` web-UI turns."""
|
|
|
|
def __init__(self, bridge_dir: Path | None = None) -> None:
|
|
self._bridge_dir = bridge_dir or _bridge_dir_from_env()
|
|
self._inject_lock = asyncio.Lock()
|
|
|
|
def supports_streaming(self) -> bool:
|
|
""":returns: ``False`` — output is shown by the embedded terminal."""
|
|
return False
|
|
|
|
def supports_live_message_queue(self) -> bool:
|
|
""":returns: ``True`` — messages can be injected mid-turn."""
|
|
return True
|
|
|
|
async def enqueue_session_message(self, session_key: str, content: Any) -> bool:
|
|
"""Inject a live steering message into the Kiro terminal."""
|
|
del session_key
|
|
text = _content_to_text(content, self._bridge_dir)
|
|
if not text:
|
|
return False
|
|
try:
|
|
async with self._inject_lock:
|
|
await asyncio.to_thread(inject_user_message, self._bridge_dir, content=text)
|
|
except RuntimeError:
|
|
return False
|
|
return True
|
|
|
|
async def run_turn(
|
|
self,
|
|
messages: list[Message],
|
|
tools: list[ToolSpec],
|
|
system_prompt: str,
|
|
config: ExecutorConfig | None = None,
|
|
) -> AsyncIterator[ExecutorEvent]:
|
|
"""Inject the latest web-UI user message into the Kiro TUI pane."""
|
|
del tools, system_prompt, config
|
|
text = _latest_user_text(messages, self._bridge_dir)
|
|
if not text:
|
|
yield ExecutorError(message="kiro native turn had no user text to send")
|
|
return
|
|
try:
|
|
async with self._inject_lock:
|
|
await asyncio.to_thread(inject_user_message, self._bridge_dir, content=text)
|
|
except RuntimeError as exc:
|
|
yield ExecutorError(message=str(exc))
|
|
return
|
|
yield TurnComplete(response=None)
|
|
|
|
|
|
def _bridge_dir_from_env() -> Path:
|
|
"""Resolve the kiro-native bridge dir from the harness spawn env."""
|
|
raw = os.environ.get(KIRO_NATIVE_BRIDGE_DIR_ENV_VAR, "").strip()
|
|
if not raw:
|
|
raise RuntimeError(
|
|
f"{KIRO_NATIVE_BRIDGE_DIR_ENV_VAR} is required for the kiro-native harness"
|
|
)
|
|
return Path(raw)
|
|
|
|
|
|
def _latest_user_text(messages: list[Message], bridge_dir: Path) -> str:
|
|
"""Return the latest user message's text."""
|
|
for message in reversed(messages):
|
|
if message.get("role") == "user":
|
|
return _content_to_text(message.get("content"), bridge_dir)
|
|
return ""
|
|
|
|
|
|
def _content_to_text(content: Any, bridge_dir: Path) -> str:
|
|
"""Normalize executor content into text the Kiro TUI receives."""
|
|
if isinstance(content, str):
|
|
return content
|
|
if isinstance(content, list):
|
|
from omnigent.inner.native_attachments import materialize_attachment
|
|
|
|
attachment_lines: list[str] = []
|
|
text_parts: list[str] = []
|
|
for block in content:
|
|
if not isinstance(block, dict):
|
|
continue
|
|
block_type = block.get("type", "")
|
|
if block_type in ("input_text", "text"):
|
|
text = block.get("text")
|
|
if isinstance(text, str):
|
|
text_parts.append(text)
|
|
elif block_type in ("input_image", "input_file"):
|
|
path = materialize_attachment(block, bridge_dir)
|
|
if path is not None:
|
|
attachment_lines.append(f"[Attached: {path}]")
|
|
return "\n\n".join(attachment_lines + text_parts)
|
|
return ""
|