`8643315`

Move TUI onto the runtime shell-owner contract

Authored by

espadonne 1 month ago

SHA: 864331578d81573e5bb4e462373138b239c2e5ac
Parents: e1c5bfc
Tree: 1a155fb

5 changed files

Status	File	+	-
M	`src/loader/cli/main.py`	5	6
M	`src/loader/runtime/runtime_handle.py`	11	0
M	`src/loader/ui/adapter.py`	1	1
M	`src/loader/ui/app.py`	61	23
M	`tests/test_runtime_handle.py`	19	0

src/loader/cli/main.pymodified

      """Build the CLI runtime owner for the requested integration path.
      Non-TUI CLI flows use the runtime-first internal handle so internal
 -    integrations stop depending on `Agent` by reflex. The Textual app still
 -    receives the public `Agent` facade intentionally because it relies on the
 -    documented public steering shell.
 +    integrations stop depending on `Agent` by reflex. Public-shell construction
 +    remains available for explicit compatibility paths, but the CLI can choose a
 +    runtime-first owner even for interactive surfaces.
      """
      if require_public_agent:
          workflow_mode_override="clarify" if clarify else ("plan" if plan else None),
          reasoning=reasoning_config,
+     )
 -    require_public_agent = not no_tui and prompt is None
      try:
          shell_owner = _build_cli_shell_owner(
              backend=llm,
              registry=registry,
              config=config,
 -            require_public_agent=require_public_agent,
 +            require_public_agent=False,
+         )
      except ValueError as exc:
          console.print(f"[red]Permission policy error:[/red] {exc}")
          from ..ui.app import LoaderApp
          app = LoaderApp(
 -            agent=shell_owner,
 +            shell_owner=shell_owner,
              model_name=model,
              mode=mode_str,
              capability_profile=(

src/loader/runtime/runtime_handle.pymodified

          return self.permission_policy.rule_counts()
 +    @property
 +    def is_running(self) -> bool:
 +        """Return whether the runtime-owned shell is currently running."""
++
 +        return self.steering.is_running
++
      @property
      def use_react(self) -> bool:
          """Determine whether to use ReAct prompting or native tools."""
          return resolve_runtime_shell_use_react(self)
 +    def steer(self, message: str) -> bool:
 +        """Queue one steering message when the runtime shell is active."""
++
 +        return self.steering.steer(message)
++
      def resume_session(self, session_id: str | None = None) -> bool:
          """Resume the latest or named persisted session."""

src/loader/ui/adapter.pymodified

  from textual.message import Message
 -from ..agent.loop import AgentEvent
 +from ..runtime.events import AgentEvent
  if TYPE_CHECKING:
      from ..runtime.reasoning_types import (

src/loader/ui/app.pymodified

  import asyncio
  import time
 +from collections.abc import Awaitable, Callable
  from pathlib import Path
 +from typing import Protocol
  from rich.markup import escape
  from textual import work
  from textual.widgets import Footer, Input, Static
  from textual.worker import Worker, get_current_worker
 -from ..agent.loop import Agent, AgentEvent
 +from ..runtime.events import AgentEvent
  from .adapter import (
      ArtifactCreated,
      ClearStream,
+ )
 +class LoaderUIShellOwner(Protocol):
 +    """Small shell-owner contract used by the Textual UI."""
++
 +    backend: object
 +    capability_profile: object
 +    safeguards: object
 +    is_running: bool
++
 +    def steer(self, message: str) -> bool:
 +        """Queue one steering message while the owner is running."""
++
 +    def refresh_capability_profile(self) -> None:
 +        """Refresh the active capability profile."""
++
 +    async def run(
 +        self,
 +        user_message: str,
 +        on_event: (
 +            Callable[[AgentEvent], None]
 +            | Callable[[AgentEvent], Awaitable[None]]
 +            | None
 +        ) = None,
 +        on_confirmation: Callable[[str, str, str], Awaitable[bool]] | None = None,
 +        on_user_question: Callable[[str, list[str] | None], Awaitable[str]] | None = None,
 +        use_plan: bool | None = None,
 +    ) -> str:
 +        """Run one user message through the shell owner."""
++
 +    def clear_history(self) -> None:
 +        """Reset the owner history."""
++
++
  class LoaderApp(App):
      """Main Textual application for Loader."""
      def __init__(
          self,
 -        agent: Agent,
 +        shell_owner: LoaderUIShellOwner,
          model_name: str = "",
          mode: str = "Native",
          capability_profile: str = "",
          **kwargs,
      ) -> None:
          super().__init__(**kwargs)
 -        self.agent = agent
 +        self.shell_owner = shell_owner
          self.model_name = model_name
          self.mode = mode
          self.capability_profile = capability_profile
              self.action_clear_messages()
              return
 -        # If agent is running, this is a steering message
 -        if self.is_generating and self.agent.is_running:
 +        # If the runtime owner is running, this is a steering message
 +        if self.is_generating and self.shell_owner.is_running:
              # Finalize current streaming so new content appears below user's message
              if self._current_streaming is not None:
                  self._current_streaming.stop_streaming()
                  self._current_streaming = None
              self._add_steering_message(user_input)
 -            self.agent.steer(user_input)
 +            self.shell_owner.steer(user_input)
              return
          # Add user message to display
          try:
              models = []
 -            if hasattr(self.agent.backend, "list_models"):
 -                models = await self.agent.backend.list_models()
 +            if hasattr(self.shell_owner.backend, "list_models"):
 +                models = await self.shell_owner.backend.list_models()
              if models:
 -                current = self.agent.backend.model if hasattr(self.agent.backend, "model") else ""
 +                current = (
 +                    self.shell_owner.backend.model
 +                    if hasattr(self.shell_owner.backend, "model")
 +                    else ""
 +                )
                  def on_select(selected: str | None) -> None:
                      if selected:
      def _switch_model(self, model_name: str) -> None:
          """Switch to a different model."""
 -        if hasattr(self.agent.backend, "model"):
 -            old_model = self.agent.backend.model
 -            self.agent.backend.model = model_name
 -            if hasattr(self.agent, "refresh_capability_profile"):
 -                self.agent.refresh_capability_profile()
 +        if hasattr(self.shell_owner.backend, "model"):
 +            old_model = self.shell_owner.backend.model
 +            self.shell_owner.backend.model = model_name
 +            if hasattr(self.shell_owner, "refresh_capability_profile"):
 +                self.shell_owner.refresh_capability_profile()
              self.model_name = model_name
              # Update status line
              status = self.query_one(StatusLine)
              status.model = model_name
              # Update mode based on new model's capabilities
 -            if hasattr(self.agent.backend, "supports_native_tools"):
 -                supports_native = self.agent.backend.supports_native_tools()
 +            if hasattr(self.shell_owner.backend, "supports_native_tools"):
 +                supports_native = self.shell_owner.backend.supports_native_tools()
                  self.mode = "Native" if supports_native else "ReAct"
                  status.mode = self.mode
 -            if hasattr(self.agent, "capability_profile"):
 +            if hasattr(self.shell_owner, "capability_profile"):
                  self.capability_profile = (
 -                    f"{self.agent.capability_profile.preferred_tool_call_format}/"
 -                    f"{self.agent.capability_profile.verification_strictness}"
 +                    f"{self.shell_owner.capability_profile.preferred_tool_call_format}/"
 +                    f"{self.shell_owner.capability_profile.verification_strictness}"
+                 )
                  status.capability_profile = self.capability_profile
              self._add_message(
                  await asyncio.sleep(0)
          async def on_confirmation(tool_name: str, message: str, details: str) -> bool:
 -            """Handle confirmation requests from agent."""
 +            """Handle confirmation requests from the runtime owner."""
              if worker.is_cancelled:
                  return False
              return await self._request_confirmation(tool_name, message, details)
              return await self._request_user_question(question, options)
          try:
 -            return await self.agent.run(
 +            return await self.shell_owner.run(
                  user_input,
                  on_event=on_event,
                  on_confirmation=on_confirmation,
          # Filter content through safeguards before displaying
          # This removes bracket tool calls, code blocks, etc. from stream
 -        filtered_content = self.agent.safeguards.filter_stream_chunk(message.content)
 +        filtered_content = self.shell_owner.safeguards.filter_stream_chunk(message.content)
          if filtered_content:
              self._current_streaming.append(filtered_content)
          """Clear all messages."""
          msg_area = self.query_one("#message-area", ScrollableContainer)
          msg_area.remove_children()
 -        self.agent.clear_history()
 +        self.shell_owner.clear_history()
          self.query_one(StatusLine).clear_definition_of_done()
          self.query_one(StatusLine).update_workflow_mode("execute")
          self._add_message("[dim]Conversation cleared.[/dim]")

tests/test_runtime_handle.pymodified

      assert run.response == "Runtime harness reply."
      assert isinstance(explore_run.agent, RuntimeHandle)
      assert explore_run.response == "Explore harness reply."
++
++
 +def test_runtime_handle_exposes_public_shell_steering_contract(
 +    temp_dir: Path,
 +) -> None:
 +    handle = RuntimeHandle(
 +        backend=ScriptedBackend(),
 +        config=AgentConfig(auto_context=False),
 +        project_root=temp_dir,
 +    )
++
 +    assert handle.is_running is False
 +    assert handle.steer("stay in runtime") is False
++
 +    handle.steering.mark_running()
++
 +    assert handle.is_running is True
 +    assert handle.steer("stay in runtime") is True
 +    assert handle.drain_steering_messages() == ["stay in runtime"]