mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
[F011] playbook: de-index rejected/archived playbooks from the PLAYBOOKS RAG index
This commit is contained in:
@@ -877,6 +877,48 @@ class OptimalService:
|
|||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
|
||||||
|
async def unindex_playbook(self, playbook_id: str) -> None:
|
||||||
|
"""De-index a playbook from the PLAYBOOKS index (best-effort).
|
||||||
|
|
||||||
|
The mirror of :meth:`index_playbook`: removes the playbook's embedded
|
||||||
|
chunks from the vector store AND drops its tracking row, so a
|
||||||
|
rejected/archived playbook stops surfacing in agent briefings. Both
|
||||||
|
steps are idempotent — a never-indexed draft playbook is a clean no-op.
|
||||||
|
Failures are logged and swallowed so a curation action (reject/archive)
|
||||||
|
never errors on the index side.
|
||||||
|
"""
|
||||||
|
from roboco.db import get_db_context
|
||||||
|
from roboco.services.repositories import IndexedDocumentRepository
|
||||||
|
|
||||||
|
try:
|
||||||
|
plugin = self._get_plugin(IndexType.PLAYBOOKS)
|
||||||
|
if isinstance(plugin, PlaybooksIndexPlugin):
|
||||||
|
await plugin.delete_playbook(playbook_id)
|
||||||
|
else:
|
||||||
|
source = f"roboco://playbooks/{playbook_id}"
|
||||||
|
await plugin._require_store.delete_by_source(source)
|
||||||
|
except Exception as exc:
|
||||||
|
logger.warning(
|
||||||
|
"Playbook de-index (vector store) failed; continuing",
|
||||||
|
playbook_id=playbook_id,
|
||||||
|
error=str(exc),
|
||||||
|
)
|
||||||
|
return
|
||||||
|
|
||||||
|
try:
|
||||||
|
async with get_db_context() as db:
|
||||||
|
repo = IndexedDocumentRepository(db)
|
||||||
|
await repo.delete_by_source(
|
||||||
|
IndexType.PLAYBOOKS.value,
|
||||||
|
f"roboco://playbooks/{playbook_id}",
|
||||||
|
)
|
||||||
|
except Exception as exc:
|
||||||
|
logger.warning(
|
||||||
|
"Playbook de-index (tracking row) failed; continuing",
|
||||||
|
playbook_id=playbook_id,
|
||||||
|
error=str(exc),
|
||||||
|
)
|
||||||
|
|
||||||
# =========================================================================
|
# =========================================================================
|
||||||
# INDEXING OPERATIONS (New - Optimal Brain)
|
# INDEXING OPERATIONS (New - Optimal Brain)
|
||||||
# =========================================================================
|
# =========================================================================
|
||||||
|
|||||||
@@ -70,6 +70,19 @@ class PlaybooksIndexPlugin(BaseIndexPlugin):
|
|||||||
tags=params.tags,
|
tags=params.tags,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
async def delete_playbook(self, playbook_id: str) -> None:
|
||||||
|
"""Remove a playbook's embedded chunks from the vector store.
|
||||||
|
|
||||||
|
Used when a playbook is rejected/archived after it was approved+indexed,
|
||||||
|
so it stops surfacing in agent briefings as a stale, no-longer-canonical
|
||||||
|
procedure. Idempotent: the store's ``delete_by_source`` no-ops when no
|
||||||
|
chunks match the source URI.
|
||||||
|
"""
|
||||||
|
source = self.build_source_uri(doc_id=playbook_id)
|
||||||
|
if not source:
|
||||||
|
return
|
||||||
|
await self._require_store.delete_by_source(source)
|
||||||
|
|
||||||
async def search_playbooks(
|
async def search_playbooks(
|
||||||
self,
|
self,
|
||||||
query: str,
|
query: str,
|
||||||
|
|||||||
@@ -120,9 +120,33 @@ class PlaybookService(BaseService):
|
|||||||
playbook.approved_by = approver_id
|
playbook.approved_by = approver_id
|
||||||
playbook.approved_at = datetime.now(UTC)
|
playbook.approved_at = datetime.now(UTC)
|
||||||
await self.session.flush()
|
await self.session.flush()
|
||||||
|
await self._unindex_playbook(playbook)
|
||||||
self.log.info("Playbook rejected", playbook_id=str(playbook_id), reason=reason)
|
self.log.info("Playbook rejected", playbook_id=str(playbook_id), reason=reason)
|
||||||
return playbook
|
return playbook
|
||||||
|
|
||||||
|
async def _unindex_playbook(self, playbook: PlaybookTable) -> None:
|
||||||
|
"""De-index a playbook from the PLAYBOOKS RAG index (best-effort).
|
||||||
|
|
||||||
|
The mirror of :meth:`_index_approved`: a rejected/archived playbook that
|
||||||
|
was previously approved+indexed must stop surfacing in agent briefings,
|
||||||
|
so ``reject`` drops its chunks + tracking row. Gated on
|
||||||
|
``org_memory_enabled`` (inert when the loop is off) and best-effort (a
|
||||||
|
failure never blocks the curation). Idempotent on a never-indexed draft.
|
||||||
|
"""
|
||||||
|
if not settings.org_memory_enabled:
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
from roboco.services.optimal import get_optimal_service
|
||||||
|
|
||||||
|
optimal = await get_optimal_service()
|
||||||
|
await optimal.unindex_playbook(str(playbook.id))
|
||||||
|
except Exception as exc:
|
||||||
|
self.log.warning(
|
||||||
|
"Playbook de-index-on-reject failed (best-effort)",
|
||||||
|
playbook_id=str(playbook.id),
|
||||||
|
error=str(exc),
|
||||||
|
)
|
||||||
|
|
||||||
async def list_drafts(self) -> list[PlaybookTable]:
|
async def list_drafts(self) -> list[PlaybookTable]:
|
||||||
return await self._list_by_status(PlaybookStatus.DRAFT)
|
return await self._list_by_status(PlaybookStatus.DRAFT)
|
||||||
|
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ Repository for managing indexed documents in the knowledge base.
|
|||||||
import hashlib
|
import hashlib
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
from sqlalchemy import select
|
from sqlalchemy import delete, select
|
||||||
|
|
||||||
from roboco.db.tables import IndexedDocumentTable
|
from roboco.db.tables import IndexedDocumentTable
|
||||||
from roboco.services.repositories.base import BaseRepository
|
from roboco.services.repositories.base import BaseRepository
|
||||||
@@ -100,6 +100,25 @@ class IndexedDocumentRepository(BaseRepository[IndexedDocumentTable]):
|
|||||||
"""Count documents in an index type."""
|
"""Count documents in an index type."""
|
||||||
return await self.count(IndexedDocumentTable.index_type == index_type)
|
return await self.count(IndexedDocumentTable.index_type == index_type)
|
||||||
|
|
||||||
|
async def delete_by_source(self, index_type: str, source: str) -> bool:
|
||||||
|
"""Delete the indexed-document tracking row for one source URI.
|
||||||
|
|
||||||
|
Used by de-index paths (e.g. a rejected/archived playbook) to drop the
|
||||||
|
tracking row whose chunks the vector store has already removed by the
|
||||||
|
same source. Idempotent: returns ``False`` (nothing deleted) when no
|
||||||
|
tracking row exists, ``True`` when one was removed.
|
||||||
|
"""
|
||||||
|
source_hash = hashlib.sha256(source.encode()).hexdigest()
|
||||||
|
result = await self.session.execute(
|
||||||
|
delete(IndexedDocumentTable).where(
|
||||||
|
IndexedDocumentTable.index_type == index_type,
|
||||||
|
IndexedDocumentTable.source_hash == source_hash,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
await self.session.flush()
|
||||||
|
rowcount: int = getattr(result, "rowcount", 0) or 0
|
||||||
|
return rowcount > 0
|
||||||
|
|
||||||
async def delete_by_index_type(self, index_type: str) -> int:
|
async def delete_by_index_type(self, index_type: str) -> int:
|
||||||
"""Delete all documents for an index type."""
|
"""Delete all documents for an index type."""
|
||||||
docs = await self.get_by_index_type(index_type, limit=10000)
|
docs = await self.get_by_index_type(index_type, limit=10000)
|
||||||
|
|||||||
@@ -38,3 +38,24 @@ def test_build_source_uri_with_id() -> None:
|
|||||||
|
|
||||||
def test_build_source_uri_none_when_missing() -> None:
|
def test_build_source_uri_none_when_missing() -> None:
|
||||||
assert _plugin().build_source_uri(doc_id=None) is None
|
assert _plugin().build_source_uri(doc_id=None) is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_delete_playbook_removes_its_chunks_by_source() -> None:
|
||||||
|
"""F011: deleting a playbook removes its embedded chunks from the vector
|
||||||
|
store by the playbook's source URI (idempotent — no-op if absent). A
|
||||||
|
rejected/archived playbook must not stay retrievable in the PLAYBOOKS index."""
|
||||||
|
from unittest.mock import AsyncMock, MagicMock
|
||||||
|
|
||||||
|
plugin = PlaybooksIndexPlugin.__new__(PlaybooksIndexPlugin)
|
||||||
|
store = MagicMock()
|
||||||
|
store.delete_by_source = AsyncMock(return_value=None)
|
||||||
|
# Bypass the initialized-guard property so the unit test doesn't need a
|
||||||
|
# live pgvector store.
|
||||||
|
object.__setattr__(plugin, "_initialized", True)
|
||||||
|
object.__setattr__(plugin, "_store", store)
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
asyncio.run(plugin.delete_playbook("pb-42"))
|
||||||
|
|
||||||
|
store.delete_by_source.assert_awaited_once_with("roboco://playbooks/pb-42")
|
||||||
|
|||||||
@@ -0,0 +1,128 @@
|
|||||||
|
"""F011 — PlaybookService.reject must de-index the playbook.
|
||||||
|
|
||||||
|
A rejected/archived playbook that was previously approved stays in the
|
||||||
|
PLAYBOOKS RAG index (no de-index on reject/archive), so it keeps surfacing
|
||||||
|
in agent briefings as a stale, no-longer-canonical procedure. The fix
|
||||||
|
mirrors the index-on-approve path: ``reject`` calls an
|
||||||
|
``_unindex_playbook`` helper (gated on ``org_memory_enabled``, best-effort)
|
||||||
|
that removes the playbook's chunks + tracking row.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from unittest.mock import AsyncMock, MagicMock, patch
|
||||||
|
from uuid import uuid4
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from roboco.config import settings
|
||||||
|
from roboco.models.base import PlaybookStatus
|
||||||
|
from roboco.services.playbook import PlaybookService
|
||||||
|
|
||||||
|
|
||||||
|
def _mock_playbook(playbook_id, *, status=PlaybookStatus.APPROVED.value):
|
||||||
|
pb = MagicMock()
|
||||||
|
pb.id = playbook_id
|
||||||
|
pb.status = status
|
||||||
|
pb.title = "Retry a flaky pg test"
|
||||||
|
pb.problem = "..."
|
||||||
|
pb.procedure = "..."
|
||||||
|
pb.tags = ["backend"]
|
||||||
|
pb.team = "backend"
|
||||||
|
pb.scope = "org"
|
||||||
|
return pb
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_reject_deindexes_approved_playbook(
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
) -> None:
|
||||||
|
"""Rejecting an approved (indexed) playbook removes it from the index."""
|
||||||
|
monkeypatch.setattr(settings, "org_memory_enabled", True)
|
||||||
|
playbook_id = uuid4()
|
||||||
|
pb = _mock_playbook(playbook_id, status=PlaybookStatus.APPROVED.value)
|
||||||
|
|
||||||
|
session = MagicMock()
|
||||||
|
# _get_or_raise returns the playbook; flush is a no-op.
|
||||||
|
result = MagicMock()
|
||||||
|
result.scalar_one_or_none.return_value = pb
|
||||||
|
session.execute = AsyncMock(return_value=result)
|
||||||
|
session.flush = AsyncMock()
|
||||||
|
|
||||||
|
optimal = MagicMock()
|
||||||
|
optimal.unindex_playbook = AsyncMock(return_value=None)
|
||||||
|
|
||||||
|
svc = PlaybookService(session)
|
||||||
|
with (
|
||||||
|
patch(
|
||||||
|
"roboco.services.optimal.get_optimal_service",
|
||||||
|
AsyncMock(return_value=optimal),
|
||||||
|
),
|
||||||
|
):
|
||||||
|
out = await svc.reject(playbook_id, approver_id=uuid4(), reason="stale")
|
||||||
|
|
||||||
|
assert out.status == PlaybookStatus.ARCHIVED.value
|
||||||
|
optimal.unindex_playbook.assert_awaited_once_with(str(playbook_id))
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_reject_of_unindexed_draft_does_not_error(
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
) -> None:
|
||||||
|
"""A draft (never indexed) reject still de-indexes (idempotent no-op) and
|
||||||
|
never errors the curation."""
|
||||||
|
monkeypatch.setattr(settings, "org_memory_enabled", True)
|
||||||
|
playbook_id = uuid4()
|
||||||
|
pb = _mock_playbook(playbook_id, status=PlaybookStatus.DRAFT.value)
|
||||||
|
|
||||||
|
session = MagicMock()
|
||||||
|
result = MagicMock()
|
||||||
|
result.scalar_one_or_none.return_value = pb
|
||||||
|
session.execute = AsyncMock(return_value=result)
|
||||||
|
session.flush = AsyncMock()
|
||||||
|
|
||||||
|
optimal = MagicMock()
|
||||||
|
optimal.unindex_playbook = AsyncMock(return_value=None)
|
||||||
|
|
||||||
|
svc = PlaybookService(session)
|
||||||
|
with (
|
||||||
|
patch(
|
||||||
|
"roboco.services.optimal.get_optimal_service",
|
||||||
|
AsyncMock(return_value=optimal),
|
||||||
|
),
|
||||||
|
):
|
||||||
|
out = await svc.reject(playbook_id, approver_id=uuid4(), reason="nope")
|
||||||
|
|
||||||
|
assert out.status == PlaybookStatus.ARCHIVED.value
|
||||||
|
# De-index is still called (idempotent — the store no-ops on an absent
|
||||||
|
# source), so a draft reject mirrors the approved reject path uniformly.
|
||||||
|
optimal.unindex_playbook.assert_awaited_once_with(str(playbook_id))
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_reject_skips_deindex_when_org_memory_disabled(
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
) -> None:
|
||||||
|
"""org_memory_enabled=False ⇒ the index is inert; reject must not touch it."""
|
||||||
|
monkeypatch.setattr(settings, "org_memory_enabled", False)
|
||||||
|
playbook_id = uuid4()
|
||||||
|
pb = _mock_playbook(playbook_id)
|
||||||
|
|
||||||
|
session = MagicMock()
|
||||||
|
result = MagicMock()
|
||||||
|
result.scalar_one_or_none.return_value = pb
|
||||||
|
session.execute = AsyncMock(return_value=result)
|
||||||
|
session.flush = AsyncMock()
|
||||||
|
|
||||||
|
optimal = MagicMock()
|
||||||
|
optimal.unindex_playbook = AsyncMock(return_value=None)
|
||||||
|
|
||||||
|
svc = PlaybookService(session)
|
||||||
|
with (
|
||||||
|
patch(
|
||||||
|
"roboco.services.optimal.get_optimal_service",
|
||||||
|
AsyncMock(return_value=optimal),
|
||||||
|
),
|
||||||
|
):
|
||||||
|
await svc.reject(playbook_id, approver_id=uuid4(), reason="nope")
|
||||||
|
|
||||||
|
optimal.unindex_playbook.assert_not_awaited()
|
||||||
Reference in New Issue
Block a user