fix(api): unconditional /git/file window cap; defer Telegram sends after commit; docs/map periodic re-index

_compute_file_range now caps any resolved window at _FILE_MAX_LINES instead of only the exact whole-file shape, closing the near-whole-file bypass. Telegram sends ride a generalized after-commit outbox (defer_after_commit over the F107 machinery) so a slow Bot API can no longer hold the caller's transaction open; TelegramClient grows an abstract close(). The KB update loop iterates AUTO_INDEX_DIRS so docs/map edits re-index without a restart. PR-label application catches all exceptions per its never-raises contract, and pr_merge's CEO-only message names the resolved branch.
2026-08-03 07:23:24 +02:00 · 2026-07-15 08:25:20 +02:00
parent 85ac6422ff
commit f1ff149b70
9 changed files with 337 additions and 102 deletions
@@ -102,9 +102,9 @@ def _compute_file_range(
    """Resolve the (start, end, truncated) slice for a file-content read.

    Explicit ``start``/``end`` win; else ``line`` centers a context window;
-    else the whole file. The whole-file case is capped at ``_FILE_MAX_LINES``.
-    Returns 1-based inclusive [start, end] and whether the slice is shorter
-    than the file.
+    else the whole file. Whichever branch resolves the window, it is capped
+    at ``_FILE_MAX_LINES`` lines afterward. Returns 1-based inclusive
+    [start, end] and whether the slice is shorter than the file.
    """
    if start is not None and end is not None:
        s, e_ = start, end
@@ -118,8 +118,8 @@ def _compute_file_range(
    e_ = max(s, min(e_, total))

    truncated = e_ < total
-    if s == 1 and e_ == total and total > _FILE_MAX_LINES:
-        e_ = _FILE_MAX_LINES
+    if e_ - s + 1 > _FILE_MAX_LINES:
+        e_ = s + _FILE_MAX_LINES - 1
        truncated = True
    return s, e_, truncated

@@ -2383,7 +2383,7 @@ class GitService(BaseService):
                    },
                    json={"name": name, "color": self._PR_LABEL_COLOR},
                )
-        except httpx.HTTPError as e:
+        except Exception as e:
            self.log.warning("PR label ensure HTTP error", label=name, error=str(e))
            return
        # 422 (already_exists) / 409 (conflict) = the label is already present.
@@ -2423,7 +2423,7 @@ class GitService(BaseService):
                    },
                    json={"labels": labels},
                )
-        except httpx.HTTPError as e:
+        except Exception as e:
            self.log.warning("add PR labels HTTP error", pr=pr_number, error=str(e))
            return
        if not resp.is_success:
@@ -4575,20 +4575,23 @@ class GitService(BaseService):
        git_token = await self._get_project_token_or_raise(project.slug)
        owner, repo = self._parse_github_remote(workspace)

-        # CEO is the only one who merges to master. This agent-facing merge path
-        # (a cell PM merging a leaf/cell PR up the chain) may NEVER target a
-        # repo's default branch — a root→master PR is merged solely by the CEO
-        # via approve-&-merge (merge_pr_for_task, CEO-gated from
-        # awaiting_ceo_approval). Agents open the master PR and escalate.
+        # CEO is the only one who merges into the project's head environment
+        # branch (ladder index 0 — "master" only when no ladder is declared;
+        # see _project_default_branch). This agent-facing merge path (a cell
+        # PM merging a leaf/cell PR up the chain) may NEVER target it — that
+        # PR is merged solely by the CEO via approve-&-merge
+        # (merge_pr_for_task, CEO-gated from awaiting_ceo_approval). Agents
+        # open the PR to it and escalate.
        default_branch = await self._project_default_branch(project.slug)
        if target == default_branch:
            raise UnauthorizedError(
                action="pr_merge",
                reason=(
-                    "CEO_ONLY: merging into the default branch "
-                    f"('{default_branch}') is reserved for the CEO via "
-                    "approve-&-merge from awaiting_ceo_approval. Open the PR "
-                    "and escalate; agents never merge to master."
+                    f"CEO_ONLY: merging into '{default_branch}' (this "
+                    "project's head environment branch) is reserved for the "
+                    "CEO via approve-&-merge from awaiting_ceo_approval. "
+                    "Open the PR and escalate; agents never merge directly "
+                    f"into '{default_branch}'."
                ),
            )

