mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
fix: journals/learnings never reached the RAG corpus + git-readonly slug 404s (#339)
* fix(rag): per-index chunk floors — journals and learnings were never indexed The global 200-char garbage floor (sized for code/doc chunks) discarded every templated journal note and most distilled org-memory lessons, silently: ingest returned success with zero chunks, so agent journals and learnings were never retrievable via RAG. IndexConfig now carries a per-type min_chunk_length (journals 40, learnings 80, others unchanged). * fix(mcp): git-readonly tools default project_slug from the container env Agents 404ed /api/git/status with 'Project not found: roboco' — the tools made the LLM supply the slug and six doc examples taught a slug that matches no registered project. The tools now fall back to the ROBOCO_PROJECT_SLUG the orchestrator already injects, and the stale examples are corrected. * feat(rag): startup backfill re-ingests zero-chunk journals and learnings Before the per-index chunk-floor fix, ingest() returned success with chunk_count=0 for undersized content: every historical journal entry and distilled learning below the (then-global) 200-char floor was durably recorded in journal_entries but silently never got a chunks_journals / chunks_learnings row, and no exception meant the existing dead-letter (rag_index_failures) never saw it either. Extends the startup reconcile (roboco/api/app.py _reconcile_rag_indexes) with a new pass: backfill_unindexed_journals (roboco/services/ rag_index_failures.py) queries journal_entries for rows missing from each vector table and re-ingests them through the same live code paths (_reindex_journal_entry / record_learning). Journals and learnings are backfilled independently since a LEARNING entry can clear the (lower) JOURNALS floor while still failing the (higher) LEARNINGS floor — a learning's doc_source is a content hash, not the entry id, so presence there is checked by hashing each candidate the same way LearningsIndexPlugin.record_learning does and batch-querying chunks_learnings for those exact sources. Bounded to 200 rows per pass per boot (converges over restarts on a larger backlog) and best-effort per row (one failure never aborts the pass). Rows still under the current floor are excluded by a length filter in the SELECT so they are never retried forever, and private entries are excluded from the JOURNALS pass exactly like the live indexing path. * test(rag): scope backfill assertions to their own rows --------- Co-authored-by: Renn F <rennf93@users.noreply.github.com>
This commit is contained in:
Renzo F
@@ -12,10 +12,13 @@ There is **no** "roboco_git_commit / _push / _create_pr / _merge_pr / _checkout"
|
||||
| `roboco_git_diff` | View changes |
|
||||
|
||||
```python
|
||||
status = roboco_git_status(project_slug="roboco")
|
||||
diff = roboco_git_diff(project_slug="roboco")
|
||||
log = roboco_git_log(project_slug="roboco", branch="feature/backend/a1b2c3d4")
|
||||
branches = roboco_git_branch_list(project_slug="roboco")
|
||||
# project_slug is optional — omit it and your own project is used
|
||||
# (from this agent's environment). Pass it explicitly only to inspect
|
||||
# a different project than the one you're assigned to.
|
||||
status = roboco_git_status()
|
||||
diff = roboco_git_diff()
|
||||
log = roboco_git_log(branch="feature/backend/a1b2c3d4")
|
||||
branches = roboco_git_branch_list()
|
||||
```
|
||||
|
||||
## Branch Lifecycle — automatic
|
||||
|
||||
@@ -15,7 +15,7 @@
|
||||
roboco_kb_search(
|
||||
query="rate limiting redis",
|
||||
top_k=5,
|
||||
project="roboco",
|
||||
project="roboco-api",
|
||||
index_types=["code", "docs"]
|
||||
)
|
||||
```
|
||||
@@ -71,14 +71,14 @@ roboco_docs_read(path="backend/api/endpoints.md")
|
||||
# Index code (PM, Developer)
|
||||
roboco_kb_index_code(
|
||||
sources=["src/**/*.py"],
|
||||
project="roboco"
|
||||
project="roboco-api"
|
||||
)
|
||||
|
||||
# Index docs (PM, Documenter) - for bulk/explicit indexing
|
||||
# Note: roboco_docs_write() auto-indexes when writing
|
||||
roboco_kb_index_docs(
|
||||
sources=["docs/**/*.md"],
|
||||
project="roboco"
|
||||
project="roboco-api"
|
||||
)
|
||||
```
|
||||
|
||||
|
||||
@@ -16,10 +16,12 @@ A task carries its project linkage; you don't look it up with a tool. The task o
|
||||
Read-only git inspection is available through the `roboco-git-readonly` MCP server (developers and QA):
|
||||
|
||||
```python
|
||||
roboco_git_status(project_slug="roboco")
|
||||
roboco_git_log(project_slug="roboco")
|
||||
roboco_git_diff(project_slug="roboco")
|
||||
roboco_git_branch_list(project_slug="roboco")
|
||||
# project_slug is optional on all four — omit it and your own project
|
||||
# is used (from this agent's environment).
|
||||
roboco_git_status()
|
||||
roboco_git_log()
|
||||
roboco_git_diff()
|
||||
roboco_git_branch_list()
|
||||
```
|
||||
|
||||
There is **no** `roboco_git_commit / _push / _checkout / _create_pr / _merge_pr` tool. Commits go through the `commit` content tool (auto- prefixed with `[task-id]`, auto-pushed by the choreographer); PRs open at `open_pr` time; merges are a PM `complete` operation.
|
||||
@@ -29,7 +31,7 @@ There is **no** `roboco_git_commit / _push / _checkout / _create_pr / _merge_pr`
|
||||
To learn how a project's codebase is laid out or how a subsystem works, query the knowledge base rather than a project tool:
|
||||
|
||||
```python
|
||||
roboco_kb_search(query="rate limiting redis", project="roboco",
|
||||
roboco_kb_search(query="rate limiting redis", project="roboco-api",
|
||||
index_types=["code", "documentation"])
|
||||
roboco_ask_mentor(question="How is auth wired up in this project?")
|
||||
```
|
||||
|
||||
@@ -57,8 +57,9 @@ You don't push or create a PR yourself. The choreographer pushed the commit duri
|
||||
|
||||
```python
|
||||
# Read-only inspection (any role) — roboco-git-readonly MCP
|
||||
status = roboco_git_status(project_slug="roboco")
|
||||
log = roboco_git_log(project_slug="roboco", branch="feature/backend/a1b2c3d4--def67890")
|
||||
diff = roboco_git_diff(project_slug="roboco")
|
||||
branches = roboco_git_branch_list(project_slug="roboco")
|
||||
# project_slug is optional — omit it and your own project is used.
|
||||
status = roboco_git_status()
|
||||
log = roboco_git_log(branch="feature/backend/a1b2c3d4--def67890")
|
||||
diff = roboco_git_diff()
|
||||
branches = roboco_git_branch_list()
|
||||
```
|
||||
|
||||
@@ -26,7 +26,7 @@ It searches ALL knowledge sources and supports follow-up questions.
|
||||
roboco_kb_search(
|
||||
query="rate limiting redis implementation",
|
||||
top_k=5, # Results to return
|
||||
project="roboco", # Optional project filter
|
||||
project="roboco-api", # Optional project filter
|
||||
index_types=["code", "docs"] # Filter by type
|
||||
)
|
||||
```
|
||||
|
||||
@@ -17,9 +17,10 @@ give_me_work()
|
||||
# guard at pass/fail time)
|
||||
claim_review(task_id="<task>")
|
||||
|
||||
# 3. Inspect the diff
|
||||
roboco_git_diff(project_slug="roboco")
|
||||
roboco_git_log(project_slug="roboco", branch="<dev's branch>")
|
||||
# 3. Inspect the diff (project_slug is optional — omit it and your
|
||||
# own project is used)
|
||||
roboco_git_diff()
|
||||
roboco_git_log(branch="<dev's branch>")
|
||||
|
||||
# 4. Run the relevant suite
|
||||
# Backend: uv run pytest && uv run ruff check . && uv run mypy roboco/
|
||||
|
||||
+18
-2
@@ -67,7 +67,7 @@ from roboco.services.extraction import ExtractionPipeline, ExtractionService
|
||||
from roboco.services.learning import get_learning_service
|
||||
from roboco.services.optimal import close_optimal_service, get_optimal_service
|
||||
from roboco.services.playbook import PlaybookService
|
||||
from roboco.services.rag_index_failures import reclaim_due
|
||||
from roboco.services.rag_index_failures import backfill_unindexed_journals, reclaim_due
|
||||
from roboco.services.settings import apply_persisted_feature_flags
|
||||
from roboco.services.transcription import TranscriptionService
|
||||
|
||||
@@ -125,10 +125,26 @@ async def _reclaim_rag_index_failures(app: FastAPI) -> None:
|
||||
logger.warning("RAG index dead-letter reclaim failed; continuing", error=str(e))
|
||||
|
||||
|
||||
async def _backfill_unindexed_journals(app: FastAPI) -> None:
|
||||
"""Re-index journal/learning entries silently zero-chunked before the
|
||||
per-index chunk-floor fix (see ``backfill_unindexed_journals``'s
|
||||
docstring). Best-effort: a failure here never blocks startup — the rows
|
||||
stay and the next startup retries them. Skipped when RAG is disabled.
|
||||
"""
|
||||
if app.state.optimal is None:
|
||||
return
|
||||
try:
|
||||
await backfill_unindexed_journals(app.state.optimal)
|
||||
except Exception as e:
|
||||
logger.warning("Journal/learning RAG backfill failed; continuing", error=str(e))
|
||||
|
||||
|
||||
async def _reconcile_rag_indexes(app: FastAPI) -> None:
|
||||
"""Run both RAG index reconcile passes (playbooks + dead-letter reclaim)."""
|
||||
"""Run all RAG index reconcile passes: playbooks, dead-letter reclaim,
|
||||
and the journals/learnings zero-chunk backfill."""
|
||||
await _reconcile_unindexed_playbooks(app)
|
||||
await _reclaim_rag_index_failures(app)
|
||||
await _backfill_unindexed_journals(app)
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
|
||||
@@ -14,8 +14,8 @@ Workspace Structure:
|
||||
+-- [git repo files]
|
||||
|
||||
Example:
|
||||
/data/workspaces/roboco/backend/be-dev-1/
|
||||
/data/workspaces/roboco/backend/be-dev-2/
|
||||
/data/workspaces/roboco-api/backend/be-dev-1/
|
||||
/data/workspaces/roboco-api/backend/be-dev-2/
|
||||
|
||||
This allows multiple agents to work on the same project in parallel,
|
||||
each on their own branch, without file conflicts.
|
||||
|
||||
+49
-12
@@ -70,6 +70,27 @@ def _get(path: str, params: dict[str, Any]) -> dict[str, Any]:
|
||||
return result
|
||||
|
||||
|
||||
_PROJECT_SLUG_ENV = "ROBOCO_PROJECT_SLUG"
|
||||
|
||||
|
||||
def _resolve_project_slug(project_slug: str | None) -> str | dict[str, Any]:
|
||||
"""Resolve the slug to query: explicit arg wins, else the orchestrator's
|
||||
env-injected slug for this agent's own project. Returns an error dict
|
||||
(never raises) when neither is available.
|
||||
"""
|
||||
slug = project_slug or os.environ.get(_PROJECT_SLUG_ENV)
|
||||
if not slug:
|
||||
return {
|
||||
"error": "missing_project_slug",
|
||||
"detail": (
|
||||
"no project_slug given and "
|
||||
f"{_PROJECT_SLUG_ENV} is not set in this agent's environment — "
|
||||
"pass project_slug explicitly"
|
||||
),
|
||||
}
|
||||
return slug
|
||||
|
||||
|
||||
def _cap_diff(result: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Truncate an oversized diff for context embedding; annotate the cut."""
|
||||
diff = result.get("diff")
|
||||
@@ -85,35 +106,43 @@ def _cap_diff(result: dict[str, Any]) -> dict[str, Any]:
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def roboco_git_status(project_slug: str) -> dict[str, Any]:
|
||||
def roboco_git_status(project_slug: str | None = None) -> dict[str, Any]:
|
||||
"""Read-only: current git status of your workspace.
|
||||
|
||||
Args:
|
||||
project_slug: Project slug (e.g. "roboco").
|
||||
project_slug: Optional — omit it and your own project is used
|
||||
(from this agent's environment).
|
||||
|
||||
Returns:
|
||||
Current branch, staged/unstaged/untracked files, ahead/behind counts.
|
||||
"""
|
||||
return _get("/api/git/status", {"project_slug": project_slug})
|
||||
slug = _resolve_project_slug(project_slug)
|
||||
if isinstance(slug, dict):
|
||||
return slug
|
||||
return _get("/api/git/status", {"project_slug": slug})
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def roboco_git_log(
|
||||
project_slug: str,
|
||||
project_slug: str | None = None,
|
||||
limit: int = 10,
|
||||
branch: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Read-only: recent commits on the named branch (default: current).
|
||||
|
||||
Args:
|
||||
project_slug: Project slug.
|
||||
project_slug: Optional — omit it and your own project is used
|
||||
(from this agent's environment).
|
||||
limit: Number of commits to return (max 50).
|
||||
branch: Branch to inspect; defaults to the current checked-out branch.
|
||||
|
||||
Returns:
|
||||
List of commits with hash, short_hash, message, author, date.
|
||||
"""
|
||||
params: dict[str, Any] = {"project_slug": project_slug, "limit": limit}
|
||||
slug = _resolve_project_slug(project_slug)
|
||||
if isinstance(slug, dict):
|
||||
return slug
|
||||
params: dict[str, Any] = {"project_slug": slug, "limit": limit}
|
||||
if branch is not None:
|
||||
params["branch"] = branch
|
||||
return _get("/api/git/log", params)
|
||||
@@ -121,21 +150,25 @@ def roboco_git_log(
|
||||
|
||||
@mcp.tool()
|
||||
def roboco_git_diff(
|
||||
project_slug: str,
|
||||
project_slug: str | None = None,
|
||||
staged: bool = False,
|
||||
file_path: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Read-only: diff of your workspace against the index.
|
||||
|
||||
Args:
|
||||
project_slug: Project slug.
|
||||
project_slug: Optional — omit it and your own project is used
|
||||
(from this agent's environment).
|
||||
staged: If True, show staged changes; otherwise show unstaged.
|
||||
file_path: Optional path to scope the diff to a single file.
|
||||
|
||||
Returns:
|
||||
Diff text plus files_changed count.
|
||||
"""
|
||||
params: dict[str, Any] = {"project_slug": project_slug, "staged": staged}
|
||||
slug = _resolve_project_slug(project_slug)
|
||||
if isinstance(slug, dict):
|
||||
return slug
|
||||
params: dict[str, Any] = {"project_slug": slug, "staged": staged}
|
||||
if file_path is not None:
|
||||
params["file_path"] = file_path
|
||||
return _cap_diff(_get("/api/git/diff", params))
|
||||
@@ -143,21 +176,25 @@ def roboco_git_diff(
|
||||
|
||||
@mcp.tool()
|
||||
def roboco_git_branch_list(
|
||||
project_slug: str,
|
||||
project_slug: str | None = None,
|
||||
include_remote: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
"""Read-only: list local (and optionally remote) branches.
|
||||
|
||||
Args:
|
||||
project_slug: Project slug.
|
||||
project_slug: Optional — omit it and your own project is used
|
||||
(from this agent's environment).
|
||||
include_remote: If True, also include remote-tracking branches.
|
||||
|
||||
Returns:
|
||||
Branches with current branch marked.
|
||||
"""
|
||||
slug = _resolve_project_slug(project_slug)
|
||||
if isinstance(slug, dict):
|
||||
return slug
|
||||
return _get(
|
||||
"/api/git/branches",
|
||||
{"project_slug": project_slug, "include_remote": include_remote},
|
||||
{"project_slug": slug, "include_remote": include_remote},
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -41,18 +41,30 @@ def build_doc_source(*, kind: str, id_: str | None) -> str | None:
|
||||
|
||||
_MIN_CHUNK_LENGTH = 200
|
||||
|
||||
# Per-index-type override of _MIN_CHUNK_LENGTH. Journal/learning entries are
|
||||
# short by design (templated notes, distilled Problem->Approach->Gotcha
|
||||
# lessons) — the global 200-char floor discarded every one of them as
|
||||
# "garbage" (raw_count=1 every time), so org-memory/journal retrieval never
|
||||
# had anything to find. Every other index keeps the default.
|
||||
_MIN_CHUNK_LENGTH_BY_TYPE: dict[IndexType, int] = {
|
||||
IndexType.JOURNALS: 40,
|
||||
IndexType.LEARNINGS: 80,
|
||||
}
|
||||
|
||||
def _filter_quality_chunks(raw_chunks: list[Any]) -> list[Any]:
|
||||
|
||||
def _filter_quality_chunks(
|
||||
raw_chunks: list[Any], min_chunk_length: int = _MIN_CHUNK_LENGTH
|
||||
) -> list[Any]:
|
||||
"""Drop tiny chunks and chunks that are mostly markdown formatting."""
|
||||
kept: list[Any] = []
|
||||
for chunk in raw_chunks:
|
||||
text = chunk.text.strip()
|
||||
if len(text) < _MIN_CHUNK_LENGTH:
|
||||
if len(text) < min_chunk_length:
|
||||
continue
|
||||
non_formatting = (
|
||||
text.replace("```", "").replace("---", "").replace("#", "").strip()
|
||||
)
|
||||
if len(non_formatting) < _MIN_CHUNK_LENGTH // 2:
|
||||
if len(non_formatting) < min_chunk_length // 2:
|
||||
continue
|
||||
kept.append(chunk)
|
||||
return kept
|
||||
@@ -70,6 +82,7 @@ class IndexConfig:
|
||||
embedding_model: str = "qwen3-embedding:0.6b"
|
||||
llm_model: str = "glm-5.2:cloud"
|
||||
llm_base_url: str = "http://roboco-ollama:11434/v1"
|
||||
min_chunk_length: int = _MIN_CHUNK_LENGTH
|
||||
|
||||
@classmethod
|
||||
def from_settings(cls, index_type: IndexType) -> "IndexConfig":
|
||||
@@ -90,6 +103,9 @@ class IndexConfig:
|
||||
embedding_model=settings.default_embedding_model,
|
||||
llm_model=settings.local_llm_model,
|
||||
llm_base_url=settings.local_llm_base_url,
|
||||
min_chunk_length=_MIN_CHUNK_LENGTH_BY_TYPE.get(
|
||||
index_type, _MIN_CHUNK_LENGTH
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@@ -435,7 +451,7 @@ class BaseIndexPlugin(ABC):
|
||||
embedder = self._require_embedder
|
||||
|
||||
raw_chunks: list[Chunk] = chunker.chunk_document(doc)
|
||||
chunks = _filter_quality_chunks(raw_chunks)
|
||||
chunks = _filter_quality_chunks(raw_chunks, self.config.min_chunk_length)
|
||||
if not chunks:
|
||||
logger.warning(
|
||||
"All chunks filtered as garbage",
|
||||
|
||||
@@ -8,24 +8,43 @@ reclaims due rows with backoff — on success the row is deleted, on failure
|
||||
``attempts`` bumps and ``next_retry_at`` advances. Best-effort throughout: a
|
||||
persist failure never blocks the caller's commit, and a reclaim failure never
|
||||
blocks startup.
|
||||
|
||||
A second, unrelated problem this module also repairs: before the per-index
|
||||
chunk-floor fix, ``ingest()`` returned success with ``chunk_count=0`` for
|
||||
undersized content (journal reflections, distilled learnings) — no exception
|
||||
was ever raised, so those rows never reached the dead-letter above and never
|
||||
got a vector row either. ``backfill_unindexed_journals`` re-ingests that
|
||||
silent-failure history straight from ``journal_entries`` (see its docstring).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import logging
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from typing import Any
|
||||
from uuid import UUID
|
||||
|
||||
from sqlalchemy import delete, func, select
|
||||
from sqlalchemy import delete, func, select, text
|
||||
|
||||
from roboco.db.base import get_db_context
|
||||
from roboco.db.tables import RagIndexFailureTable
|
||||
from roboco.models.optimal import IndexJournalEntryParams
|
||||
from roboco.models.optimal import IndexJournalEntryParams, IndexType
|
||||
from roboco.services.optimal_brain.indexes.base import IndexConfig
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Backoff ceiling: 1m, 2m, 4m, ... capped at 1h.
|
||||
_MAX_BACKOFF_SECONDS = 3600
|
||||
_BASE_BACKOFF_SECONDS = 60
|
||||
|
||||
# Per-pass cap on the startup backfill (below): bounded so a large historical
|
||||
# backlog can't stall startup. The candidate query already excludes rows below
|
||||
# the CURRENT floor (they would zero-chunk again), so anything left over past
|
||||
# the cap is picked up on the next restart — it converges, it may just take
|
||||
# more than one boot for a very large backlog.
|
||||
_BACKFILL_CAP = 200
|
||||
|
||||
|
||||
def _backoff(attempts: int) -> timedelta:
|
||||
seconds = min(_BASE_BACKOFF_SECONDS * (2 ** (attempts - 1)), _MAX_BACKOFF_SECONDS)
|
||||
@@ -206,8 +225,197 @@ async def _reindex_completion_learning(optimal: Any, payload: dict[str, Any]) ->
|
||||
)
|
||||
|
||||
|
||||
def _learning_source(content: str) -> str:
|
||||
"""Derive a learning's doc_source exactly as record_learning does.
|
||||
|
||||
Mirrors ``LearningsIndexPlugin.record_learning`` (learnings.py): the
|
||||
doc_id is a hash of the raw content, not the entry id, so presence in
|
||||
``chunks_learnings`` can't be joined by primary key — this lets the
|
||||
backfill below compute the same source and check for it directly.
|
||||
"""
|
||||
content_hash = hashlib.md5(content.encode(), usedforsecurity=False).hexdigest()[:16]
|
||||
return f"roboco://learnings/lrn-{content_hash}"
|
||||
|
||||
|
||||
async def _backfill_journals(optimal: Any) -> tuple[int, int]:
|
||||
"""Re-ingest non-private journal entries with no ``chunks_journals`` row.
|
||||
|
||||
Candidates are entries at/above the JOURNALS floor (a still-undersized
|
||||
entry would zero-chunk again — excluded here so it's never retried
|
||||
forever) whose ``roboco://journals/<id>`` source has zero chunk rows.
|
||||
Returns ``(processed, still_missing)`` for this capped batch.
|
||||
"""
|
||||
if not optimal.is_index_registered(IndexType.JOURNALS):
|
||||
return 0, 0
|
||||
floor = IndexConfig.from_settings(IndexType.JOURNALS).min_chunk_length
|
||||
async with get_db_context() as session:
|
||||
rows = (
|
||||
await session.execute(
|
||||
text(
|
||||
"""
|
||||
SELECT je.id, je.content, je.type, j.agent_id, je.task_id, je.tags
|
||||
FROM journal_entries je
|
||||
JOIN journals j ON j.id = je.journal_id
|
||||
LEFT JOIN chunks_journals cj
|
||||
ON cj.source = 'roboco://journals/' || je.id::text
|
||||
WHERE cj.source IS NULL
|
||||
AND je.is_private = false
|
||||
AND length(je.content) >= :floor
|
||||
ORDER BY je.created_at
|
||||
LIMIT :cap
|
||||
"""
|
||||
),
|
||||
{"floor": floor, "cap": _BACKFILL_CAP},
|
||||
)
|
||||
).all()
|
||||
|
||||
processed = 0
|
||||
for row in rows:
|
||||
try:
|
||||
await _reindex_journal_entry(
|
||||
optimal,
|
||||
{
|
||||
"content": row.content,
|
||||
"entry_type": row.type,
|
||||
"entry_id": str(row.id),
|
||||
"agent_id": str(row.agent_id) if row.agent_id else None,
|
||||
"task_id": str(row.task_id) if row.task_id else None,
|
||||
"tags": list(row.tags or []),
|
||||
"is_private": False,
|
||||
},
|
||||
)
|
||||
processed += 1
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"RAG backfill: journal re-index failed (best-effort), "
|
||||
f"entry_id={row.id}, error={e}"
|
||||
)
|
||||
return processed, len(rows) - processed
|
||||
|
||||
|
||||
async def _backfill_learnings(optimal: Any) -> tuple[int, int]:
|
||||
"""Re-ingest LEARNING journal entries with no matching ``chunks_learnings`` row.
|
||||
|
||||
A learning's doc_source is a content hash, not the entry id, so presence
|
||||
can't be joined in SQL — hash each candidate (``_learning_source``) and
|
||||
batch-check ``chunks_learnings`` for those exact sources. Independent of
|
||||
:func:`_backfill_journals`: a LEARNING entry can pass the (lower) JOURNALS
|
||||
floor but fail the (higher) LEARNINGS floor, so it may already have a
|
||||
``chunks_journals`` row while still missing here.
|
||||
"""
|
||||
if not optimal.is_index_registered(IndexType.LEARNINGS):
|
||||
return 0, 0
|
||||
floor = IndexConfig.from_settings(IndexType.LEARNINGS).min_chunk_length
|
||||
async with get_db_context() as session:
|
||||
rows = (
|
||||
await session.execute(
|
||||
text(
|
||||
"""
|
||||
SELECT je.id, je.content, j.agent_id, je.task_id, je.tags,
|
||||
je.is_private
|
||||
FROM journal_entries je
|
||||
JOIN journals j ON j.id = je.journal_id
|
||||
WHERE je.type = 'learning' AND length(je.content) >= :floor
|
||||
ORDER BY je.created_at
|
||||
LIMIT :cap
|
||||
"""
|
||||
),
|
||||
{"floor": floor, "cap": _BACKFILL_CAP},
|
||||
)
|
||||
).all()
|
||||
if not rows:
|
||||
return 0, 0
|
||||
|
||||
sources = [_learning_source(row.content) for row in rows]
|
||||
existing = {
|
||||
row[0]
|
||||
for row in (
|
||||
await session.execute(
|
||||
text(
|
||||
"SELECT source FROM chunks_learnings "
|
||||
"WHERE source = ANY(CAST(:sources AS text[]))"
|
||||
),
|
||||
{"sources": sources},
|
||||
)
|
||||
).all()
|
||||
}
|
||||
|
||||
from roboco.services.optimal_brain.indexes.learnings import (
|
||||
RecordLearningParams as _JournalLearningParams,
|
||||
)
|
||||
|
||||
processed = 0
|
||||
missing = 0
|
||||
for row, source in zip(rows, sources, strict=True):
|
||||
if source in existing:
|
||||
continue
|
||||
missing += 1
|
||||
try:
|
||||
await optimal.record_learning(
|
||||
_JournalLearningParams(
|
||||
content=row.content,
|
||||
category="journal_learning",
|
||||
agent_id=row.agent_id,
|
||||
task_id=row.task_id,
|
||||
shareable=not row.is_private,
|
||||
tags=list(row.tags or []),
|
||||
)
|
||||
)
|
||||
processed += 1
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"RAG backfill: learning re-index failed (best-effort), "
|
||||
f"entry_id={row.id}, error={e}"
|
||||
)
|
||||
return processed, missing - processed
|
||||
|
||||
|
||||
async def backfill_unindexed_journals(optimal: Any) -> dict[str, int]:
|
||||
"""Re-ingest journal/learning entries silently zero-chunked before the
|
||||
per-index chunk-floor fix.
|
||||
|
||||
``ingest()`` used to return success with ``chunk_count=0`` for undersized
|
||||
content, so historical journal entries and their derived learnings were
|
||||
durably recorded in ``journal_entries`` but never landed a row in
|
||||
``chunks_journals`` / ``chunks_learnings`` — invisible to RAG search
|
||||
though the dead-letter above only ever covered rows that raised, not rows
|
||||
that silently zero-chunked.
|
||||
|
||||
Each pass is independently gated (skipped if that index's plugin never
|
||||
initialized), capped per boot, and best-effort per row — one failing row
|
||||
never blocks the rest. Best-effort at this level too: a hard failure
|
||||
(e.g. a lost DB connection) never blocks startup, and it converges over
|
||||
restarts since a successfully re-indexed row simply stops matching the
|
||||
candidate query.
|
||||
"""
|
||||
try:
|
||||
j_processed, j_remaining = await _backfill_journals(optimal)
|
||||
except Exception as e:
|
||||
logger.warning(f"RAG backfill: journals pass failed; continuing, error={e}")
|
||||
j_processed, j_remaining = 0, 0
|
||||
try:
|
||||
l_processed, l_remaining = await _backfill_learnings(optimal)
|
||||
except Exception as e:
|
||||
logger.warning(f"RAG backfill: learnings pass failed; continuing, error={e}")
|
||||
l_processed, l_remaining = 0, 0
|
||||
|
||||
if j_processed or l_processed:
|
||||
logger.info(
|
||||
"RAG backfill: re-indexed historical zero-chunk entries "
|
||||
f"(journals processed={j_processed} remaining={j_remaining}, "
|
||||
f"learnings processed={l_processed} remaining={l_remaining})"
|
||||
)
|
||||
return {
|
||||
"journals_processed": j_processed,
|
||||
"journals_remaining": j_remaining,
|
||||
"learnings_processed": l_processed,
|
||||
"learnings_remaining": l_remaining,
|
||||
}
|
||||
|
||||
|
||||
__all__ = [
|
||||
"_serialize_journal_payload",
|
||||
"backfill_unindexed_journals",
|
||||
"count_failures",
|
||||
"persist_failure",
|
||||
"reclaim_due",
|
||||
|
||||
@@ -13,9 +13,9 @@ parallel development without conflicts:
|
||||
└── [git repo files]
|
||||
|
||||
Example:
|
||||
/data/workspaces/roboco/backend/be-dev-1/
|
||||
/data/workspaces/roboco/backend/be-dev-2/
|
||||
/data/workspaces/roboco/frontend/fe-dev-1/
|
||||
/data/workspaces/roboco-api/backend/be-dev-1/
|
||||
/data/workspaces/roboco-api/backend/be-dev-2/
|
||||
/data/workspaces/roboco-api/frontend/fe-dev-1/
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
"""project_slug defaults to ROBOCO_PROJECT_SLUG when omitted on every
|
||||
roboco-git-readonly tool — agents no longer need to guess a slug, and the
|
||||
RAG-taught literal "roboco" example (a non-existent slug) can't 404 them.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import types
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def git_module(monkeypatch: pytest.MonkeyPatch) -> types.ModuleType:
|
||||
monkeypatch.setenv("ROBOCO_AGENT_ID", "00000000-0000-0000-0000-000000000042")
|
||||
monkeypatch.setenv("ROBOCO_AGENT_ROLE", "developer")
|
||||
monkeypatch.setenv("ROBOCO_ORCHESTRATOR_URL", "http://test-orchestrator:8000")
|
||||
monkeypatch.delenv("ROBOCO_PROJECT_SLUG", raising=False)
|
||||
import roboco.mcp.git_readonly as srv
|
||||
|
||||
importlib.reload(srv)
|
||||
return srv
|
||||
|
||||
|
||||
def test_resolve_omitted_falls_back_to_env(
|
||||
git_module: types.ModuleType, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setenv("ROBOCO_PROJECT_SLUG", "roboco-api")
|
||||
assert git_module._resolve_project_slug(None) == "roboco-api"
|
||||
assert git_module._resolve_project_slug("") == "roboco-api"
|
||||
|
||||
|
||||
def test_resolve_supplied_passes_through(
|
||||
git_module: types.ModuleType, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setenv("ROBOCO_PROJECT_SLUG", "roboco-api")
|
||||
assert git_module._resolve_project_slug("roboco-panel") == "roboco-panel"
|
||||
|
||||
|
||||
def test_resolve_neither_present_returns_clear_error(
|
||||
git_module: types.ModuleType,
|
||||
) -> None:
|
||||
result = git_module._resolve_project_slug(None)
|
||||
assert isinstance(result, dict)
|
||||
assert result["error"] == "missing_project_slug"
|
||||
assert "ROBOCO_PROJECT_SLUG" in result["detail"]
|
||||
|
||||
|
||||
def _mock_response(payload: dict) -> MagicMock:
|
||||
resp = MagicMock()
|
||||
resp.raise_for_status.return_value = None
|
||||
resp.json.return_value = payload
|
||||
return resp
|
||||
|
||||
|
||||
def test_git_status_omitted_slug_uses_env(
|
||||
git_module: types.ModuleType, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setenv("ROBOCO_PROJECT_SLUG", "roboco-api")
|
||||
with patch("httpx.Client") as client_cls:
|
||||
client = client_cls.return_value.__enter__.return_value
|
||||
client.get.return_value = _mock_response({"branch": "main"})
|
||||
git_module.roboco_git_status()
|
||||
_, kwargs = client.get.call_args
|
||||
assert kwargs["params"]["project_slug"] == "roboco-api"
|
||||
|
||||
|
||||
def test_git_status_missing_both_returns_error_without_http_call(
|
||||
git_module: types.ModuleType,
|
||||
) -> None:
|
||||
with patch("httpx.Client") as client_cls:
|
||||
result = git_module.roboco_git_status()
|
||||
client_cls.assert_not_called()
|
||||
assert result["error"] == "missing_project_slug"
|
||||
@@ -0,0 +1,86 @@
|
||||
"""Per-index-type chunk-length floor — journals/learnings are short by
|
||||
design (templated notes, distilled lessons) and were always discarded by
|
||||
the global 200-char garbage filter ("All chunks filtered as garbage",
|
||||
raw_count=1 on every journal write). Journals get a 40-char floor,
|
||||
learnings an 80-char floor; every other index keeps the 200-char default.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from roboco.models.optimal import IndexType
|
||||
from roboco.services.optimal_brain.indexes.base import (
|
||||
_MIN_CHUNK_LENGTH,
|
||||
IndexConfig,
|
||||
_filter_quality_chunks,
|
||||
)
|
||||
from roboco.services.optimal_brain.text_chunker import Chunk
|
||||
|
||||
# Named constants — ruff PLR2004 forbids magic-number comparisons.
|
||||
_JOURNALS_FLOOR = 40
|
||||
_LEARNINGS_FLOOR = 80
|
||||
_DEFAULT_FLOOR = _MIN_CHUNK_LENGTH # 200
|
||||
|
||||
|
||||
def _chunk(text: str) -> Chunk:
|
||||
return Chunk(text=text, source="roboco://journals/entry-1")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# IndexConfig.from_settings — per-index-type floor
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_journals_config_lowers_floor_to_40() -> None:
|
||||
assert IndexConfig.from_settings(IndexType.JOURNALS).min_chunk_length == (
|
||||
_JOURNALS_FLOOR
|
||||
)
|
||||
|
||||
|
||||
def test_learnings_config_lowers_floor_to_80() -> None:
|
||||
assert IndexConfig.from_settings(IndexType.LEARNINGS).min_chunk_length == (
|
||||
_LEARNINGS_FLOOR
|
||||
)
|
||||
|
||||
|
||||
def test_other_index_keeps_default_floor() -> None:
|
||||
assert IndexConfig.from_settings(IndexType.DOCUMENTATION).min_chunk_length == (
|
||||
_DEFAULT_FLOOR
|
||||
)
|
||||
assert IndexConfig.from_settings(IndexType.CODE).min_chunk_length == _DEFAULT_FLOOR
|
||||
assert IndexConfig(persist_dir="/tmp/idx").min_chunk_length == _DEFAULT_FLOOR
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _filter_quality_chunks — the actual gate
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_120_char_journal_chunk_passes_journals_floor() -> None:
|
||||
text = (
|
||||
"Reflected on task rate-limiter-fix: fixed the off-by-one boundary "
|
||||
"check in the 429 threshold and added a regression test."
|
||||
)
|
||||
assert len(text) >= _JOURNALS_FLOOR * 3
|
||||
kept = _filter_quality_chunks([_chunk(text)], min_chunk_length=_JOURNALS_FLOOR)
|
||||
assert len(kept) == 1
|
||||
|
||||
|
||||
def test_150_char_chunk_still_fails_default_200_floor() -> None:
|
||||
text = "x" * (_DEFAULT_FLOOR - 50)
|
||||
kept = _filter_quality_chunks([_chunk(text)], min_chunk_length=_DEFAULT_FLOOR)
|
||||
assert kept == []
|
||||
|
||||
|
||||
def test_default_index_behavior_unchanged_no_arg() -> None:
|
||||
"""Omitting min_chunk_length keeps the historical 200-char behavior."""
|
||||
short = _chunk("x" * (_DEFAULT_FLOOR - 50))
|
||||
long_enough = _chunk("x" * (_DEFAULT_FLOOR + 50))
|
||||
assert _filter_quality_chunks([short]) == []
|
||||
assert _filter_quality_chunks([long_enough]) == [long_enough]
|
||||
|
||||
|
||||
def test_markdown_only_chunk_still_filtered_at_lowered_floor() -> None:
|
||||
"""A 40-char chunk that's mostly ``` / --- / # formatting is still junk."""
|
||||
text = "#" * (_JOURNALS_FLOOR // 2) + "-" * (_JOURNALS_FLOOR // 2)
|
||||
kept = _filter_quality_chunks([_chunk(text)], min_chunk_length=_JOURNALS_FLOOR)
|
||||
assert kept == []
|
||||
@@ -0,0 +1,505 @@
|
||||
"""DB-backed tests for the journals/learnings zero-chunk RAG backfill.
|
||||
|
||||
Before the per-index chunk-floor fix, ``ingest()`` returned success with
|
||||
``chunk_count=0`` for undersized content, so historical journal entries and
|
||||
learnings were durably recorded in ``journal_entries`` but never landed a row
|
||||
in the vector store. These tests exercise the REAL SQL against a real
|
||||
Postgres (`db_session`, see tests/conftest.py) — the JOIN/floor/ANY(...)
|
||||
logic can't be verified with a mocked session. ``optimal`` itself is mocked
|
||||
(index_journal_entry / record_learning) since the embedder is out of scope.
|
||||
|
||||
Run with: ROBOCO_TEST_DB_PORT=55432 ROBOCO_TEST_DB_USER=renzof pytest ...
|
||||
|
||||
``db_session`` rolls back at the end of each test, so nothing THIS file seeds
|
||||
leaks between its own tests. But the DB itself is session-scoped for the
|
||||
whole pytest run, and other test files commit real journal/learning rows
|
||||
against it (e.g. via gateway note-writing tests) — a full-suite / CI run can
|
||||
start this file's tests with a non-empty ``journal_entries`` table. Every
|
||||
assertion below is therefore scoped to the specific rows a test creates
|
||||
(by entry_id or by a uniquified content string), never to a global
|
||||
processed/remaining count — except the cap test, which measures the
|
||||
pre-existing "stray" candidate count first and sizes the cap around it.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from roboco.db.tables import AgentTable, JournalEntryTable, JournalTable
|
||||
from roboco.models.base import AgentRole, AgentStatus, JournalEntryType, Team
|
||||
from roboco.services import rag_index_failures as rif
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from roboco.models.optimal import IndexJournalEntryParams
|
||||
from roboco.services.optimal_brain.indexes.learnings import RecordLearningParams
|
||||
|
||||
# Journals floor is 40, learnings floor is 80 (see optimal_brain/indexes/base.py).
|
||||
_LONG_CONTENT = "x" * 90 # clears both floors
|
||||
_MID_CONTENT = "y" * 50 # clears JOURNALS (40) but not LEARNINGS (80)
|
||||
_SHORT_CONTENT = "short" # clears neither floor
|
||||
|
||||
_TEST_CAP = 2 # how many of THIS test's own rows the cap test expects to fit
|
||||
|
||||
|
||||
def _unique_content(base: str) -> str:
|
||||
"""A floor-clearing content string that can't collide with another
|
||||
test's stray committed row, so hash/content-keyed presence checks stay
|
||||
deterministic under DB pollution."""
|
||||
return f"{base}-{uuid4().hex[:8]}"
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="session", loop_scope="session", autouse=True)
|
||||
async def _chunks_tables(_test_database_url: str) -> None:
|
||||
"""Create minimal chunks_journals/chunks_learnings tables once.
|
||||
|
||||
VectorStore normally provisions these (id, content, source, embedding,
|
||||
metadata, tsv); the backfill queries only touch ``source``, so a minimal
|
||||
shape is enough. Created via a throwaway engine so the DDL commits
|
||||
outside any per-test rolled-back transaction.
|
||||
"""
|
||||
engine = create_async_engine(_test_database_url, future=True)
|
||||
try:
|
||||
async with engine.begin() as conn:
|
||||
await conn.execute(
|
||||
text(
|
||||
"CREATE TABLE IF NOT EXISTS chunks_journals "
|
||||
"(id serial primary key, source text not null)"
|
||||
)
|
||||
)
|
||||
await conn.execute(
|
||||
text(
|
||||
"CREATE TABLE IF NOT EXISTS chunks_learnings "
|
||||
"(id serial primary key, source text not null)"
|
||||
)
|
||||
)
|
||||
finally:
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _redirect_db_context(
|
||||
monkeypatch: pytest.MonkeyPatch, db_session: AsyncSession
|
||||
) -> None:
|
||||
"""Make get_db_context() (used inside the backfill functions) reuse the
|
||||
test's own session/transaction, so uncommitted seed rows are visible
|
||||
(read-your-own-writes) without needing an explicit commit + rollback of
|
||||
the whole test DB's shared tables between tests."""
|
||||
|
||||
class _Ctx:
|
||||
async def __aenter__(self) -> AsyncSession:
|
||||
return db_session
|
||||
|
||||
async def __aexit__(self, *_a: object) -> None:
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(rif, "get_db_context", _Ctx)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def _journal(db_session: AsyncSession) -> UUID:
|
||||
"""Seed one agent + its journal; returns the journal id."""
|
||||
agent = AgentTable(
|
||||
id=uuid4(),
|
||||
name="Backfill Test Agent",
|
||||
slug=f"backfill-{uuid4().hex[:8]}",
|
||||
role=AgentRole.DEVELOPER,
|
||||
team=Team.BACKEND,
|
||||
status=AgentStatus.ACTIVE,
|
||||
model_config={},
|
||||
system_prompt="x",
|
||||
capabilities=[],
|
||||
permissions={},
|
||||
metrics={},
|
||||
)
|
||||
db_session.add(agent)
|
||||
await db_session.flush()
|
||||
journal = JournalTable(id=uuid4(), agent_id=agent.id)
|
||||
db_session.add(journal)
|
||||
await db_session.flush()
|
||||
return UUID(str(journal.id))
|
||||
|
||||
|
||||
async def _seed_entry(
|
||||
db_session: AsyncSession,
|
||||
journal_id: UUID,
|
||||
content: str,
|
||||
*,
|
||||
entry_type: JournalEntryType = JournalEntryType.GENERAL,
|
||||
is_private: bool = False,
|
||||
) -> UUID:
|
||||
entry = JournalEntryTable(
|
||||
id=uuid4(),
|
||||
journal_id=journal_id,
|
||||
type=entry_type,
|
||||
title="t",
|
||||
content=content,
|
||||
is_private=is_private,
|
||||
tags=[],
|
||||
)
|
||||
db_session.add(entry)
|
||||
await db_session.flush()
|
||||
return UUID(str(entry.id))
|
||||
|
||||
|
||||
async def _insert_chunk(db_session: AsyncSession, table: str, source: str) -> None:
|
||||
await db_session.execute(
|
||||
text(f"INSERT INTO {table} (source) VALUES (:s)"), {"s": source}
|
||||
)
|
||||
|
||||
|
||||
def _optimal(**overrides: Any) -> MagicMock:
|
||||
optimal = MagicMock()
|
||||
optimal.is_index_registered = MagicMock(return_value=True)
|
||||
optimal.index_journal_entry = AsyncMock()
|
||||
optimal.record_learning = AsyncMock()
|
||||
for key, value in overrides.items():
|
||||
setattr(optimal, key, value)
|
||||
return optimal
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _backfill_journals
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backfill_journals_selects_only_zero_chunk_entries(
|
||||
db_session: AsyncSession, _journal: UUID
|
||||
) -> None:
|
||||
"""An entry with an existing chunks_journals row is left alone; one with
|
||||
none is re-ingested. Scoped to these two entry_ids since a polluted DB
|
||||
may carry other zero-chunk stray rows the pass also processes."""
|
||||
missing_id = await _seed_entry(db_session, _journal, _LONG_CONTENT)
|
||||
present_id = await _seed_entry(db_session, _journal, _LONG_CONTENT)
|
||||
await _insert_chunk(
|
||||
db_session, "chunks_journals", f"roboco://journals/{present_id}"
|
||||
)
|
||||
|
||||
optimal = _optimal()
|
||||
await rif._backfill_journals(optimal)
|
||||
|
||||
called_ids = {
|
||||
c.args[0].entry_id for c in optimal.index_journal_entry.await_args_list
|
||||
}
|
||||
assert missing_id in called_ids
|
||||
assert present_id not in called_ids
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backfill_journals_excludes_private_entries(
|
||||
db_session: AsyncSession, _journal: UUID
|
||||
) -> None:
|
||||
"""A private entry is never a candidate — it's deliberately excluded
|
||||
from the shared JOURNALS corpus, not a pending backlog item."""
|
||||
private_id = await _seed_entry(db_session, _journal, _LONG_CONTENT, is_private=True)
|
||||
|
||||
optimal = _optimal()
|
||||
await rif._backfill_journals(optimal)
|
||||
|
||||
called_ids = {
|
||||
c.args[0].entry_id for c in optimal.index_journal_entry.await_args_list
|
||||
}
|
||||
assert private_id not in called_ids
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backfill_journals_excludes_sub_floor_entries(
|
||||
db_session: AsyncSession, _journal: UUID
|
||||
) -> None:
|
||||
"""Content still under the JOURNALS floor would zero-chunk again — the
|
||||
SELECT excludes it so it is never retried forever."""
|
||||
short_id = await _seed_entry(db_session, _journal, _SHORT_CONTENT)
|
||||
|
||||
optimal = _optimal()
|
||||
await rif._backfill_journals(optimal)
|
||||
|
||||
called_ids = {
|
||||
c.args[0].entry_id for c in optimal.index_journal_entry.await_args_list
|
||||
}
|
||||
assert short_id not in called_ids
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backfill_journals_respects_cap(
|
||||
db_session: AsyncSession, _journal: UUID, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""The per-boot cap bounds how many rows a single pass fetches.
|
||||
|
||||
A polluted DB's stray zero-chunk rows sort first (earliest created_at)
|
||||
and spend part of the cap before this test's own rows do. So: measure
|
||||
the stray count with an unbounded pass first (mocked — nothing is
|
||||
actually written to chunks_journals), then size the cap to exactly
|
||||
`strays + _TEST_CAP` and seed one row more than that budget covers.
|
||||
Only _TEST_CAP of this test's own rows can then fit.
|
||||
"""
|
||||
monkeypatch.setattr(rif, "_BACKFILL_CAP", 1_000_000)
|
||||
strays, _ = await rif._backfill_journals(_optimal())
|
||||
|
||||
own_ids = [
|
||||
await _seed_entry(db_session, _journal, _LONG_CONTENT)
|
||||
for _ in range(_TEST_CAP + 1)
|
||||
]
|
||||
|
||||
monkeypatch.setattr(rif, "_BACKFILL_CAP", strays + _TEST_CAP)
|
||||
optimal = _optimal()
|
||||
processed, _remaining = await rif._backfill_journals(optimal)
|
||||
|
||||
assert processed == strays + _TEST_CAP
|
||||
called_ids = {
|
||||
c.args[0].entry_id for c in optimal.index_journal_entry.await_args_list
|
||||
}
|
||||
assert len(called_ids & set(own_ids)) == _TEST_CAP
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backfill_journals_tolerates_failing_row(
|
||||
db_session: AsyncSession, _journal: UUID
|
||||
) -> None:
|
||||
"""One row's re-index failure never aborts the rest of the pass — the
|
||||
fake ingest rejects THIS test's own fail_id specifically (by entry_id),
|
||||
so the proof holds regardless of how many stray rows are also in play."""
|
||||
fail_id = await _seed_entry(db_session, _journal, _LONG_CONTENT)
|
||||
ok_id = await _seed_entry(db_session, _journal, _LONG_CONTENT)
|
||||
succeeded: set[UUID] = set()
|
||||
|
||||
async def _index(params: IndexJournalEntryParams) -> None:
|
||||
if params.entry_id == fail_id:
|
||||
raise RuntimeError("ollama down")
|
||||
succeeded.add(params.entry_id)
|
||||
|
||||
optimal = _optimal(index_journal_entry=AsyncMock(side_effect=_index))
|
||||
await rif._backfill_journals(optimal)
|
||||
|
||||
called_ids = {
|
||||
c.args[0].entry_id for c in optimal.index_journal_entry.await_args_list
|
||||
}
|
||||
assert fail_id in called_ids
|
||||
assert ok_id in succeeded
|
||||
assert fail_id not in succeeded
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backfill_journals_noop_when_index_not_registered(
|
||||
db_session: AsyncSession, _journal: UUID
|
||||
) -> None:
|
||||
"""A JOURNALS plugin that never initialized is skipped, not errored."""
|
||||
await _seed_entry(db_session, _journal, _LONG_CONTENT)
|
||||
optimal = _optimal(is_index_registered=MagicMock(return_value=False))
|
||||
|
||||
processed, remaining = await rif._backfill_journals(optimal)
|
||||
|
||||
assert (processed, remaining) == (0, 0)
|
||||
optimal.index_journal_entry.assert_not_awaited()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _backfill_learnings
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backfill_learnings_selects_only_missing_from_chunks_learnings(
|
||||
db_session: AsyncSession, _journal: UUID
|
||||
) -> None:
|
||||
"""A LEARNING entry already present in chunks_journals (lower floor) can
|
||||
still be missing from chunks_learnings (higher floor) — the two passes
|
||||
check presence independently. Content is uniquified so this test's own
|
||||
row can't be shadowed by a stray row of identical content."""
|
||||
content = _unique_content(_LONG_CONTENT)
|
||||
entry_id = await _seed_entry(
|
||||
db_session, _journal, content, entry_type=JournalEntryType.LEARNING
|
||||
)
|
||||
# Already indexed into JOURNALS — irrelevant to the LEARNINGS check.
|
||||
await _insert_chunk(db_session, "chunks_journals", f"roboco://journals/{entry_id}")
|
||||
|
||||
optimal = _optimal()
|
||||
await rif._backfill_learnings(optimal)
|
||||
|
||||
calls_by_content = {
|
||||
c.args[0].content: c.args[0] for c in optimal.record_learning.await_args_list
|
||||
}
|
||||
assert content in calls_by_content
|
||||
assert calls_by_content[content].shareable is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backfill_learnings_skips_already_present(
|
||||
db_session: AsyncSession, _journal: UUID
|
||||
) -> None:
|
||||
"""A learning whose hashed source already has a chunks_learnings row is
|
||||
left alone."""
|
||||
content = _unique_content(_LONG_CONTENT)
|
||||
await _seed_entry(
|
||||
db_session, _journal, content, entry_type=JournalEntryType.LEARNING
|
||||
)
|
||||
await _insert_chunk(db_session, "chunks_learnings", rif._learning_source(content))
|
||||
|
||||
optimal = _optimal()
|
||||
await rif._backfill_learnings(optimal)
|
||||
|
||||
called_contents = {
|
||||
c.args[0].content for c in optimal.record_learning.await_args_list
|
||||
}
|
||||
assert content not in called_contents
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backfill_learnings_excludes_sub_floor_entries(
|
||||
db_session: AsyncSession, _journal: UUID
|
||||
) -> None:
|
||||
"""Content between the JOURNALS floor and the (higher) LEARNINGS floor
|
||||
would zero-chunk again in LEARNINGS — excluded so it's never retried
|
||||
forever. The length-based SQL filter excludes it structurally, so this
|
||||
holds regardless of any other candidate rows in play."""
|
||||
await _seed_entry(
|
||||
db_session, _journal, _MID_CONTENT, entry_type=JournalEntryType.LEARNING
|
||||
)
|
||||
|
||||
optimal = _optimal()
|
||||
await rif._backfill_learnings(optimal)
|
||||
|
||||
called_contents = {
|
||||
c.args[0].content for c in optimal.record_learning.await_args_list
|
||||
}
|
||||
assert _MID_CONTENT not in called_contents
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backfill_learnings_shareable_reflects_privacy(
|
||||
db_session: AsyncSession, _journal: UUID
|
||||
) -> None:
|
||||
"""A private learning is still recorded, just non-shareable — mirrors
|
||||
the live _schedule_rag_index path."""
|
||||
content = _unique_content(_LONG_CONTENT)
|
||||
await _seed_entry(
|
||||
db_session,
|
||||
_journal,
|
||||
content,
|
||||
entry_type=JournalEntryType.LEARNING,
|
||||
is_private=True,
|
||||
)
|
||||
|
||||
optimal = _optimal()
|
||||
await rif._backfill_learnings(optimal)
|
||||
|
||||
calls_by_content = {
|
||||
c.args[0].content: c.args[0] for c in optimal.record_learning.await_args_list
|
||||
}
|
||||
assert content in calls_by_content
|
||||
assert calls_by_content[content].shareable is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backfill_learnings_tolerates_failing_row(
|
||||
db_session: AsyncSession, _journal: UUID
|
||||
) -> None:
|
||||
"""One row's re-index failure never aborts the rest of the pass — the
|
||||
fake ingest rejects THIS test's own fail_content specifically (by
|
||||
content), so the proof holds regardless of stray rows also in play."""
|
||||
fail_content = _unique_content(f"{_LONG_CONTENT}-fail")
|
||||
ok_content = _unique_content(f"{_LONG_CONTENT}-ok")
|
||||
await _seed_entry(
|
||||
db_session, _journal, fail_content, entry_type=JournalEntryType.LEARNING
|
||||
)
|
||||
await _seed_entry(
|
||||
db_session, _journal, ok_content, entry_type=JournalEntryType.LEARNING
|
||||
)
|
||||
succeeded: set[str] = set()
|
||||
|
||||
async def _record(params: RecordLearningParams) -> None:
|
||||
if params.content == fail_content:
|
||||
raise RuntimeError("ollama down")
|
||||
succeeded.add(params.content)
|
||||
|
||||
optimal = _optimal(record_learning=AsyncMock(side_effect=_record))
|
||||
await rif._backfill_learnings(optimal)
|
||||
|
||||
called_contents = {
|
||||
c.args[0].content for c in optimal.record_learning.await_args_list
|
||||
}
|
||||
assert fail_content in called_contents
|
||||
assert ok_content in succeeded
|
||||
assert fail_content not in succeeded
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backfill_learnings_noop_when_index_not_registered(
|
||||
db_session: AsyncSession, _journal: UUID
|
||||
) -> None:
|
||||
"""A LEARNINGS plugin that never initialized is skipped, not errored."""
|
||||
await _seed_entry(
|
||||
db_session, _journal, _LONG_CONTENT, entry_type=JournalEntryType.LEARNING
|
||||
)
|
||||
optimal = _optimal(is_index_registered=MagicMock(return_value=False))
|
||||
|
||||
processed, remaining = await rif._backfill_learnings(optimal)
|
||||
|
||||
assert (processed, remaining) == (0, 0)
|
||||
optimal.record_learning.assert_not_awaited()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _learning_source — must match LearningsIndexPlugin.record_learning exactly,
|
||||
# or the presence check can never find what the live path indexed.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_learning_source_matches_plugin_hash() -> None:
|
||||
content = "a distilled lesson"
|
||||
expected_hash = hashlib.md5(content.encode(), usedforsecurity=False).hexdigest()[
|
||||
:16
|
||||
]
|
||||
assert rif._learning_source(content) == f"roboco://learnings/lrn-{expected_hash}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# backfill_unindexed_journals — orchestration + isolation between passes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backfill_unindexed_journals_isolates_pass_failures(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""A hard failure in one pass (e.g. lost DB connection) doesn't prevent
|
||||
the other pass from running."""
|
||||
monkeypatch.setattr(
|
||||
rif, "_backfill_journals", AsyncMock(side_effect=RuntimeError("db down"))
|
||||
)
|
||||
learnings_mock = AsyncMock(return_value=(3, 1))
|
||||
monkeypatch.setattr(rif, "_backfill_learnings", learnings_mock)
|
||||
|
||||
result = await rif.backfill_unindexed_journals(MagicMock())
|
||||
|
||||
assert result == {
|
||||
"journals_processed": 0,
|
||||
"journals_remaining": 0,
|
||||
"learnings_processed": 3,
|
||||
"learnings_remaining": 1,
|
||||
}
|
||||
learnings_mock.assert_awaited_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backfill_unindexed_journals_returns_combined_counts(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Happy path: both passes run and their counts are combined."""
|
||||
monkeypatch.setattr(rif, "_backfill_journals", AsyncMock(return_value=(2, 0)))
|
||||
monkeypatch.setattr(rif, "_backfill_learnings", AsyncMock(return_value=(1, 0)))
|
||||
|
||||
result = await rif.backfill_unindexed_journals(MagicMock())
|
||||
|
||||
assert result == {
|
||||
"journals_processed": 2,
|
||||
"journals_remaining": 0,
|
||||
"learnings_processed": 1,
|
||||
"learnings_remaining": 0,
|
||||
}
|
||||
Reference in New Issue
Block a user