refactor(session): consolidate session and agent runtime loops

flocks/agent/agents/hephaestus/prompt_builder.py

@@ -29,15 +29,13 @@ def inject(
         available_agents=available_agents,
         available_tools=tools,
         available_skills=skills,
-        use_task_system=False,
     )
 
 
 def build_hephaestus_prompt(
     available_agents: List["AvailableAgent"],
     available_tools: List["AvailableTool"],
     available_skills: List["AvailableSkill"],
-    use_task_system: bool = False,
 ) -> str:
     from flocks.agent.prompt_utils import (
         build_agent_selection_table,
@@ -62,7 +60,7 @@ def build_hephaestus_prompt(
     oracle_section = build_oracle_section(available_agents)
     hard_blocks = build_hard_blocks_section()
     anti_patterns = build_anti_patterns_section()
-    todo_discipline = _todo_discipline_section(use_task_system)
+    todo_discipline = _todo_discipline_section()
 
     template = """You are Hephaestus, an autonomous deep worker for software engineering.
 
@@ -245,44 +243,7 @@ def build_hephaestus_prompt(
     return prompt
 
 
-def _todo_discipline_section(use_task_system: bool) -> str:
-    if use_task_system:
-        return """## Task Discipline (NON-NEGOTIABLE)
-
-**Track ALL multi-step work with tasks. This is your execution backbone.**
-
-### When to Create Tasks (MANDATORY)
-
-| Trigger | Action |
-|---------|--------|
-| 2+ step task | `TaskCreate` FIRST, atomic breakdown |
-| Uncertain scope | `TaskCreate` to clarify thinking |
-| Complex single task | Break down into trackable steps |
-
-### Workflow (STRICT)
-
-1. **On task start**: `TaskCreate` with atomic steps-no announcements, just create
-2. **Before each step**: `TaskUpdate(status="in_progress")` (ONE at a time)
-3. **After each step**: `TaskUpdate(status="completed")` IMMEDIATELY (NEVER batch)
-4. **Scope changes**: Update tasks BEFORE proceeding
-
-### Why This Matters
-
-- **Execution anchor**: Tasks prevent drift from original request
-- **Recovery**: If interrupted, tasks enable seamless continuation
-- **Accountability**: Each task = explicit commitment to deliver
-
-### Anti-Patterns (BLOCKING)
-
-| Violation | Why It Fails |
-|-----------|--------------|
-| Skipping tasks on multi-step work | Steps get forgotten, user has no visibility |
-| Batch-completing multiple tasks | Defeats real-time tracking purpose |
-| Proceeding without `in_progress` | No indication of current work |
-| Finishing without completing tasks | Task appears incomplete |
-
-**NO TASKS ON MULTI-STEP WORK = INCOMPLETE WORK.**"""
-
+def _todo_discipline_section() -> str:
     return """## Todo Discipline (NON-NEGOTIABLE)
 
 **Track ALL multi-step work with todos. This is your execution backbone.**

flocks/agent/agents/rex/prompt_builder.py

@@ -30,7 +30,6 @@ def inject(
         available_tools=tools,
         available_skills=skills,
         available_workflows=workflows or [],
-        use_task_system=False,
     )
 
 
@@ -39,7 +38,6 @@ def build_dynamic_rex_prompt(
     available_tools: List["AvailableTool"],
     available_skills: List["AvailableSkill"],
     available_workflows: Optional[List["AvailableWorkflow"]] = None,
-    use_task_system: bool = False,
 ) -> str:
     from flocks.agent.prompt_utils import (
         build_agent_selection_table,
@@ -58,12 +56,8 @@ def build_dynamic_rex_prompt(
     im_send_section = _build_im_send_pointer_section()
     anti_patterns = _build_rex_anti_patterns_section()
     command_guidance_section = _build_command_guidance_section()
-    task_management_section = _task_management_section(use_task_system)
-    todo_hook_note = (
-        "YOUR TASK CREATION WOULD BE TRACKED BY HOOK([SYSTEM REMINDER - TASK CONTINUATION])"
-        if use_task_system
-        else "YOUR TODO CREATION WOULD BE TRACKED BY HOOK([SYSTEM REMINDER - TODO CONTINUATION])"
-    )
+    task_management_section = _task_management_section()
+    todo_hook_note = "YOUR TODO CREATION WOULD BE TRACKED BY HOOK([SYSTEM REMINDER - TODO CONTINUATION])"
 
     template = """<Role>
 You are "Rex" - Powerful AI orchestrator for security operations.
@@ -143,7 +137,7 @@ def build_dynamic_rex_prompt(
 - Match existing codebase patterns when editing.
 - Fix bugs minimally; do not refactor during a bugfix unless required.
 - Keep search bounded: stop when you have enough context, when results repeat, or when direct evidence already answers the question.
-- For independent parallel branches whose results are needed this turn, emit multiple foreground `delegate_task` / `task` tool calls in the same assistant turn. The runtime executes those sibling tool calls concurrently and returns all tool results before you continue.
+- For independent parallel branches whose results are needed this turn, emit multiple foreground `delegate_task` tool calls in the same assistant turn. The runtime executes those sibling tool calls concurrently and returns all tool results before you continue.
 - Do not use `run_in_background=true`; background subagent execution is disabled.
 
 ## 5. Verify
@@ -278,55 +272,42 @@ def _build_clarification_protocol() -> str:
 ```"""
 
 
-def _task_management_section(use_task_system: bool) -> str:
-    title = "Task Management" if use_task_system else "Todo Management"
-    unit = "tasks" if use_task_system else "todos"
-    create_action = "`TaskCreate`" if use_task_system else '`todo(action="write")`'
-    progress_action = (
-        '`TaskUpdate(status="in_progress")`'
-        if use_task_system
-        else "mark `in_progress`"
-    )
-    complete_action = (
-        '`TaskUpdate(status="completed")`'
-        if use_task_system
-        else "mark `completed`"
-    )
+def _task_management_section() -> str:
     clarification_protocol = _build_clarification_protocol()
 
-    return f"""<Task_Management>
-## {title}
+    return f"""<Todo_Management>
+## Todo Management
 
-Use {unit} as the primary coordination mechanism for non-trivial execution work.
+Use todos as the primary coordination mechanism for non-trivial execution work.
 
 ### When They Are Mandatory
 
 | Trigger | Action |
 |---------|--------|
-| Multi-step work (2+ steps) | Create {unit} first |
-| Uncertain scope | Create {unit} to structure the work |
-| User request with multiple items | Create {unit} first |
-| Complex single task | Break it into {unit} |
+| Multi-step work (2+ steps) | Create todos first |
+| Uncertain scope | Create todos to structure the work |
+| User request with multiple items | Create todos first |
+| Complex single task | Break it into todos |
 
 ### Operating Rules
 
-1. Start with {create_action} before implementation work begins.
-2. ONLY add {unit} when the user wants execution, not when they only want analysis or planning.
-3. Before each step, {progress_action}. Keep only one item in progress.
-4. After each step, {complete_action} immediately. Never batch updates.
-5. If scope changes, update the {unit} before continuing.
+1. Start with `todo(action="write")` before implementation work begins.
+2. ONLY add todos when the user wants execution, not when they only want analysis or planning.
+3. Before each step, mark it `in_progress`. Keep only one item in progress.
+4. After each step, mark it `completed` immediately. Never batch updates.
+5. If scope changes, update the todos before continuing.
 
 ### Failure Modes
 
 | Violation | Why It Breaks the Workflow |
 |-----------|----------------------------|
-| Skipping {unit} on non-trivial work | The user loses progress visibility and steps get dropped |
-| Batch-completing multiple {unit} | Real-time tracking becomes meaningless |
+| Skipping todos on non-trivial work | The user loses progress visibility and steps get dropped |
+| Batch-completing multiple todos | Real-time tracking becomes meaningless |
 | Proceeding without an in-progress item | It is unclear what is being worked on |
 | Finishing without closing items | The work appears incomplete |
 
 {clarification_protocol}
-</Task_Management>"""
+</Todo_Management>"""
 
 
 def _build_security_priority_section(available_agents: List["AvailableAgent"]) -> str:

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/commands/import_.py

@@ -98,6 +98,17 @@ def _normalize_part_data(
     metadata = normalized.get("metadata")
     metadata_dict = metadata if isinstance(metadata, dict) else {}
 
+    if part_type == "subtask":
+        normalized["type"] = "text"
+        normalized["text"] = ""
+        normalized["ignored"] = True
+        normalized["metadata"] = {
+            **metadata_dict,
+            "legacyPartType": "subtask",
+        }
+        part_type = "text"
+        metadata_dict = normalized["metadata"]
+
     if "content" in normalized and "text" not in normalized:
         normalized["text"] = normalized.get("content", "")
 
@@ -135,10 +146,6 @@ def _normalize_part_data(
         )
     elif part_type == "agent":
         normalized.setdefault("name", metadata_dict.get("name") or normalized.get("content") or "agent")
-    elif part_type == "subtask":
-        normalized.setdefault("prompt", metadata_dict.get("prompt") or normalized.get("content", ""))
-        normalized.setdefault("description", metadata_dict.get("description") or "")
-        normalized.setdefault("agent", metadata_dict.get("agent") or "agent")
     elif part_type == "retry":
         normalized.setdefault("attempt", metadata_dict.get("attempt") or 1)
         normalized.setdefault("error", metadata_dict.get("error") or {})

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",
 ]