@@ -11,6 +11,7 @@ Also implements the ACK system for tracking acknowledgments.

 import asyncio
 import contextlib
+from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from typing import TYPE_CHECKING, Any, ClassVar, Literal, cast
@@ -64,7 +65,7 @@ def _format_completion_body(task: TaskTable, metrics: "TaskMetrics | None") -> s


 # =============================================================================
-# Deferred bus publish — transactional outbox (F107)
+# Deferred after-commit work — transactional outbox (F107)
 # =============================================================================
 # `deliver`/`_persist_and_deliver` run inside the caller's open transaction:
 # the notification row is flushed but not committed. Publishing the
@@ -73,74 +74,73 @@ def _format_completion_body(task: TaskTable, metrics: "TaskMetrics | None") -> s
 # error) rolled the row back while connected WebSocket clients had already
 # received a push for an id that no longer existed. The fix defers the bus
 # publish to the session's `after_commit` so a rollback drops the pending
-# event: the row is durable by the time the event fires.
+# event: the row is durable by the time the event fires. The same queue also
+# carries the outbound Telegram send (see `_notify_telegram`) — any
+# network-adjacent side effect that must not hold the transaction open rides
+# this mechanism.
 #
-# The pending events and the scheduled drain tasks live on `session.info` so
+# The pending work and the scheduled drain tasks live on `session.info` so
 # they are scoped to the session's lifetime (no module-global state, no
 # cross-request leak). A sync `after_commit` listener schedules the async
 # drain via `asyncio.create_task` (the listener runs synchronously inside
 # `await AsyncSession.commit()` on the loop thread, so the running loop is
 # available); `after_rollback` clears the pending queue so a rolled-back
-# transaction emits nothing.
+# transaction runs none of it.

-_PENDING_PUBLISHES_KEY = "_roboco_pending_bus_publishes"
+_PENDING_WORK_KEY = "_roboco_pending_bus_publishes"
 _DRAIN_TASKS_KEY = "_roboco_drain_tasks"
 _DRAIN_REGISTERED_KEY = "_roboco_drain_registered"


-async def _drain_pending_publishes(pending: list[Event]) -> None:
-    """Publish every deferred event best-effort once the txn has committed.
-
-    The bus is read fresh at drain time (it may have reconnected between
-    deferral and commit); a disconnected bus is a silent no-op, matching the
-    prior inline behavior. Each publish is independent — one failure does
-    not drop the rest.
+async def _drain_pending_work(pending: list[Callable[[], Awaitable[None]]]) -> None:
+    """Run every deferred after-commit action best-effort once the txn has
+    committed. Each callable is independent — one failure does not stop the
+    rest — and is expected to be fully exception-safe on its own; the
+    try/except here is a defensive backstop, not the primary guard.
    """
-    if not pending:
-        return
-    bus = get_event_bus()
-    if not bus.is_connected():
-        return
-    for ev in pending:
+    for work in pending:
        try:
-            await bus.publish(ev)
+            await work()
        except Exception as e:  # best-effort: never break the drain
-            _log.warning("Deferred bus publish failed", error=str(e))
+            _log.warning("Deferred after-commit work failed", error=str(e))


-def _schedule_pending_publishes(session: AsyncSession) -> None:
-    """`after_commit` handler: hand the pending events to the running loop.
+def _schedule_pending_work(session: AsyncSession) -> None:
+    """`after_commit` handler: hand the pending work to the running loop.

    Sync listener — runs inside `await AsyncSession.commit()`, so the event
    loop is active. The created task is stashed on the session so callers /
    tests can await it deterministically; in production it is fire-and-forget
    (best-effort, matching the prior try/except semantics).
    """
