diff --git a/roboco/services/prompter_live.py b/roboco/services/prompter_live.py index 11633b7e..f1384f36 100644 --- a/roboco/services/prompter_live.py +++ b/roboco/services/prompter_live.py @@ -34,6 +34,13 @@ logger = structlog.get_logger() SDK_PORT = 9000 # Sentinel pushed onto a session's queue to end its SSE stream. _CLOSE = object() +# While the panel holds the SSE stream open, refresh ``last_activity`` on this +# cadence even with no events — an open stream means the human is watching (e.g. +# reading a proposed draft / MegaTask without typing), so the idle reaper must +# not retire the chat out from under them. Well under +# ``interactive_idle_reap_seconds`` (default 1800). When the tab closes the +# generator is cancelled, the refresh stops, and the chat reaps normally. +_STREAM_KEEPALIVE_SECONDS = 60.0 @dataclass @@ -182,16 +189,37 @@ class PrompterLiveRegistry: and now - s.last_activity > threshold_seconds ] + @staticmethod + async def _keepalive(session: LiveIntakeSession) -> None: + """Refresh ``last_activity`` while a stream is connected (see ``stream``).""" + while True: + await asyncio.sleep(_STREAM_KEEPALIVE_SECONDS) + session.last_activity = time.monotonic() + async def stream(self, session_id: str) -> AsyncIterator[dict[str, Any]]: - """Yield queued agent events until the session is closed.""" + """Yield queued agent events until the session is closed. + + A connected stream keeps the session's ``last_activity`` fresh (a + ``_keepalive`` task ticks every ``_STREAM_KEEPALIVE_SECONDS`` even with + no events), so the idle reaper treats an open, actively-watched chat as + alive — a human reading a proposed draft without typing is not "idle". + The keepalive runs beside an un-cancelled ``queue.get()`` (so no live + token/``_CLOSE`` is lost) and is cancelled when the stream ends; on + client disconnect the refresh stops and an abandoned chat reaps normally. + """ session = self._sessions.get(session_id) if session is None: return - while True: - item = await session.queue.get() - if item is _CLOSE: - return - yield item + keepalive = asyncio.create_task(self._keepalive(session)) + try: + while True: + item = await session.queue.get() + if item is _CLOSE: + return + session.last_activity = time.monotonic() + yield item + finally: + keepalive.cancel() # -- panel -> agent ---------------------------------------------------- diff --git a/tests/integration/test_git_conventions_pr.py b/tests/integration/test_git_conventions_pr.py index 8f7aa96a..4f7d8378 100644 --- a/tests/integration/test_git_conventions_pr.py +++ b/tests/integration/test_git_conventions_pr.py @@ -6,6 +6,7 @@ import subprocess from typing import TYPE_CHECKING, Any from uuid import uuid4 +from roboco.config import settings from roboco.db.tables import AgentTable, ProjectTable from roboco.models import AgentRole, AgentStatus, Team from roboco.services.git import _ConventionsPr, get_git_service @@ -23,7 +24,9 @@ def _git(repo: Path, *args: str) -> None: subprocess.run(["git", *args], cwd=repo, check=True, capture_output=True, text=True) -async def _seed_project(db: AsyncSession, workspace_path: str) -> ProjectTable: +async def _seed_project( + db: AsyncSession, workspace_path: str, *, slug: str | None = None +) -> ProjectTable: agent = AgentTable( id=uuid4(), name="Dev", @@ -42,7 +45,7 @@ async def _seed_project(db: AsyncSession, workspace_path: str) -> ProjectTable: project = ProjectTable( id=uuid4(), name="G-Proj", - slug=f"g-proj-{uuid4().hex[:8]}", + slug=slug or f"g-proj-{uuid4().hex[:8]}", git_url="https://example.com/r.git", default_branch="master", assigned_cell=Team.BACKEND, @@ -55,10 +58,15 @@ async def _seed_project(db: AsyncSession, workspace_path: str) -> ProjectTable: async def test_open_conventions_pr_commits_locally_without_remote( - db_session: AsyncSession, tmp_path: Path + db_session: AsyncSession, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: - repo = tmp_path / "repo" - repo.mkdir() + # open_conventions_pr only accepts a workspace_path under + # {workspaces_root}/{slug} (the containment guard), so anchor the root at + # the test dir and place the repo under the project's own slug. + monkeypatch.setattr(settings, "workspaces_root", str(tmp_path)) + slug = f"g-proj-{uuid4().hex[:8]}" + repo = tmp_path / slug / "repo" + repo.mkdir(parents=True) _git(repo, "init", "-b", "master") _git(repo, "config", "user.email", "t@example.com") _git(repo, "config", "user.name", "T") @@ -67,7 +75,7 @@ async def test_open_conventions_pr_commits_locally_without_remote( _git(repo, "add", "README.md") _git(repo, "commit", "-m", "init") - project = await _seed_project(db_session, str(repo)) + project = await _seed_project(db_session, str(repo), slug=slug) git = get_git_service(db_session) result = await git.open_conventions_pr( project.slug, diff --git a/tests/unit/services/test_prompter_live.py b/tests/unit/services/test_prompter_live.py index 974acf7a..26331eba 100644 --- a/tests/unit/services/test_prompter_live.py +++ b/tests/unit/services/test_prompter_live.py @@ -8,6 +8,7 @@ import time import httpx import pytest +import roboco.services.prompter_live as pl from roboco.services.prompter_live import ( PrompterLiveRegistry, get_live_registry, @@ -151,6 +152,35 @@ async def test_stream_yields_queued_events_then_ends_on_close() -> None: ] +@pytest.mark.asyncio +async def test_stream_keepalive_keeps_watched_chat_alive( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An open SSE stream refreshes last_activity even with no events, so a + human reading a proposal (tab open, not typing) is not idle-reaped. Without + this the chat "drops after a while" mid-review.""" + monkeypatch.setattr(pl, "_STREAM_KEEPALIVE_SECONDS", 0.01) + reg = PrompterLiveRegistry() + session = reg.open("s1", "intake-1") + session.last_activity = time.monotonic() - 4000 # silent for >1h + + # Before anyone connects, the abandoned-looking chat IS idle-reapable. + assert ("s1", "intake-1") in reg.idle_session_ids(1800) + + async def watch() -> None: + async for _ in reg.stream("s1"): + pass + + task = asyncio.create_task(watch()) + try: + await asyncio.sleep(0.05) # let a keepalive tick fire on the open stream + # The connected stream refreshed activity → no longer idle. + assert reg.idle_session_ids(1800) == [] + finally: + reg.close("s1") + await asyncio.wait_for(task, timeout=1.0) + + @pytest.mark.asyncio async def test_stream_unknown_session_is_empty() -> None: reg = PrompterLiveRegistry()