`1b724ec`

Reset task state between top-level prompts

Authored by mfwolffe <wolffemf@dukes.jmu.edu> 3 weeks ago

SHA: 1b724ecb488beb058a0904b1ad499aa9384eb0e2
Parents: 36eb700
Tree: 1561ea3

8 changed files

Status	File	+	-
M	`src/loader/agent/loop.py`	5	0
M	`src/loader/runtime/launcher.py`	17	3
M	`src/loader/runtime/runtime_handle.py`	5	0
M	`src/loader/tools/workflow_tools.py`	2	2
M	`src/loader/ui/adapter.py`	1	1
M	`src/loader/ui/app.py`	1	0
A	`src/loader/utils/todos.py`	25	0
M	`tests/test_runtime_launcher.py`	55	0

src/loader/agent/loop.pymodified

      @current_task.setter
      def current_task(self, value: str | None) -> None:
 +        if self._current_task == value:
 +            return
          self._current_task = value
 +        self._system_message = None
 +        if hasattr(self, "session") and self.session is not None:
 +            self.session.update_runtime_state(current_task=value)
      @property
      def active_permission_mode(self) -> str:

src/loader/runtime/launcher.pymodified

  from __future__ import annotations
  from ..llm.base import Message, Role
 +from ..utils.todos import active_todo_store_path, clear_active_todos
  from .bootstrap import (
      RuntimeBootstrapSource,
      RuntimeBootstrapView,
  from .conversation import ConfirmationHandler, ConversationRuntime, EventSink, UserQuestionHandler
  from .decomposition_lane import DecompositionTurnRunner
  from .deliberation import should_decompose
 -from .events import TurnSummary
 +from .events import AgentEvent, TurnSummary
  from .explore import ExploreRuntime
  from .task_classification import is_conversational
  from .workflow import WorkflowMode
          if is_conversational(user_message):
              return await self.run_conversational(user_message, emit)
 -        if self.source.current_task is None:
 -            self.source.current_task = user_message
 +        await self._begin_top_level_task(user_message, emit)
          requested_mode = self._requested_workflow_mode(use_plan)
              return WorkflowMode.PLAN.value
          return None
 +    async def _begin_top_level_task(self, user_message: str, emit: EventSink) -> None:
 +        """Reset task-scoped state before a new top-level task starts."""
++
 +        previous_task = self.source.current_task
 +        had_active_todos = active_todo_store_path(self.source.project_root).exists()
 +        self.source.current_task = user_message
++
 +        if previous_task == user_message:
 +            return
++
 +        clear_active_todos(self.source.project_root)
 +        if previous_task is not None or had_active_todos:
 +            await emit(AgentEvent(type="todo_update", todo_items=[]))
++
  def build_runtime_launcher(source: RuntimeBootstrapSource) -> RuntimeLauncher:
      """Build a public runtime launcher from the shared bootstrap source."""

src/loader/runtime/runtime_handle.pymodified

      @current_task.setter
      def current_task(self, value: str | None) -> None:
 +        if self._current_task == value:
 +            return
          self._current_task = value
 +        self._system_message = None
 +        if hasattr(self, "session") and self.session is not None:
 +            self.session.update_runtime_state(current_task=value)
      @property
      def active_permission_mode(self) -> str:

src/loader/tools/workflow_tools.pymodified

  from typing import Any
  from ..runtime.permissions import PermissionMode
 +from ..utils.todos import active_todo_store_path
  from .base import Tool, ToolResult
  UserQuestionHandler = Callable[[str, list[str] | None], Awaitable[str]]
+         )
      def _store_path(self) -> Path:
 -        root = self.workspace_root or Path.cwd()
 -        return root / ".loader" / "todos" / "active.json"
 +        return active_todo_store_path(self.workspace_root or Path.cwd())
      def _read_existing_items(self, store_path: Path) -> list[dict[str, Any]]:
          if not store_path.exists():