-    pending = session.info.pop(_PENDING_PUBLISHES_KEY, None)
+    pending = session.info.pop(_PENDING_WORK_KEY, None)
    if not pending:
        return
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:  # no running loop — nothing we can do, drop silently
        return
-    task = loop.create_task(_drain_pending_publishes(pending))
+    task = loop.create_task(_drain_pending_work(pending))
    session.info.setdefault(_DRAIN_TASKS_KEY, []).append(task)


-def _discard_pending_publishes(session: AsyncSession) -> None:
-    """`after_rollback` handler: a rolled-back txn emits nothing (no phantom)."""
-    session.info.pop(_PENDING_PUBLISHES_KEY, None)
+def _discard_pending_work(session: AsyncSession) -> None:
+    """`after_rollback` handler: a rolled-back txn runs none of it (no phantom)."""
+    session.info.pop(_PENDING_WORK_KEY, None)


-def defer_bus_publish(session: AsyncSession, ev: Event) -> None:
-    """Enqueue a bus event to fire only after the session's transaction commits.
+def defer_after_commit(
+    session: AsyncSession, work: Callable[[], Awaitable[None]]
+) -> None:
+    """Enqueue a zero-arg async callable to run only after the session's
+    transaction commits; dropped (never run) on rollback.

    Registers one-shot `after_commit` / `after_rollback` listeners on the
    session the first time it is called for that session; subsequent calls
    just append. The listeners are bound to the session instance and are
    collected with it (no global listener accumulation).
    """
-    session.info.setdefault(_PENDING_PUBLISHES_KEY, []).append(ev)
+    session.info.setdefault(_PENDING_WORK_KEY, []).append(work)
    if session.info.get(_DRAIN_REGISTERED_KEY):
        return
    session.info[_DRAIN_REGISTERED_KEY] = True
@@ -149,11 +149,27 @@ def defer_bus_publish(session: AsyncSession, ev: Event) -> None:

    @event.listens_for(sync_session, "after_commit")
    def _on_commit(_sync_session: object) -> None:
-        _schedule_pending_publishes(session)
+        _schedule_pending_work(session)

    @event.listens_for(sync_session, "after_rollback")
    def _on_rollback(_sync_session: object) -> None:
-        _discard_pending_publishes(session)
+        _discard_pending_work(session)
+
+
+def defer_bus_publish(session: AsyncSession, ev: Event) -> None:
+    """Enqueue a bus event to fire only after the session's transaction commits.
+
+    Thin wrapper over `defer_after_commit`: the bus is read fresh at drain
+    time (it may have reconnected between deferral and commit); a
+    disconnected bus is a silent no-op, matching the prior inline behavior.
+    """
+
+    async def _publish() -> None:
+        bus = get_event_bus()
+        if bus.is_connected():
+            await bus.publish(ev)
+
+    defer_after_commit(session, _publish)


 class EscalationError(ValueError):
@@ -879,8 +895,12 @@ class NotificationDeliveryService(BaseService):
        """Best-effort Telegram DM to the CEO alongside an in-app notification.

        Degrades to a no-op unless ``telegram_enabled`` is armed and credentials
