`c000559`

Add runtime-handle execution entrypoints

Authored by

espadonne 1 month ago

SHA: c00055905796ba6c78ac75cd1c229cbfe681ce6d
Parents: 28a4d01
Tree: fb4fbb1

2 changed files

Status	File	+	-
M	`src/loader/runtime/runtime_handle.py`	58	2
M	`tests/test_runtime_handle.py`	65	1

src/loader/runtime/runtime_handle.pymodified

  from __future__ import annotations
 +from collections.abc import AsyncIterator, Awaitable, Callable
  from pathlib import Path
  from loader.runtime.safeguards import RuntimeSafeguards
  from ..llm.base import LLMBackend, Message
  from ..tools.base import ToolRegistry, create_default_registry
  from .capabilities import resolve_backend_capability_profile
 -from .events import TurnSummary
 +from .events import AgentEvent, TurnSummary
  from .permissions import (
      PermissionMode,
      build_permission_policy,
      refresh_runtime_shell_capability_profile,
      resolve_runtime_shell_use_react,
      resume_runtime_shell_session,
 +    run_runtime_shell,
 +    run_runtime_shell_explore,
      set_runtime_shell_workflow_mode,
 +    stream_runtime_shell,
+ )
  from .workflow import WorkflowMode
  class RuntimeHandle:
 -    """Runtime-first internal owner for launcher and turn-runtime tests."""
 +    """Runtime-first internal owner for launcher, shell, and integration paths."""
      def __init__(
          self,
          clear_runtime_shell_history(self)
 +    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 runtime-owned shell entrypoint."""
++
 +        return await run_runtime_shell(
 +            self,
 +            user_message,
 +            on_event=on_event,
 +            on_confirmation=on_confirmation,
 +            on_user_question=on_user_question,
 +            use_plan=use_plan,
 +        )
++
 +    async def run_streaming(
 +        self,
 +        user_message: str,
 +    ) -> AsyncIterator[AgentEvent]:
 +        """Yield the streamed event sequence from the runtime-owned shell."""
++
 +        async for event in stream_runtime_shell(self, user_message):
 +            yield event
++
 +    async def run_explore(
 +        self,
 +        user_message: str,
 +        on_event: (
 +            Callable[[AgentEvent], None]
 +            | Callable[[AgentEvent], Awaitable[None]]
 +            | None
 +        ) = None,
 +        *,
 +        fresh: bool = False,
 +    ) -> str:
 +        """Run one read-only explore query through the runtime-owned shell."""
++
 +        return await run_runtime_shell_explore(
 +            self,
 +            user_message,
 +            on_event=on_event,
 +            fresh=fresh,
 +        )
++
      def set_workflow_mode(self, workflow_mode: str) -> None:
          """Update the active workflow mode used by the system prompt."""

tests/test_runtime_handle.pymodified

  import pytest
  from loader.agent.loop import AgentConfig
 -from loader.llm.base import CompletionResponse
 +from loader.llm.base import CompletionResponse, StreamChunk
  from loader.runtime.bootstrap import RuntimeBootstrapView, build_runtime_context
  from loader.runtime.conversation import ConversationRuntime
  from loader.runtime.launcher import RuntimeLauncher, build_runtime_launcher
      assert summary.final_response == "Runtime handle reply."
      assert runtime.source.metadata == {"owner_type": "RuntimeHandle"}
      assert any(event.type == "response" for event in events)
++
++
 +@pytest.mark.asyncio
 +async def test_runtime_handle_runs_public_shell_entrypoint_without_agent(
 +    temp_dir: Path,
 +) -> None:
 +    handle = RuntimeHandle(
 +        backend=ScriptedBackend(
 +            completions=[CompletionResponse(content="Runtime handle shell reply.")]
 +        ),
 +        config=AgentConfig(auto_context=False, stream=False),
 +        project_root=temp_dir,
 +    )
 +    events = []
++
 +    async def emit(event) -> None:
 +        events.append(event)
++
 +    response = await handle.run(
 +        "Summarize the runtime-first shell path.",
 +        on_event=emit,
 +        use_plan=False,
 +    )
++
 +    assert response == "Runtime handle shell reply."
 +    assert handle.last_turn_summary is not None
 +    assert handle.last_turn_summary.final_response == "Runtime handle shell reply."
 +    assert any(event.type == "response" for event in events)
++
++
 +@pytest.mark.asyncio
 +async def test_runtime_handle_runs_explore_and_streaming_entrypoints_without_agent(
 +    temp_dir: Path,
 +) -> None:
 +    handle = RuntimeHandle(
 +        backend=ScriptedBackend(
 +            completions=[CompletionResponse(content="Explore with runtime handle.")],
 +            streams=[
 +                [
 +                    StreamChunk(content="Streamed ", is_done=False),
 +                    StreamChunk(
 +                        content="runtime reply.",
 +                        full_content="Streamed runtime reply.",
 +                        is_done=True,
 +                    ),
 +                ]
 +            ],
 +        ),
 +        config=AgentConfig(auto_context=False),
 +        project_root=temp_dir,
 +    )
++
 +    stream_events = [event async for event in handle.run_streaming("thanks")]
 +    explore_response = await handle.run_explore(
 +        "Where should I start in this repo?",
 +    )
++
 +    assert any(
 +        event.type == "response" and event.content == "Streamed runtime reply."
 +        for event in stream_events
 +    )
 +    assert explore_response == "Explore with runtime handle."
 +    assert handle.last_turn_summary is not None
 +    assert handle.last_turn_summary.workflow_mode == "explore"