tenseleyflow/loader / 53d9e59

 """The main agent loop."""

 import asyncio

 from dataclasses import dataclass

 from pathlib import Path

 from ..context.project import ProjectContext, detect_project

 from ..runtime.capabilities import resolve_backend_capability_profile

 from ..runtime.conversation import ConversationRuntime

 from ..runtime.events import AgentEvent, TurnSummary

 from ..runtime.session import ConversationSession

 from .prompts import build_system_prompt

 from .reasoning import (

     DECOMPOSITION_PROMPT,

     SELF_CRITIQUE_PROMPT,

     VERIFICATION_PROMPT,

     parse_decomposition,

     parse_self_critique,

     parse_verification,

     quick_verify,

 )

 from .safeguards import RuntimeSafeguards

     def refresh_capability_profile(self) -> None:

         """Refresh the runtime capability profile from the current backend."""

         self._use_react = None

     def _get_few_shot_examples(self) -> list[Message]:

                     # Run the step

                     step_prompt = format_step_prompt(plan, step)

                         step_prompt, emit, on_confirmation,

                         original_task=self._current_task,

                     )

         self,

         user_message: str,

     ) -> AsyncIterator[AgentEvent]:

                     break

     def _contains_unexecuted_code(self, content: str) -> bool:

         """Detect if response contains code blocks that should be tool calls.

         instead of using the proper tool calling API. This method tries to

         parse and recover them.

         """

         import json

         import os

         tool_calls = []

         tool_names = ["write", "read", "edit", "bash", "glob", "grep"]

 import asyncio

 import re

 import click

 import httpx

 from rich.console import Console

     Returns:

         Selected model name, or None if cancelled/no models.

     """

     from prompt_toolkit import PromptSession

     from prompt_toolkit.completion import WordCompleter

     # Create a temporary client to list models

     backend = OllamaBackend(model="")

     models = await backend.list_models()

     reason: bool,

     prompt: str | None,

 ) -> None:

     from ..agent.loop import Agent, AgentConfig, ReasoningConfig

     from ..tools.base import create_default_registry

     # Handle model selection

     if select_model:

         timeout=timeout,

     )

     # Check health

     if not await llm.health_check():

         console.print("[red]Error: Cannot connect to Ollama. Is it running?[/red]")

         console.print("\nTry [cyan]loader --select-model[/cyan] to choose from available models.")

         return

     # Save this model as the new default

     set_last_model(model)

 async def run_once(agent, prompt: str, skip_confirmation: bool = False) -> None:

     """Run a single prompt."""

     import time

     thinking_start = None

     streamed_response = False

 async def run_interactive(agent, skip_confirmation: bool = False) -> None:

     """Run interactive chat loop."""

     from prompt_toolkit import PromptSession

     from prompt_toolkit.history import FileHistory

     from ..tools.base import ConfirmationRequired

     history_file = os.path.expanduser("~/.loader_history")

     session = PromptSession(history=FileHistory(history_file))

 """Ollama backend implementation."""

 import json

 import httpx

     CompletionResponse,

     LLMBackend,

     Message,

     StreamChunk,

     ToolCall,

 )

         self._client = httpx.AsyncClient(timeout=timeout)

         self._supports_native_tools: bool | None = None

         self._model_details_cache: dict[str, Any] | None = None

         self._capability_profile: CapabilityProfile | None = None

     def _build_options(self, temperature: float, max_tokens: int) -> dict:

         """Build Ollama options dict with performance settings."""

         return {

     async def describe_model(self) -> dict[str, Any] | None:

         """Fetch and cache Ollama model details for capability resolution."""

             return self._model_details_cache

         if not self.model:

             )

             response.raise_for_status()

             self._model_details_cache = response.json()

         except Exception:

             self._model_details_cache = None

         return self._model_details_cache

     def capability_profile(self) -> CapabilityProfile:

         """Return the resolved capability profile for the current model."""

             self._capability_profile = resolve_capability_profile(

                 self.model,

                 model_details=self._model_details_cache,

         if self.force_react:

             return False

         if self._supports_native_tools is not None:

             return self._supports_native_tools

         max_tokens: int = 4096,

     ) -> CompletionResponse:

         """Generate a completion using Ollama."""

         payload: dict[str, Any] = {

             "model": self.model,

             "messages": self._format_messages(messages),

         max_tokens: int = 4096,

     ) -> AsyncIterator[StreamChunk]:

         """Stream a completion from Ollama."""

         payload: dict[str, Any] = {

             "model": self.model,

             "messages": self._format_messages(messages),

     async def _stream_response(self, response) -> AsyncIterator[StreamChunk]:

         """Internal helper to stream response chunks."""

         full_content = ""

         display_content = ""  # Content to show (filtered)

     ) -> TurnSummary:

         """Run one task turn and return a structured summary."""

         iterations = 0

         final_response = ""

         actions_taken: list[str] = []

         for key, value in update.items():

             target[key] = target.get(key, 0) + value

     @staticmethod

     def _emit_confirmation(emit: EventSink):

         async def _emit(tool_name: str, message: str, details: str) -> None: