refactor(session): consolidate session and agent runtime loops

flocks/channel/inbound/dispatcher.py

@@ -237,14 +237,14 @@ async def deliver_text(self, text: str) -> None:
             session_id=self.session_id,
         )
 
-    def to_loop_callbacks(self, runner_callbacks=None):
+    def to_loop_callbacks(self, *, on_text_delta=None):
         """Convert to a LoopCallbacks dataclass understood by SessionLoop."""
         from flocks.session.session_loop import LoopCallbacks
         return LoopCallbacks(
             on_step_end=self.on_step_end,
+            on_text_delta=on_text_delta,
             on_error=self.on_error,
             event_publish_callback=self._publish_sse_event,
-            runner_callbacks=runner_callbacks,
         )
 
     @staticmethod
@@ -514,8 +514,8 @@ async def _dispatch(self, msg: InboundMessage) -> None:
             # what _process_session_message does in the WebUI route. Storing
             # the resolved model on the user message keeps two things aligned
             # between WebUI and channel:
-            #   - Title generation (``SessionLoop._run_loop`` reads
-            #     ``last_user.model``).
+            #   - Title generation and turn preparation read
+            #     ``last_user.model``.
             #   - The provider-specific base prompt template
             #     (``SystemPrompt.provider``) selected on the next loop tick.
             # Without this, channel sessions ended up with the hardcoded
@@ -1223,13 +1223,12 @@ async def _run_agent_with_streaming(
             return
 
         try:
-            from flocks.session.runner import RunnerCallbacks
-
             async def _on_text_delta(delta: str) -> None:
                 await card.append(delta)
 
-            runner_cbs = RunnerCallbacks(on_text_delta=_on_text_delta)
-            loop_callbacks = callbacks.to_loop_callbacks(runner_callbacks=runner_cbs)
+            loop_callbacks = callbacks.to_loop_callbacks(
+                on_text_delta=_on_text_delta,
+            )
 
             result = await InboundDispatcher._run_session_loop(
                 binding,

flocks/cli/session_runner.py

@@ -6,7 +6,7 @@
 - Tool execution display
 - Streaming text display
 
-Core logic is in session/runner.py
+Core logic is exposed through SessionLoop.
 """
 
 import asyncio
@@ -26,33 +26,18 @@
 
 from flocks.utils.log import Log
 from flocks.session.session import Session, SessionInfo
-from flocks.session.runner import SessionRunner, RunnerCallbacks, ToolResult
+from flocks.session.session_loop import LoopCallbacks
 from flocks.session.message import Message, MessageRole
 from flocks.agent.registry import Agent
 from flocks.provider.provider import Provider
-from flocks.tool.registry import ToolRegistry
+from flocks.tool.registry import ToolRegistry, ToolResult
 from flocks.project.project import Project
 from dotenv import load_dotenv
 
 
 log = Log.create(service="cli.runner")
 
 
-# Module-level storage for CLI callbacks (used by SessionRunner during loop execution)
-_CLI_CALLBACKS: Optional['RunnerCallbacks'] = None
-
-
-def _set_cli_callbacks(callbacks: Optional['RunnerCallbacks']) -> None:
-    """Set CLI callbacks for current execution"""
-    global _CLI_CALLBACKS
-    _CLI_CALLBACKS = callbacks
-
-
-def _get_cli_callbacks() -> Optional['RunnerCallbacks']:
-    """Get CLI callbacks for current execution"""
-    return _CLI_CALLBACKS
-
-
 # Tool display styles
 TOOL_STYLES: Dict[str, tuple] = {
     "todo": ("Todo", "yellow bold"),
@@ -71,7 +56,7 @@ def _get_cli_callbacks() -> Optional['RunnerCallbacks']:
 
 class CLISessionRunner:
     """
-    CLI wrapper for SessionRunner.
+    CLI wrapper for the public SessionLoop entry point.
 
     Handles all CLI-specific display logic.
     """
@@ -90,7 +75,6 @@ def __init__(
         self.agent_name = agent
         self.auto_confirm = auto_confirm
         self._session: Optional[SessionInfo] = None
-        self._runner: Optional[SessionRunner] = None
         self._live: Optional[Live] = None
         self._content_buffer: list[str] = []
 
@@ -299,8 +283,10 @@ async def _interactive_loop(self) -> None:
 
             except KeyboardInterrupt:
                 self.console.print("\n[dim]Interrupted[/dim]")
-                if self._runner:
-                    self._runner.abort()
+                if self._session:
+                    from flocks.session.session_loop import SessionLoop
+
+                    SessionLoop.abort(self._session.id)
                 break
             except EOFError:
                 break
@@ -348,7 +334,6 @@ async def _process_message(
             from flocks.input.dispatcher import dispatch_user_input
             from flocks.input.events import UserInputEvent
             from flocks.input.output import CliOutputSink
-            from flocks.session.message import Message
 
             event = UserInputEvent(
                 source_type="cli",
@@ -408,29 +393,21 @@ async def _clear_history() -> None:
             model={"providerID": provider_id, "modelID": model_id},
         )
 
-        # Import SessionLoop and LoopCallbacks
-        from flocks.session.session_loop import SessionLoop, LoopCallbacks
-        from flocks.session.runner import RunnerCallbacks
+        # Import the stable session execution entry point.
+        from flocks.session.session_loop import SessionLoop
 
-        # Create loop callbacks (wrapping runner callbacks)
+        # Pass one explicit callback set through the full runtime.
         loop_callbacks = LoopCallbacks(
             on_step_start=self._on_step_start,
             on_step_end=self._on_step_end,
-            on_error=self._on_error,
-            on_compaction=self._on_compaction,
-        )
-
-        # Store runner callbacks for tool events
-        # We need to hook into SessionRunner to get tool callbacks
-        # This is done by temporarily storing callbacks in a module-level variable
-        _set_cli_callbacks(RunnerCallbacks(
             on_text_delta=self._on_text_delta,
             on_reasoning_delta=self._on_reasoning_delta,
             on_tool_start=self._on_tool_start,
             on_tool_end=self._on_tool_end,
             on_permission_request=self._on_permission_request,
             on_error=self._on_error,
-        ))
+            on_compaction=self._on_compaction,
+        )
 
         # Start streaming display
         self._content_buffer = []
@@ -486,9 +463,6 @@ async def _clear_history() -> None:
             live.update(Text(""))
             self._live = None
 
-        # Clear callbacks
-        _set_cli_callbacks(None)
-
         # Print any remaining content not yet printed
         if self._content_buffer:
             self._flush_content()
@@ -806,8 +780,6 @@ def _print_help(self) -> None:
 __all__ = [
     "CLISessionRunner",
     "run_session",
-    "_get_cli_callbacks",
-    "_set_cli_callbacks",
 ]
 
 

flocks/provider/options.py

@@ -4,7 +4,7 @@
 Centralises the logic for assembling thinking / reasoning / token-limit
 kwargs that get forwarded to each provider's ``chat_stream`` call.
 
-Both ``SessionRunner`` (session/runner.py) and ``AgentExecutor``
+Both ``StepEngine`` and ``AgentExecutor``
 (agent/runtime/executor.py) delegate to :func:`build_provider_options`
 so that provider rules are maintained in exactly one place.
 """

flocks/provider/sdk/google.py

@@ -81,7 +81,7 @@ def _convert_messages(
         Rewrites history as text to bypass binary thought_signature requirements.
 
         ``session_id`` is forwarded by the runner via kwargs (see
-        ``SessionRunner._call_llm``).  When provided, we attempt to reconstruct
+        ``StepEngine._call_llm``).  When provided, we attempt to reconstruct
         the conversation directly from persisted session messages – including
         reasoning parts – which gives Gemini perfect context.  As a defensive
         fallback we also honour ``messages[0].sessionID`` / ``session_id``

flocks/server/routes/session.py

@@ -1740,30 +1740,24 @@ async def unshare_session_local(sessionID: str, http_request: Request) -> Sessio
 async def _abort_session_processing(sessionID: str) -> bool:
     """Abort active processing for a session and notify subscribers.
 
-    Aborts both the SessionLoop (sets abort_event so the next step check
-    stops the loop) and the SessionRunner (stops the current LLM stream).
+    Aborts SessionLoop, which owns the active StepEngine abort signal.
     Also auto-rejects any pending Question tool requests so the question
     handler polling loop unblocks immediately instead of timing out.
 
     Cascades abort to all child sub-agent sessions (synchronous subtasks
     and background tasks) so they stop together with the parent.
     """
-    from flocks.session.runner import SessionRunner
     from flocks.session.session_loop import SessionLoop
     from flocks.server.routes.question import reject_session_questions
 
     # Abort the loop-level context (propagates to runner via shared abort_event)
     loop_aborted = SessionLoop.abort(sessionID)
 
-    # Also cancel through the runner's own path (sets status to idle)
-    SessionRunner.cancel(sessionID)
-
     # Unblock any pending Question tool waiting for user input
     questions_rejected = await reject_session_questions(sessionID)
 
     # --- Cascade abort to child sub-agent sessions ---
     children_loops_aborted = SessionLoop.abort_children(sessionID)
-    children_runners_cancelled = SessionRunner.cancel_children(sessionID)
 
     # Cancel background sub-agent tasks spawned by this session
     bg_cancelled = 0
@@ -1778,7 +1772,6 @@ async def _abort_session_processing(sessionID: str) -> bool:
         "loop_aborted": loop_aborted,
         "questions_rejected": questions_rejected,
         "children_loops_aborted": children_loops_aborted,
-        "children_runners_cancelled": children_runners_cancelled,
         "bg_tasks_cancelled": bg_cancelled,
     })
 
@@ -1835,7 +1828,7 @@ class InitRequest(BaseModel):
 )
 async def initialize_session(sessionID: str, request: InitRequest, http_request: Request) -> bool:
     """Initialize session"""
-    from flocks.session.runner import SessionRunner
+    from flocks.session.actions import render_session_command
 
     current_user = require_user(http_request)
     session = await _get_session_by_id_unfiltered(sessionID)
@@ -1847,12 +1840,10 @@ async def initialize_session(sessionID: str, request: InitRequest, http_request:
     _require_session_write_access(session, current_user)
 
     # Execute INIT command
-    await SessionRunner.command(
+    await render_session_command(
         session_id=sessionID,
         command="init",
         arguments="",
-        message_id=request.messageID,
-        model=f"{request.providerID}/{request.modelID}",
     )
 
     log.info("session.initialized", {"session_id": sessionID})
@@ -3471,7 +3462,6 @@ async def _process_session_message(
     from flocks.agent.registry import Agent
     from flocks.provider.provider import Provider
     from flocks.session.session_loop import SessionLoop, LoopCallbacks
-    from flocks.session.runner import RunnerCallbacks
     import time
     import os
 
@@ -4945,7 +4935,7 @@ class ShellRequest(BaseModel):
 async def run_shell_command(sessionID: str, request: ShellRequest, http_request: Request):
     """Run shell command"""
     from flocks.hooks.execution import ExecutionStopped
-    from flocks.session.runner import SessionRunner
+    from flocks.session.actions import run_session_shell
 
     current_user = require_user(http_request)
     session = await _get_session_by_id_unfiltered(sessionID)
@@ -4956,17 +4946,12 @@ async def run_shell_command(sessionID: str, request: ShellRequest, http_request:
         )
     _require_session_write_access(session, current_user)
 
-    model = None
-    if request.model:
-        model = {"providerID": request.model.providerID, "modelID": request.model.modelID}
-
     try:
         async with Session.active_operation(sessionID):
-            result = await SessionRunner.shell(
+            result = await run_session_shell(
                 session_id=sessionID,
                 agent=request.agent,
                 command=request.command,
-                model=model,
             )
     except SessionNotFoundError as exc:
         raise HTTPException(

flocks/session/__init__.py

@@ -35,13 +35,6 @@
 from flocks.session.prompt import SessionPrompt, SystemPrompt, ContextInfo
 from flocks.session.lifecycle.compaction import SessionCompaction, CompactionResult, CompactionPolicy, ContextTier
 from flocks.session.lifecycle.summary import SessionSummary, FileDiff
-from flocks.session.runner import (
-    SessionRunner,
-    RunnerCallbacks,
-    ToolCall,
-    StepResult,
-    run_session,
-)
 from flocks.session.session_loop import (
     SessionLoop,
     LoopContext,
@@ -104,12 +97,6 @@
     # Summary
     "SessionSummary",
     "FileDiff",
-    # Runner
-    "SessionRunner",
-    "RunnerCallbacks",
-    "ToolCall",
-    "StepResult",
-    "run_session",
     # Session Loop
     "SessionLoop",
     "LoopContext",