-        are stored. Never raises into the caller — a network/credentials failure
-        only logs. The message carries a panel deep-link when ``panel_base_url``
+        are stored. Credentials are fetched now (a fast DB read on the open
+        session); the actual network send is deferred via
+        ``defer_after_commit`` so a slow Telegram Bot API call can't hold the
+        caller's open transaction for up to ``telegram_timeout_seconds``.
+        Never raises into the caller — a credentials/network failure only
+        logs. The message carries a panel deep-link when ``panel_base_url``
        is set.
        """
        from roboco.config import settings
@@ -894,23 +914,31 @@ class NotificationDeliveryService(BaseService):

        try:
            creds = await get_telegram_credentials_service(self.session).get_decrypted()
-            client = build_telegram_client(
-                creds, timeout=settings.telegram_timeout_seconds
-            )
-            text = subject
-            if settings.panel_base_url:
-                link = f"{settings.panel_base_url.rstrip('/')}/tasks/{str(task_id)[:8]}"
-                text = f"{subject}\n{link}"
-            result = await client.send_message(text)
-            if not result.sent:
-                _log.warning("telegram_notify_skip", detail=result.detail)
        except Exception as exc:  # best-effort — never block the producer
            _log.warning("telegram_notify_failed", error=str(exc))
-        finally:
-            close = getattr(locals().get("client"), "close", None)
-            if close is not None:
-                with contextlib.suppress(Exception):
-                    await close()
+            return
+
+        text = subject
+        if settings.panel_base_url:
+            link = f"{settings.panel_base_url.rstrip('/')}/tasks/{str(task_id)[:8]}"
+            text = f"{subject}\n{link}"
+        timeout = settings.telegram_timeout_seconds
+
+        async def _send() -> None:
+            client = None
+            try:
+                client = build_telegram_client(creds, timeout=timeout)
+                result = await client.send_message(text)
+                if not result.sent:
+                    _log.warning("telegram_notify_skip", detail=result.detail)
+            except Exception as exc:  # best-effort — never break the drain
+                _log.warning("telegram_notify_failed", error=str(exc))
+            finally:
+                if client is not None:
+                    with contextlib.suppress(Exception):
+                        await client.close()
+
+        defer_after_commit(self.session, _send)

    async def notify_ceo_of_escalation(
        self,
@@ -59,6 +59,13 @@ logger = structlog.get_logger()
 # qwen3-embedding:0.6b retrieves higher quality chunks, so more context helps
 MAX_CONTENT_CHARS = 800

+# Directories under docs/ that are auto-indexed at startup AND watched by the
+# periodic update loop. docs/rag is the agent-facing RAG corpus; docs/map is
+# the agent-facing exhaustive codebase map. Both dirs' files route through
+# index_documentation, except any file under a "standards" subdir, which
+# routes to the standards indexer instead (see _index_doc_file).
+AUTO_INDEX_DIRS = ("rag", "map")
+

@dataclass
 class IndexingReport:
@@ -361,12 +368,7 @@ class OptimalService:
            )
            return None

-        # Directories to auto-index (RAG-optimized docs only). docs/rag is the
-        # agent-facing RAG corpus; docs/map is the agent-facing exhaustive
-        # codebase map — indexing it makes the map roboco_kb_search-able.
-        auto_index_dirs = ["rag", "map"]
-
-        for subdir in auto_index_dirs:
+        for subdir in AUTO_INDEX_DIRS:
            target_dir = docs_root / subdir
            if not target_dir.exists():
                continue
@@ -405,14 +407,8 @@ class OptimalService:
        # Index each file with individual error tracking
        for md_file in md_files:
            try:
-                # Use standards indexer for files in standards subdirectory
-                is_standards = name == "standards" or "standards" in md_file.parts
-                if is_standards:
-                    await self.index_standards_file(str(md_file))
-                    report.successful += 1
-                else:
-                    await self.index_documentation([str(md_file)])
-                    report.successful += 1
+                await self._index_doc_file(md_file, name)
+                report.successful += 1

                # Track mtime for periodic update detection
                import contextlib
@@ -439,6 +435,19 @@ class OptimalService:
        )
        return report

+    async def _index_doc_file(self, md_file: Path, name: str) -> None:
+        """Route a single doc file to the standards or general docs indexer.
+
+        Shared by the one-shot auto-index (_index_docs_directory) and the
+        periodic re-index path so both resolve the same index type for the
+        same file.
+        """
+        is_standards = name == "standards" or "standards" in md_file.parts
+        if is_standards:
+            await self.index_standards_file(str(md_file))
+        else:
+            await self.index_documentation([str(md_file)])
+
    # =========================================================================
    # PERIODIC UPDATE (File Change Detection)
    # =========================================================================
@@ -490,8 +499,8 @@ class OptimalService:
                return path
        return None

-    def _scan_for_file_changes(self, rag_dir: Path) -> tuple[list[Path], list[Path]]:
-        """Scan ``rag_dir`` for new or modified .md files.
+    def _scan_for_file_changes(self, directory: Path) -> tuple[list[Path], list[Path]]:
+        """Scan ``directory`` for new or modified .md files.

        Updates ``self._file_mtimes`` in place as a side effect.
        Returns (new_files, modified_files).
@@ -499,7 +508,7 @@ class OptimalService:
        new_files: list[Path] = []
        modified_files: list[Path] = []

-        for md_file in rag_dir.rglob("*.md"):
+        for md_file in directory.rglob("*.md"):
            file_path = str(md_file)
            try:
                current_mtime = md_file.stat().st_mtime
@@ -515,12 +524,13 @@ class OptimalService:

        return new_files, modified_files

-    async def _reindex_files(self, files_to_index: list[Path]) -> int:
-        """Re-index the given files and return the count that succeeded."""
+    async def _reindex_files(self, files_to_index: list[Path], name: str) -> int:
+        """Re-index the given files (from auto-index dir ``name``) and return
+        the count that succeeded."""
        indexed = 0
        for md_file in files_to_index:
            try:
-                await self.index_documentation([str(md_file)])
+                await self._index_doc_file(md_file, name)
                indexed += 1
                logger.debug("Re-indexed file", file=str(md_file))
            except Exception as e:
@@ -532,32 +542,38 @@ class OptimalService:
        return indexed

    async def _check_for_updates(self) -> None:
-        """Scan for new or modified files and index them."""
+        """Scan every AUTO_INDEX_DIRS dir for new or modified files and index them."""
        docs_root = self._resolve_docs_root()
        if docs_root is None:
            return

-        rag_dir = docs_root / "rag"
-        if not rag_dir.exists():
-            return
+        total_indexed = 0
+        total_files = 0
+        for subdir in AUTO_INDEX_DIRS:
+            target_dir = docs_root / subdir
+            if not target_dir.exists():
+                continue

-        new_files, modified_files = self._scan_for_file_changes(rag_dir)
-        files_to_index = new_files + modified_files
-        if not files_to_index:
-            return
+            new_files, modified_files = self._scan_for_file_changes(target_dir)
+            files_to_index = new_files + modified_files
+            if not files_to_index:
+                continue

-        logger.info(
-            "Detected file changes, re-indexing",
-            new_count=len(new_files),
-            modified_count=len(modified_files),
-        )
+            logger.info(
+                "Detected file changes, re-indexing",
+                dir=subdir,
+                new_count=len(new_files),
+                modified_count=len(modified_files),
+            )

-        indexed = await self._reindex_files(files_to_index)
-        if indexed > 0:
+            total_indexed += await self._reindex_files(files_to_index, subdir)
+            total_files += len(files_to_index)
+
+        if total_indexed > 0:
            logger.info(
                "Periodic update complete",
-                indexed=indexed,
-                total_files=len(files_to_index),
+                indexed=total_indexed,
+                total_files=total_files,
            )

    async def close(self) -> None:
@@ -41,10 +41,17 @@ class TelegramClient(ABC):
    @abstractmethod
    async def send_message(self, text: str) -> TelegramSendResult: ...

+    @abstractmethod
+    async def close(self) -> None:
+        """Release transport resources; no-op when no transport exists."""
+

 class NullTelegramClient(TelegramClient):
    """No credentials configured — every call is a no-op, never raises."""

+    async def close(self) -> None:
+        return None
+
    @property
    def configured(self) -> bool:
        return False
@@ -12,11 +12,14 @@ from typing import TYPE_CHECKING, cast
 from uuid import UUID, uuid4

 import pytest
+from roboco.config import settings
 from roboco.db.tables import AgentTable, NotificationTable
 from roboco.events import Event, EventType
 from roboco.models import AgentRole, AgentStatus, NotificationPriority, NotificationType
 from roboco.models.base import Team
 from roboco.services.notification_delivery import get_notification_delivery_service
+from roboco.services.telegram_client import TelegramSendResult
+from roboco.services.telegram_credentials import TelegramCredentialsData

 if TYPE_CHECKING:
    from sqlalchemy.ext.asyncio import AsyncSession
@@ -251,3 +254,99 @@ async def test_acknowledge_rollback_drops_phantom(
    await _await_drain(db_session)

    assert bus.published == []
+
+
+# --- Telegram send rides the same after-commit outbox, not an inline await ---
+
+
+@pytest.mark.asyncio
+async def test_notify_telegram_send_deferred_to_after_commit(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """``_notify_telegram`` must not block the caller's open transaction on
+    the Telegram Bot API call — the network send is deferred to the same
+    after-commit outbox the bus publish above uses, and never fires on a
+    rollback."""
+    monkeypatch.setattr(settings, "telegram_enabled", True)
+    monkeypatch.setattr(settings, "panel_base_url", "")
+
+    creds = TelegramCredentialsData(bot_token="t", chat_id="1")
+
+    class _FakeCredsService:
+        async def get_decrypted(self) -> TelegramCredentialsData:
+            return creds
+
+    monkeypatch.setattr(
+        "roboco.services.telegram_credentials.get_telegram_credentials_service",
+        lambda _session: _FakeCredsService(),
+    )
+
+    sent: list[str] = []
+
+    class _FakeTelegramClient:
+        async def send_message(self, text: str) -> TelegramSendResult:
+            sent.append(text)
+            return TelegramSendResult(sent=True)
+
+        async def close(self) -> None:
+            pass
+
+    monkeypatch.setattr(
+        "roboco.services.telegram_client.build_telegram_client",
+        lambda _creds, **_kwargs: _FakeTelegramClient(),
+    )
+
+    service = get_notification_delivery_service(db_session)
+    await service._notify_telegram(task_id=uuid4(), subject="Hello CEO")
+
+    # Pre-commit: the network send must not have fired yet.
+    assert sent == []
+
+    await db_session.commit()
+    await _await_drain(db_session)
+
+    assert sent == ["Hello CEO"]
+
+
+@pytest.mark.asyncio
+async def test_notify_telegram_rollback_drops_send(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A rollback drops the deferred Telegram send — it never fires for a
+    notification whose row never became durable."""
+    monkeypatch.setattr(settings, "telegram_enabled", True)
+    monkeypatch.setattr(settings, "panel_base_url", "")
+
+    creds = TelegramCredentialsData(bot_token="t", chat_id="1")
+
+    class _FakeCredsService:
+        async def get_decrypted(self) -> TelegramCredentialsData:
+            return creds
+
+    monkeypatch.setattr(
+        "roboco.services.telegram_credentials.get_telegram_credentials_service",
+        lambda _session: _FakeCredsService(),
+    )
+
+    sent: list[str] = []
+
+    class _FakeTelegramClient:
+        async def send_message(self, text: str) -> TelegramSendResult:
+            sent.append(text)
+            return TelegramSendResult(sent=True)
+
+        async def close(self) -> None:
+            pass
+
+    monkeypatch.setattr(
+        "roboco.services.telegram_client.build_telegram_client",
+        lambda _creds, **_kwargs: _FakeTelegramClient(),
+    )
+
+    service = get_notification_delivery_service(db_session)
+    await service._notify_telegram(task_id=uuid4(), subject="Hello CEO")
+
+    await db_session.rollback()
+    await _await_drain(db_session)
+
+    assert sent == []
@@ -52,3 +52,19 @@ class TestComputeFileRange:
            total=0, line=None, context=10, start=None, end=None
        )
        assert (s, e_, trunc) == (1, 1, False)
+
+    def test_near_whole_file_explicit_range_still_capped(self) -> None:
+        # start=1, end=total-1 is not the exact-whole-file shape, but the
+        # resolved window is still oversized and must be capped.
+        total = 50000
+        s, e_, trunc = _compute_file_range(
+            total=total, line=None, context=10, start=1, end=total - 1
+        )
+        assert (s, e_, trunc) == (1, _FILE_MAX_LINES, True)
+
+    def test_oversized_line_context_window_is_capped(self) -> None:
+        total = 10000
+        s, e_, trunc = _compute_file_range(
+            total=total, line=5000, context=3000, start=None, end=None
+        )
+        assert (s, e_, trunc) == (2000, 2000 + _FILE_MAX_LINES - 1, True)
@@ -714,6 +714,43 @@ async def test_ensure_base_falls_back_to_default_when_create_fails() -> None:
    assert out == "master"


+# ---------------------------------------------------------------------------
+# _apply_pr_labels / _ensure_label_exists: never raise, even on a non-httpx
+# error — labeling must not surface as a spurious tool failure after the PR
+# itself was already created.
+# ---------------------------------------------------------------------------
+
+
+def _non_httpx_raising_client() -> MagicMock:
+    """An AsyncClient whose POST raises a plain (non-httpx) exception."""
+    fake_client = MagicMock()
+    fake_client.__aenter__ = AsyncMock(return_value=fake_client)
+    fake_client.__aexit__ = AsyncMock(return_value=False)
+    fake_client.post = AsyncMock(side_effect=RuntimeError("boom"))
+    return fake_client
+
+
+@pytest.mark.asyncio
+async def test_ensure_label_exists_swallows_non_httpx_error() -> None:
+    svc = _service()
+    with patch(
+        "roboco.services.git.httpx.AsyncClient",
+        return_value=_non_httpx_raising_client(),
+    ):
+        await svc._ensure_label_exists("acme", "repo", "tok", "cell/backend")
+
+
+@pytest.mark.asyncio
+async def test_apply_pr_labels_swallows_non_httpx_error() -> None:
+    svc = _service()
+    _bind(svc, "_ensure_label_exists", AsyncMock())
+    with patch(
+        "roboco.services.git.httpx.AsyncClient",
+        return_value=_non_httpx_raising_client(),
+    ):
+        await svc._apply_pr_labels("acme", "repo", "tok", 11, ["cell/backend"])
+
+
 # ---------------------------------------------------------------------------
 # pr_merge: returns merge commit dict
 # ---------------------------------------------------------------------------