src/loader/ui/adapter.pymodified

                  self.app.post_message(ResponseComplete(content=event.content))
              case "todo_update":
 -                if event.todo_items:
 +                if event.todo_items is not None:
                      self.app.post_message(TodoListUpdated(todos=event.todo_items))
              case "confirmation":

src/loader/ui/app.pymodified

          self._streamed_content = False
          self._tool_widget_queue.clear()
          self.shell_owner.clear_history()
 +        self.query_one("#todo-list", TodoListWidget).update_todos([])
          self.query_one(StatusLine).clear_definition_of_done()
          self.query_one(StatusLine).update_session_id(self.shell_owner.session.session_id)
          self.query_one(StatusLine).update_runtime_owner(

src/loader/utils/todos.pyadded

 +"""Helpers for the active Loader todo store."""
++
 +from __future__ import annotations
++
 +from pathlib import Path
++
++
 +def active_todo_store_path(workspace_root: Path | str | None) -> Path:
 +    """Return the path to the active todo store for a workspace."""
++
 +    root = (
 +        Path(workspace_root).expanduser().resolve()
 +        if workspace_root is not None
 +        else Path.cwd()
 +    )
 +    return root / ".loader" / "todos" / "active.json"
++
++
 +def clear_active_todos(workspace_root: Path | str | None) -> Path:
 +    """Delete the active todo store if it exists."""
++
 +    path = active_todo_store_path(workspace_root)
 +    if path.exists():
 +        path.unlink()
 +    return path

tests/test_runtime_launcher.pymodified

  from loader.llm.base import CompletionResponse, StreamChunk
  from loader.runtime.bootstrap import RuntimeBootstrapView
  from loader.runtime.launcher import RuntimeLauncher, build_runtime_launcher
 +from loader.runtime.public_shell import get_runtime_shell_system_message
  from loader.runtime.runtime_handle import RuntimeHandle
 +from loader.utils.todos import active_todo_store_path
  from tests.helpers.runtime_harness import ScriptedBackend
+         }
+     ]
      assert events == []
++
++
 +@pytest.mark.asyncio
 +async def test_runtime_launcher_resets_task_scoped_state_for_new_top_level_prompt(
 +    temp_dir: Path,
 +) -> None:
 +    backend = ScriptedBackend(
 +        completions=[CompletionResponse(content="Penguins page shipped.")]
 +    )
 +    handle = RuntimeHandle(
 +        backend=backend,
 +        config=AgentConfig(
 +            auto_context=False,
 +            stream=False,
 +            reasoning=ReasoningConfig(completion_check=False),
 +        ),
 +        project_root=temp_dir,
 +    )
 +    handle.current_task = "Create a collection of animal pages."
 +    cached_prompt = get_runtime_shell_system_message(handle)
 +    assert "Create a collection of animal pages." in cached_prompt.content
++
 +    todo_store = active_todo_store_path(temp_dir)
 +    todo_store.parent.mkdir(parents=True, exist_ok=True)
 +    todo_store.write_text(
 +        '[{"content": "Build cat page", "active_form": "Building cat page", "status": "pending"}]'
 +    )
++
 +    launcher = build_runtime_launcher(handle)
 +    events = []
++
 +    async def emit(event) -> None:
 +        events.append(event)
++
 +    response = await launcher.run_user_message(
 +        "Generate penguins.html and penguins.css for the new page.",
 +        emit,
 +        use_plan=False,
 +    )
++
 +    assert response == "Penguins page shipped."
 +    assert handle.current_task == "Generate penguins.html and penguins.css for the new page."
 +    assert not todo_store.exists()
 +    assert any(
 +        event.type == "todo_update" and event.todo_items == []
 +        for event in events
 +    )
 +    invocation = backend.invocations[-1]
 +    assert (
 +        "Current task: Generate penguins.html and penguins.css for the new page."
 +        in invocation.messages[0].content
 +    )
 +    assert "Current task: Create a collection of animal pages." not in invocation.messages[0].content