@@ -9,11 +9,15 @@ branch is needed — every docs/map/*.md rides index_documentation like docs/rag

 from __future__ import annotations

+from typing import TYPE_CHECKING
 from unittest.mock import AsyncMock, patch

 import pytest
 from roboco.services.optimal import IndexingReport, OptimalService

+if TYPE_CHECKING:
+    from pathlib import Path
+

@pytest.mark.asyncio
 async def test_auto_index_docs_includes_map_subdir() -> None:
@@ -30,3 +34,28 @@ async def test_auto_index_docs_includes_map_subdir() -> None:
    names = [c.args[1] for c in mock.call_args_list]
    assert "rag" in names
    assert "map" in names
+
+
+@pytest.mark.asyncio
+async def test_periodic_check_reindexes_modified_map_file(tmp_path: Path) -> None:
+    """The periodic loop must watch docs/map too, not just docs/rag — a
+    modified docs/map file with an already-tracked (stale) mtime is picked
+    up by _check_for_updates and routed through _index_doc_file with the
+    "map" dir name, exactly like the one-shot path resolves it."""
+    docs_root = tmp_path / "docs"
+    (docs_root / "rag").mkdir(parents=True)
+    (docs_root / "map").mkdir(parents=True)
+    map_file = docs_root / "map" / "CLAUDE.md"
+    map_file.write_text("# initial")
+
+    svc = object.__new__(OptimalService)
+    svc._docs_root = docs_root
+    # Simulate this file was already indexed at startup with a stale mtime,
+    # so the next scan sees it as modified.
+    svc._file_mtimes = {str(map_file): 0.0}
+
+    mock = AsyncMock()
+    with patch.object(svc, "_index_doc_file", new=mock):
+        await svc._check_for_updates()
+
+    mock.assert_awaited_once_with(map_file, "map")