diff --git a/CHANGELOG.md b/CHANGELOG.md index 33192f6..6ee5d4a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,20 +26,63 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [1.1.4] - 2026-07-20 -### Added - -- **Langfuse integration example** — added an OpenTelemetry-based wrapper for - tracing EverOS add, flush/extract, search, and reflection operations, with a - built-in mock and support for connecting to a real EverOS server. +> The entries below reflect the code shipped as `everos==1.1.4` on PyPI. +> The 1.1.4 sdist was built from the internal release lane and contains +> fixes that were not represented in this file when 1.1.4 was tagged; +> this section restores them so the changelog matches the wheel. ### Fixed -- **Cascade delete/modify race** — when a file disappears after its modified - event is queued, the worker now processes it as a deletion instead of leaving - a stale indexed row and permanently failed queue item. -- **Langfuse live-server traces use only real telemetry** — synthetic child - spans are now limited to responses that provide stage details, while real - servers emit accurate top-level latency, output, and recall-quality scores. +- **Knowledge upload path traversal (CWE-22)** — the original-file write + path is now contained to the document directory; adversarial filenames + (`..`, absolute paths, symlink games) are rejected on + `POST /api/v1/knowledge/documents` and `POST /api/v2/knowledge/documents`. +- **Cascade reliability — retry classification, budget, and races** — the + worker catches `ExternalServiceError` (embedding / LLM / rerank transient + failures) and retries inline up to 3 times before marking + `retryable=True`; a total retry budget of 12 attempts across scanner + cycles bounds retries on prolonged outages. The reconciler no longer + re-enqueues `pending` / `processing` rows on stable mtime (previously + overwrote the worker's `mark_done`); `failed` rows with + `retryable=False` on stable mtime skip auto-retry so users can edit and + re-save. SQLite `REAL` float precision loss in mtime comparisons is + now absorbed via a 10 ms tolerance. LanceDB `optimize()` failures + escalate to a `drop_index + create_index` rebuild after 5 consecutive + misses (workaround for `lance-format/lance#7653` panic path). +- **Cascade delete/modify race** — when a file disappears after its + modified event is queued, the worker processes it as a deletion + instead of leaving a stale indexed row and permanently failed queue + item. +- **Embedding provider raises on empty API data** — the provider now + raises `EmbeddingServiceError` when the API returns HTTP 200 with an + empty `data` array (previously silently returned zero-length vectors, + corrupting search). +- **Episode extraction retries on malformed LLM output** — the `/flush` + synchronous path retries everalgo `ValueError` (typically OpenRouter + truncated responses) twice with 1 s / 2 s backoff before surfacing a + 500. +- **Langfuse live-server traces use only real telemetry** — synthetic + child spans are now limited to responses that provide stage details, + while real servers emit accurate top-level latency, output, and + recall-quality scores. + +### Added + +- **Optional `dimensions` parameter for MRL-capable embedding models** — + opt-in via `[embedding] dimensions = N` in `everos.toml`; forwarded to + the API for server-side truncation with re-normalization (OpenAI + text-embedding-3-\*, Qwen3-Embedding). +- **LLM `finish_reason` diagnostic warnings** — logs `content_len` / + `content_tail` / `model` when the provider returns a non-`stop` + `finish_reason`, aiding OpenRouter truncation triage. +- **Langfuse integration example** — added an OpenTelemetry-based + wrapper for tracing EverOS add, flush/extract, search, and reflection + operations, with a built-in mock and support for connecting to a real + EverOS server. + +### Changed + +- **`everalgo-user-memory` bumped 0.3.1 → 0.3.2**. ## [1.1.3] - 2026-07-10 diff --git a/pyproject.toml b/pyproject.toml index 398cf91..10b88fa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -62,7 +62,7 @@ dependencies = [ "anyio>=4.0", # Async file I/O (anyio.Path, to_thread.run_sync) for the markdown layer # Algorithm library (everalgo monorepo, published on PyPI). - "everalgo-user-memory==0.3.1", + "everalgo-user-memory==0.3.2", "everalgo-agent-memory==0.3.1", "everalgo-rank==0.4.1", "everalgo-knowledge==0.1.1", diff --git a/src/everos/component/embedding/factory.py b/src/everos/component/embedding/factory.py index 4f63493..f32e744 100644 --- a/src/everos/component/embedding/factory.py +++ b/src/everos/component/embedding/factory.py @@ -49,6 +49,7 @@ def build_embedding_provider( api_key=settings.api_key.get_secret_value(), base_url=settings.base_url, dim=dim, + dimensions=settings.dimensions, timeout=settings.timeout_seconds, max_retries=settings.max_retries, batch_size=settings.batch_size, diff --git a/src/everos/component/embedding/openai_provider.py b/src/everos/component/embedding/openai_provider.py index b26b89e..b556b94 100644 --- a/src/everos/component/embedding/openai_provider.py +++ b/src/everos/component/embedding/openai_provider.py @@ -2,10 +2,11 @@ Wraps :class:`openai.AsyncOpenAI` so any OpenAI-protocol endpoint (DeepInfra, OpenAI, Together, Fireworks, …) works without per-provider -forks. Self-hosted vLLM also exposes the same shape; the only quirk it -imposes is that the ``dimensions`` request parameter is ignored — we -truncate client-side to ``dim`` so callers always see the declared -shape regardless of backend. +forks. When ``dimensions`` is set, the parameter is forwarded to the +API so MRL-capable models (OpenAI text-embedding-3-*, Qwen3-Embedding, +…) can do server-side truncation with proper re-normalization. +Client-side truncation to ``dim`` is always applied as a fallback for +backends that ignore or don't support the parameter. Concurrency model: @@ -37,8 +38,11 @@ class OpenAIEmbeddingProvider: base_url: OpenAI-protocol endpoint (e.g. ``"https://api.deepinfra.com/v1/openai"``). dim: Target vector dimension. Vectors longer than this are - truncated client-side (matches the LanceDB column shape — - see ``17_lancedb_tables_design.md``). + truncated client-side (matches the LanceDB column shape). + dimensions: Optional API-level ``dimensions`` parameter for + MRL-capable models. When set, forwarded to the embedding + API for server-side truncation with re-normalization. + When ``None``, the parameter is omitted from the request. timeout: Per-request timeout, seconds. max_retries: Retry budget exposed via the openai SDK. batch_size: How many inputs per ``/embeddings`` call. @@ -52,12 +56,14 @@ class OpenAIEmbeddingProvider: api_key: str, base_url: str, dim: int = 1024, + dimensions: int | None = None, timeout: float = 30.0, max_retries: int = 3, batch_size: int = 10, max_concurrent: int = 5, ) -> None: self.dim = dim + self._dimensions = dimensions self._model = model self._batch_size = batch_size self._semaphore = asyncio.Semaphore(max_concurrent) @@ -100,14 +106,21 @@ class OpenAIEmbeddingProvider: response = await self._client.embeddings.create( model=self._model, input=chunk, + dimensions=self._dimensions + if self._dimensions is not None + else openai.NOT_GIVEN, ) except openai.OpenAIError as exc: raise EmbeddingServiceError(str(exc)) from exc + if not response.data: + raise EmbeddingServiceError( + f"Embedding API returned empty data for {len(chunk)} inputs" + ) # Embeddings report only input (prompt) tokens. usage = getattr(response, "usage", None) set_generation_usage( model=self._model, input_tokens=usage.prompt_tokens if usage else None, ) - # OpenAI returns ``data`` indexed by request order; truncate to ``dim``. + # Client-side truncation — always applied as fallback. return [list(item.embedding[: self.dim]) for item in response.data] diff --git a/src/everos/component/llm/client.py b/src/everos/component/llm/client.py index 5b0de8c..dfcbcaf 100644 --- a/src/everos/component/llm/client.py +++ b/src/everos/component/llm/client.py @@ -9,9 +9,13 @@ provider) instead of silently failing per-request downstream. from __future__ import annotations +from typing import Any + from everalgo.llm import build_client from everalgo.llm.config import LLMConfig from everalgo.llm.protocols import LLMClient +from everalgo.llm.types import ChatMessage, ChatResponse +from pydantic import BaseModel from everos.config import load_settings from everos.core.observability.logging import get_logger @@ -21,6 +25,48 @@ from ._usage_client import UsageRecordingClient logger = get_logger(__name__) +class _LoggingLLMClient: + """Wrapper that logs non-stop ``finish_reason`` for diagnostics. + + Always active — cost is one branch per chat() call. OpenRouter and + a few compatible providers occasionally return HTTP 200 with a + truncated body and ``finish_reason != "stop"`` (length cap, filter + trigger). Recording the reason plus the tail of the content lets + us triage those without needing a repro from the caller. + """ + + def __init__(self, inner: LLMClient) -> None: + self._inner = inner + + async def chat( + self, + messages: list[ChatMessage], + *, + model: str | None = None, + temperature: float | None = None, + max_tokens: int | None = None, + response_format: type[BaseModel] | None = None, + **extra: Any, + ) -> ChatResponse: + resp = await self._inner.chat( + messages, + model=model, + temperature=temperature, + max_tokens=max_tokens, + response_format=response_format, + **extra, + ) + if resp.finish_reason and resp.finish_reason != "stop": + logger.warning( + "llm_non_stop_finish", + finish_reason=resp.finish_reason, + content_len=len(resp.content), + content_tail=resp.content[-200:] if resp.content else "", + model=resp.model, + ) + return resp + + class LLMNotConfiguredError(RuntimeError): """Raised when ``settings.llm`` is missing ``api_key`` or ``base_url``.""" @@ -60,7 +106,11 @@ def get_llm_client() -> LLMClient: # disabled path (the default) allocation- and overhead-free. if settings.observability.enabled: client = UsageRecordingClient(client) - _llm_client = client + # Finish-reason diagnostic wrapper is always outermost: it must see + # the response even when tracing is off, and it must observe the + # exact reason the underlying provider reported (not one synthesised + # by an inner wrapper). + _llm_client = _LoggingLLMClient(client) logger.info("llm_client_built", model=llm_cfg.model) return _llm_client diff --git a/src/everos/config/settings.py b/src/everos/config/settings.py index 354d6c5..c230916 100644 --- a/src/everos/config/settings.py +++ b/src/everos/config/settings.py @@ -176,6 +176,7 @@ class EmbeddingSettings(BaseModel): EVEROS_EMBEDDING__MODEL EVEROS_EMBEDDING__API_KEY EVEROS_EMBEDDING__BASE_URL + EVEROS_EMBEDDING__DIMENSIONS EVEROS_EMBEDDING__TIMEOUT_SECONDS EVEROS_EMBEDDING__MAX_RETRIES EVEROS_EMBEDDING__BATCH_SIZE @@ -185,6 +186,12 @@ class EmbeddingSettings(BaseModel): model: str | None = None api_key: SecretStr | None = None base_url: str | None = None + dimensions: int | None = None + """API-level ``dimensions`` parameter for MRL-capable models. + When set, passed to the embedding API so the server truncates + with proper re-normalization. When ``None`` (default), the + parameter is omitted and client-side truncation to ``dim`` + handles dimension alignment.""" timeout_seconds: float = Field(default=30.0, gt=0) max_retries: int = Field(default=3, ge=0) batch_size: int = Field(default=10, ge=1) diff --git a/src/everos/infra/persistence/sqlite/__init__.py b/src/everos/infra/persistence/sqlite/__init__.py index bf243fc..57ecb6b 100644 --- a/src/everos/infra/persistence/sqlite/__init__.py +++ b/src/everos/infra/persistence/sqlite/__init__.py @@ -31,6 +31,7 @@ not need to manage either. # ``SQLModel.metadata`` so ``SqliteLifespanProvider.startup`` can # ``create_all`` without callers having to import each model module. from . import tables as tables +from .repos import MTIME_TOLERANCE_SECONDS as MTIME_TOLERANCE_SECONDS from .repos import DocumentListPage as DocumentListPage from .repos import DocumentUpsertPayload as DocumentUpsertPayload from .repos import QueueSummary as QueueSummary @@ -58,6 +59,7 @@ from .tables import ReflectionReport as ReflectionReport from .tables import UnprocessedBuffer as UnprocessedBuffer __all__ = [ + "MTIME_TOLERANCE_SECONDS", "Cluster", "ClusterMember", "ConversationStatus", diff --git a/src/everos/infra/persistence/sqlite/repos/__init__.py b/src/everos/infra/persistence/sqlite/repos/__init__.py index c4f485b..9a656fd 100644 --- a/src/everos/infra/persistence/sqlite/repos/__init__.py +++ b/src/everos/infra/persistence/sqlite/repos/__init__.py @@ -12,6 +12,7 @@ from .knowledge import DocumentUpsertPayload as DocumentUpsertPayload from .knowledge import TopicUpsertPayload as TopicUpsertPayload from .knowledge import knowledge_document_repo as knowledge_document_repo from .knowledge import knowledge_topic_sqlite_repo as knowledge_topic_sqlite_repo +from .md_change_state import MTIME_TOLERANCE_SECONDS as MTIME_TOLERANCE_SECONDS from .md_change_state import QueueSummary as QueueSummary from .md_change_state import md_change_state_repo as md_change_state_repo from .memcell import memcell_repo as memcell_repo @@ -19,6 +20,7 @@ from .reflection_report import reflection_report_repo as reflection_report_repo from .unprocessed_buffer import unprocessed_buffer_repo as unprocessed_buffer_repo __all__ = [ + "MTIME_TOLERANCE_SECONDS", "DocumentListPage", "DocumentUpsertPayload", "QueueSummary", diff --git a/src/everos/infra/persistence/sqlite/repos/md_change_state.py b/src/everos/infra/persistence/sqlite/repos/md_change_state.py index b008292..f1ac06b 100644 --- a/src/everos/infra/persistence/sqlite/repos/md_change_state.py +++ b/src/everos/infra/persistence/sqlite/repos/md_change_state.py @@ -29,7 +29,7 @@ from __future__ import annotations import dataclasses -from sqlalchemy import func, select, update +from sqlalchemy import func, select, text, update from sqlalchemy.dialects.sqlite import insert as sqlite_insert from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker @@ -39,6 +39,20 @@ from everos.core.persistence.sqlite import RepoBase, session_scope from ..sqlite_manager import get_session_factory from ..tables import MdChangeState +MTIME_TOLERANCE_SECONDS = 0.01 +"""Absolute mtime delta (seconds) treated as "unchanged". + +Used both by the upsert's retry_count carry-over CASE (below) and by the +cascade reconciler's stable-mtime check +(:func:`everos.memory.cascade.reconciler.reconcile`). The two comparisons +share this single source of truth so their notions of "same file" stay +in lock-step; drift would leave upsert preserving ``retry_count`` while +the reconciler skips re-enqueue (or vice versa), corrupting the retry +budget. + +Sized to absorb SQLite ``REAL`` round-trip loss (~5 µs) so a stable +mtime does not oscillate the reconcile decision every scanner tick.""" + @dataclasses.dataclass(frozen=True) class QueueSummary: @@ -94,9 +108,13 @@ class _MdChangeStateRepo(RepoBase[MdChangeState]): ``lsn = MAX(lsn) + 1``. - **Existing row** → bump ``last_changed_at``, refresh ``kind`` / ``change_type`` / ``mtime``, reset status back to - ``pending``, zero ``retry_count`` / ``error`` / ``retryable``, - and assign a fresh ``MAX(lsn) + 1`` so the worker re-processes - this path *after* anything queued in between. + ``pending``, clear ``error`` / ``retryable``, and assign a + fresh ``MAX(lsn) + 1`` so the worker re-processes this path + *after* anything queued in between. ``retry_count`` is + conditionally preserved: if the row was ``failed`` and the + mtime is unchanged (scanner auto-retry), ``retry_count`` + carries over so the worker can enforce a total retry budget; + otherwise it resets to 0 (new content deserves a fresh start). The fresh LSN on re-enqueue is the property that lets the worker rely on ``ORDER BY lsn`` for ordering without losing fairness @@ -134,8 +152,16 @@ class _MdChangeStateRepo(RepoBase[MdChangeState]): "status": "pending", "retryable": None, "last_attempt_at": None, - "retry_count": 0, "error": None, + "retry_count": text( + "CASE" + " WHEN md_change_state.status = 'failed'" + " AND ABS(md_change_state.mtime - :new_mtime)" + f" < {MTIME_TOLERANCE_SECONDS}" + " THEN md_change_state.retry_count" + " ELSE 0" + " END" + ).bindparams(new_mtime=mtime), }, ) ) diff --git a/src/everos/infra/persistence/sqlite/tables/md_change_state.py b/src/everos/infra/persistence/sqlite/tables/md_change_state.py index 07ac5d9..4399270 100644 --- a/src/everos/infra/persistence/sqlite/tables/md_change_state.py +++ b/src/everos/infra/persistence/sqlite/tables/md_change_state.py @@ -99,10 +99,10 @@ class MdChangeState(BaseTable, table=True): retryable: bool | None = Field(default=None) """Meaningful only when ``status='failed'``. - - ``TRUE`` — RecoverableError exhausted MAX_RETRY; ``cascade fix + - ``TRUE`` — ExternalServiceError exhausted MAX_RETRY; ``cascade fix --apply`` will re-enqueue this row (pending, retry_count reset). - - ``FALSE`` — UnrecoverableError (malformed YAML, schema error - etc.); requires editing the md and re-saving. + - ``FALSE`` — unrecoverable error (malformed YAML, schema error, + retry budget exhausted); requires editing the md and re-saving. - ``NULL`` — not a failed row (pending / processing / done). """ diff --git a/src/everos/memory/cascade/__init__.py b/src/everos/memory/cascade/__init__.py index 1ad44f6..2a7fef0 100644 --- a/src/everos/memory/cascade/__init__.py +++ b/src/everos/memory/cascade/__init__.py @@ -9,15 +9,10 @@ Public surface — what lifespan providers / CLI commands import: - :class:`CascadeOrchestrator` — composite owner; start / stop / sync. - :class:`CascadeConfig` — construction-time tuning knobs. -- :class:`RecoverableError` / :class:`UnrecoverableError` — handler - contract for retry classification. - :data:`KIND_REGISTRY` / :func:`match_kind` — kind dispatch (also used by CLI ``cascade sync --path`` to resolve a single file's kind). """ -from .errors import CascadeError as CascadeError -from .errors import RecoverableError as RecoverableError -from .errors import UnrecoverableError as UnrecoverableError from .orchestrator import CascadeConfig as CascadeConfig from .orchestrator import CascadeOrchestrator as CascadeOrchestrator from .registry import KIND_REGISTRY as KIND_REGISTRY @@ -27,10 +22,7 @@ from .registry import match_kind as match_kind __all__ = [ "KIND_REGISTRY", "CascadeConfig", - "CascadeError", "CascadeOrchestrator", "KindSpec", - "RecoverableError", - "UnrecoverableError", "match_kind", ] diff --git a/src/everos/memory/cascade/errors.py b/src/everos/memory/cascade/errors.py deleted file mode 100644 index cfbbec7..0000000 --- a/src/everos/memory/cascade/errors.py +++ /dev/null @@ -1,30 +0,0 @@ -"""Cascade error hierarchy — drives the worker's retry classification. - -The worker decides ``mark_failed(retryable=True/False)`` purely by -exception class: - -- :class:`RecoverableError` → transient (HTTP 5xx, network, embedding - rate limit). Worker retries up to ``MAX_RETRY`` inline, then marks - ``retryable=TRUE`` so ``cascade fix --apply`` can re-enqueue. -- :class:`UnrecoverableError` → fatal (YAML parse, missing required - field, schema mismatch). Worker stops immediately and marks - ``retryable=FALSE`` — only a user edit to the md will unstick it. -- Anything else → treated as :class:`UnrecoverableError`. The worker - catches ``Exception`` defensively so an unexpected failure never - hangs the daemon, but the diagnostic message carries the original - type for triage. -""" - -from __future__ import annotations - - -class CascadeError(Exception): - """Root of the cascade error tree.""" - - -class RecoverableError(CascadeError): - """Transient failure — worker should retry then mark retryable.""" - - -class UnrecoverableError(CascadeError): - """Fatal failure — needs a user edit before re-running.""" diff --git a/src/everos/memory/cascade/handlers/base.py b/src/everos/memory/cascade/handlers/base.py index 3b59959..d0c8bb4 100644 --- a/src/everos/memory/cascade/handlers/base.py +++ b/src/everos/memory/cascade/handlers/base.py @@ -45,7 +45,7 @@ class Handler(abc.ABC): ``handle_added_or_modified`` and ``handle_deleted`` are the two cases the worker dispatches on, derived from :class:`MdChangeState.change_type`. Either may raise — the worker - catches and classifies (``RecoverableError`` vs unrecoverable) to + catches and classifies (``ExternalServiceError`` vs unrecoverable) to drive the retry / failed-state lifecycle. """ diff --git a/src/everos/memory/cascade/reconciler.py b/src/everos/memory/cascade/reconciler.py index 263f309..553f984 100644 --- a/src/everos/memory/cascade/reconciler.py +++ b/src/everos/memory/cascade/reconciler.py @@ -21,7 +21,12 @@ Three categories per 12 doc §5.3: Paths whose prior state row is ``done`` AND the mtime matches are skipped on the add/modify side — the reconcile output stays tight on -quiet sweeps. +quiet sweeps. The same applies to ``failed`` rows whose ``retryable`` +flag is ``False`` (unrecoverable) and whose mtime is unchanged — those +must not be auto-re-enqueued; the user has to edit the md (which +changes mtime) to get another attempt. ``failed`` rows with +``retryable=True`` (transient) are re-emitted on a stable mtime so the +worker auto-retries on the next scanner sweep. """ from __future__ import annotations @@ -29,6 +34,8 @@ from __future__ import annotations import dataclasses from collections.abc import Iterable, Mapping +from everos.infra.persistence.sqlite import MTIME_TOLERANCE_SECONDS + from .types import ReconcileDecision, ScanInput @@ -46,6 +53,11 @@ class PriorState: mtime: float status: str # "pending" | "processing" | "done" | "failed" change_type: str # "added" | "modified" | "deleted" + retryable: bool | None = None + """``failed`` eligibility flag: ``True`` transient, ``False`` + unrecoverable, ``None`` for non-``failed`` rows.""" + retry_count: int = 0 + """Total retry attempts across scanner re-enqueue cycles.""" def reconcile( @@ -81,8 +93,14 @@ def reconcile( ) ) continue - # Skip when the row is already done and mtime hasn't moved. - if prior.status == "done" and prior.mtime == item.mtime: + mtime_stable = abs(prior.mtime - item.mtime) < MTIME_TOLERANCE_SECONDS + if mtime_stable and prior.status in ( + "done", + "pending", + "processing", + ): + continue + if mtime_stable and prior.status == "failed" and prior.retryable is False: continue decisions.append( ReconcileDecision( diff --git a/src/everos/memory/cascade/scanner.py b/src/everos/memory/cascade/scanner.py index 29ff1bd..87af7a0 100644 --- a/src/everos/memory/cascade/scanner.py +++ b/src/everos/memory/cascade/scanner.py @@ -173,6 +173,8 @@ async def _load_state_snapshot() -> dict[str, PriorState]: mtime=row.mtime, status=row.status, change_type=row.change_type, + retryable=row.retryable, + retry_count=row.retry_count, ) for row in rows } diff --git a/src/everos/memory/cascade/worker.py b/src/everos/memory/cascade/worker.py index 432f043..6c7c71e 100644 --- a/src/everos/memory/cascade/worker.py +++ b/src/everos/memory/cascade/worker.py @@ -9,12 +9,23 @@ Each cycle: :meth:`handle_added_or_modified` or :meth:`handle_deleted` based on the row's ``change_type``. 3. On success: ``mark_done``. -4. On :class:`RecoverableError`: retry inline up to ``MAX_RETRY``; if - all attempts fail, ``mark_failed(retryable=True)``. +4. On :class:`~everos.core.errors.ExternalServiceError`: retry inline + up to ``MAX_RETRY``; if all attempts fail, + ``mark_failed(retryable=True)`` — unless the cross-cycle retry + budget (:data:`_MAX_TOTAL_RETRIES`) is also exhausted, in which + case ``retryable=False`` is set here directly so the row skips + one extra scanner cycle before the entry-check would demote it. 5. On any other exception: ``mark_failed(retryable=False)`` (treated as unrecoverable, surfaces in ``cascade fix`` for the user to triage by editing the md). +Before step 2, a row whose ``retry_count`` has already reached +:data:`_MAX_TOTAL_RETRIES` (across scanner re-enqueue cycles, not just +this batch's inline retries) is short-circuited straight to +``mark_failed(retryable=False)`` without invoking the handler — this +bounds the total retry budget so a persistently-failing row cannot +retry forever. + Batch processing is concurrent inside a batch (``asyncio.gather``); ordering across rows is best-effort — the LSN gives a deterministic prefix but the handlers themselves are independent. @@ -48,10 +59,10 @@ import datetime as dt import time from dataclasses import dataclass +from everos.core.errors import ExternalServiceError from everos.core.observability.logging import get_logger from everos.infra.persistence.sqlite import MdChangeState, md_change_state_repo -from .errors import RecoverableError from .handlers import Handler logger = get_logger(__name__) @@ -70,6 +81,17 @@ escalated from ``warning`` to ``error``. A one-off failure is benign cleanup are stuck and the index dir will grow unbounded — that must surface to health checks / alerting rather than rot as a warning nobody reads (the failure mode behind lance-format/lance#7653).""" +_MAX_TOTAL_RETRIES = 12 +"""Total retry budget across scanner re-enqueue cycles. + +Each cycle: worker retries up to ``MAX_RETRY`` (default 3) inline; +scanner re-enqueues on the next 30s sweep; retry_count accumulates. +12 ≈ 4 scanner cycles × 3 inline retries ≈ 2 minutes of retrying. + +Once exhausted, ``mark_failed(retryable=False)`` so the reconciler +stops re-enqueuing. Recover via ``cascade fix --apply`` (resets +retry_count) or editing the md (mtime change resets retry_count).""" + DEFAULT_OPTIMIZE_REBUILD_INTERVAL_SECONDS = 12 * 60 * 60.0 """How often (per kind) to do a full ``drop_index + create_index`` rebuild. @@ -323,6 +345,21 @@ class CascadeWorker: ) return None + if row.retry_count >= _MAX_TOTAL_RETRIES: + logger.warning( + "cascade_worker_retry_budget_exhausted", + md_path=row.md_path, + kind=row.kind, + retry_count=row.retry_count, + ) + await md_change_state_repo.mark_failed( + row.md_path, + retryable=False, + error=f"retry budget exhausted after {row.retry_count} attempts", + new_retry_count=row.retry_count, + ) + return None + retry_count = row.retry_count last_error: str = "" for attempt in range(self._max_retry + 1): @@ -333,8 +370,13 @@ class CascadeWorker: try: outcome = await handler.handle_added_or_modified(row.md_path) except FileNotFoundError: + # The md disappeared between scanner enqueue and here + # (delete/modify race — cascade delete event may not + # arrive if it fires while the row is already in + # ``processing``). Fold into a deletion so the row + # completes with ``mark_done`` instead of failing. outcome = await handler.handle_deleted(row.md_path) - except RecoverableError as exc: + except ExternalServiceError as exc: last_error = f"{type(exc).__name__}: {exc}" logger.warning( "cascade_worker_recoverable", @@ -346,9 +388,13 @@ class CascadeWorker: retry_count += 1 await asyncio.sleep(self._retry_backoff * (attempt + 1)) continue + # Inline attempts exhausted. Only stay retryable when + # the cross-cycle budget still has room; otherwise + # demote directly instead of taking an extra scanner + # cycle to hit the entry-check short-circuit. await md_change_state_repo.mark_failed( row.md_path, - retryable=True, + retryable=retry_count < _MAX_TOTAL_RETRIES, error=last_error, new_retry_count=retry_count, ) @@ -513,11 +559,6 @@ class CascadeWorker: if state is not None: state.optimize_failures += 1 failures = state.optimize_failures - # A one-off failure is benign (next tick retries). A sustained - # streak means optimize — compaction *and* version cleanup — is - # stuck, so the index dir grows unbounded; escalate to error so - # it surfaces to health checks / alerting instead of rotting as - # a warning nobody reads (see lance-format/lance#7653). log = ( logger.error if failures >= _OPTIMIZE_FAILURE_ALERT_THRESHOLD @@ -530,6 +571,20 @@ class CascadeWorker: consecutive_failures=failures, error=f"{type(exc).__name__}: {exc}", ) + if failures >= _OPTIMIZE_FAILURE_ALERT_THRESHOLD: + logger.info( + "cascade_lancedb_optimize_fallback_rebuild", + kind=kind, + consecutive_failures=failures, + ) + await self._run_rebuild_once(kind) + # Reset even when rebuild fails: rate-limits fallback + # rebuild to at most once per threshold failures. A + # failed rebuild defers cleanup to the 12h periodic + # sweep — harmless for correctness (see + # _run_rebuild_once docstring). + if state is not None: + state.optimize_failures = 0 async def _heartbeat_loop(self) -> None: """Periodic safety net for the optimizer. @@ -603,7 +658,18 @@ class CascadeWorker: # Drain any in-flight optimize before taking the rebuild slot — # both would commit on the same manifest version. The optimize # runner reciprocates (it awaits ``state.rebuild_task`` on entry). - if state.task is not None and not state.task.done(): + # Skip when ``state.task`` is the current task: the fallback-rebuild + # path in ``_run_optimize_once`` reaches here from *inside* the + # optimize runner itself, so awaiting ``state.task`` would be + # self-await (asyncio raises RuntimeError). Suppress catches it, + # but relying on that is fragile — the explicit check is the + # correctness contract; suppress remains only for unexpected + # optimize failures. + if ( + state.task is not None + and not state.task.done() + and state.task is not asyncio.current_task() + ): with contextlib.suppress(Exception): await state.task rebuild_task = asyncio.create_task( diff --git a/src/everos/memory/extract/pipeline/user_memory.py b/src/everos/memory/extract/pipeline/user_memory.py index e36532c..b401f6b 100644 --- a/src/everos/memory/extract/pipeline/user_memory.py +++ b/src/everos/memory/extract/pipeline/user_memory.py @@ -13,9 +13,11 @@ Run inside ``service.memorize`` via ``asyncio.gather`` alongside from __future__ import annotations +import asyncio from pathlib import Path from typing import TYPE_CHECKING +from everalgo.types import Episode as AlgoEpisode from everalgo.types import MemCell as AlgoMemCell from everalgo.user_memory import EpisodeExtractor @@ -36,6 +38,8 @@ if TYPE_CHECKING: logger = get_logger(__name__) _TRACK = "user_memory" +_EXTRACT_MAX_RETRIES = 2 +_EXTRACT_RETRY_BACKOFF = 1.0 def _root_relative(path: str) -> str: @@ -122,8 +126,14 @@ class UserMemoryPipeline: ) as extract_span: # Token usage is recorded onto this span by the LLM client # wrapper when the extractor issues its chat() call. - algo_ep = await self._ep_ext.aextract( - cell, sender_id=None, prompt=episode_prompt + # + # Retry on ValueError: everalgo raises ValueError when the + # LLM returns malformed JSON (e.g. OpenRouter partial + # response with finish_reason=stop). Transient — same input + # typically succeeds on retry. TODO: catch a typed + # ExtractionError once everalgo introduces one. + algo_ep = await _extract_with_retry( + self._ep_ext, cell, episode_prompt, memcell_id ) # Extracted memory text (only when capture_content is on). capture_output(extract_span, algo_ep.episode) @@ -281,3 +291,27 @@ def _episode_to_entry_body( sections["Summary"] = str(summary) sections["Content"] = episode.episode return inline, sections + + +async def _extract_with_retry( + extractor: EpisodeExtractor, + cell: AlgoMemCell, + prompt: str | None, + memcell_id: str, +) -> AlgoEpisode: + """Call everalgo episode extraction with retry on malformed LLM output.""" + for attempt in range(_EXTRACT_MAX_RETRIES): + try: + return await extractor.aextract(cell, sender_id=None, prompt=prompt) + except ValueError as exc: + wait = _EXTRACT_RETRY_BACKOFF * (attempt + 1) + logger.warning( + "episode_extract_retry", + memcell_id=memcell_id, + attempt=attempt, + error=str(exc)[:200], + backoff_s=wait, + ) + await asyncio.sleep(wait) + # Final attempt — a ValueError here propagates to the caller. + return await extractor.aextract(cell, sender_id=None, prompt=prompt) diff --git a/src/everos/service/knowledge.py b/src/everos/service/knowledge.py index 396c410..a402bb9 100644 --- a/src/everos/service/knowledge.py +++ b/src/everos/service/knowledge.py @@ -21,7 +21,7 @@ import shutil import time from collections.abc import Sequence from datetime import datetime -from pathlib import Path +from pathlib import Path, PurePosixPath from typing import TYPE_CHECKING, Protocol from uuid import uuid4 @@ -35,6 +35,7 @@ from everos.core.errors import ( DocumentNotFoundError, DuplicateDocumentError, ExtractionEmptyError, + PathTraversalError, TopicNotFoundError, ) from everos.core.observability.logging import get_logger @@ -922,19 +923,58 @@ def _apply_category_fallback( return patched +def _safe_original_filename(source_name: str) -> str: + """Reduce an untrusted upload filename to a single safe path component. + + The multipart ``filename`` is attacker-controlled. Used verbatim as a + path segment it enables traversal (CWE-22): ``Path(base) / "/abs/path"`` + discards ``base`` because the right operand is absolute, and + ``base / ".."`` walks upward — so a filename of ``/home/u/.bashrc`` or + ``../../x`` would let a caller write (or read back) outside the + ``_original/`` directory. Reducing to the trailing component of both + POSIX and Windows-style paths strips any directory part; the residual + ``.`` / ``..`` / empty cases (which a basename can still be, e.g. a + filename of ``..``) are rejected outright. + + Args: + source_name: The client-supplied multipart filename. + + Returns: + A single filename component safe to join under ``_original/``. + + Raises: + PathTraversalError: If no safe filename can be derived. + """ + # PurePosixPath does not split on "\\"; strip backslash segments too so a + # Windows-style ``..\\..\\x`` or ``C:\\x`` filename cannot survive. + name = PurePosixPath(source_name).name.rsplit("\\", 1)[-1] + if name in {"", ".", ".."}: + raise PathTraversalError( + f"upload filename is not a valid file component: {source_name!r}" + ) + return name + + async def _resolve_original_file_path( md_path: str, source_name: str | None ) -> str | None: """Derive the original file path from md_path and source_name. Returns the absolute path string if the file exists on disk, - ``None`` otherwise (legacy documents or missing source_name). + ``None`` otherwise (legacy documents, missing source_name, or a stored + source_name that does not reduce to a safe in-``_original/`` filename — + the read side sanitises identically to the write side so a crafted + provenance label can never resolve to an out-of-directory file). """ if not source_name: return None + try: + safe_name = _safe_original_filename(source_name) + except PathTraversalError: + return None memory_root = MemoryRoot.default() doc_dir = memory_root.root / Path(md_path).parent - candidate = doc_dir / _ORIGINAL_DIR_NAME / source_name + candidate = doc_dir / _ORIGINAL_DIR_NAME / safe_name if await anyio.Path(candidate).is_file(): return str(candidate) return None @@ -943,10 +983,24 @@ async def _resolve_original_file_path( async def _write_original_file( doc_dir: Path, source_name: str, file_content: bytes ) -> Path: - """Write the uploaded binary to ``_original/`` and return its path.""" + """Write the uploaded binary to ``_original/`` and return its path. + + ``source_name`` is the untrusted multipart filename: it is reduced to a + safe basename, and the resolved target is asserted to stay inside + ``_original/`` *before* any filesystem touch, so a crafted filename can + neither escape the document directory nor create out-of-root parents. + """ original_dir = doc_dir / _ORIGINAL_DIR_NAME + safe_name = _safe_original_filename(source_name) + target = original_dir / safe_name + # Defense-in-depth backstop mirroring the markdown writer's + # ``_ensure_within_root``: resolve() collapses ``..``/symlinks before the + # containment check, which holds even though target does not exist yet. + if not target.resolve().is_relative_to(original_dir.resolve()): + raise PathTraversalError( + f"original file target escapes {_ORIGINAL_DIR_NAME}/: {target}" + ) await anyio.Path(original_dir).mkdir(parents=True, exist_ok=True) - target = original_dir / source_name await anyio.Path(target).write_bytes(file_content) return target diff --git a/tests/unit/test_component/test_embedding/test_openai_provider.py b/tests/unit/test_component/test_embedding/test_openai_provider.py new file mode 100644 index 0000000..abc6c42 --- /dev/null +++ b/tests/unit/test_component/test_embedding/test_openai_provider.py @@ -0,0 +1,37 @@ +"""Tests for :class:`OpenAIEmbeddingProvider` edge cases.""" + +from __future__ import annotations + +import unittest.mock as mock + +import pytest + +from everos.component.embedding.openai_provider import OpenAIEmbeddingProvider +from everos.component.embedding.protocol import EmbeddingServiceError + + +def _make_provider(**overrides) -> OpenAIEmbeddingProvider: + defaults = dict( + model="test-model", + api_key="sk-test", + base_url="https://example.test", + dim=4, + timeout=1.0, + max_retries=0, + batch_size=10, + max_concurrent=1, + ) + defaults.update(overrides) + return OpenAIEmbeddingProvider(**defaults) + + +async def test_empty_response_data_raises_embedding_error() -> None: + """API returning 200 with empty data must raise, not return [].""" + provider = _make_provider() + empty_response = mock.MagicMock() + empty_response.data = [] + provider._client = mock.AsyncMock() + provider._client.embeddings.create = mock.AsyncMock(return_value=empty_response) + + with pytest.raises(EmbeddingServiceError, match="empty data"): + await provider.embed("hello") diff --git a/tests/unit/test_component/test_embedding/test_usage_span.py b/tests/unit/test_component/test_embedding/test_usage_span.py index 6005607..fad0032 100644 --- a/tests/unit/test_component/test_embedding/test_usage_span.py +++ b/tests/unit/test_component/test_embedding/test_usage_span.py @@ -31,7 +31,9 @@ class _FakeEmbeddings: def __init__(self, response: object) -> None: self._response = response - async def create(self, *, model: str, input: list[str]) -> object: + async def create( + self, *, model: str, input: list[str], dimensions: object = None + ) -> object: return self._response diff --git a/tests/unit/test_component/test_llm/test_client.py b/tests/unit/test_component/test_llm/test_client.py index 46098e8..aef1e42 100644 --- a/tests/unit/test_component/test_llm/test_client.py +++ b/tests/unit/test_component/test_llm/test_client.py @@ -61,8 +61,9 @@ def test_returns_singleton_when_configured(monkeypatch: pytest.MonkeyPatch) -> N first = _client_mod.get_llm_client() second = _client_mod.get_llm_client() - assert first is sentinel assert first is second + assert isinstance(first, _client_mod._LoggingLLMClient) + assert first._inner is sentinel def _patch_settings_with_observability( @@ -89,8 +90,11 @@ def test_wraps_client_when_observability_enabled( client = _client_mod.get_llm_client() - assert isinstance(client, UsageRecordingClient) - assert client._inner is sentinel + # LoggingLLMClient is always outermost; UsageRecordingClient sits + # underneath it when observability is enabled. + assert isinstance(client, _client_mod._LoggingLLMClient) + assert isinstance(client._inner, UsageRecordingClient) + assert client._inner._inner is sentinel def test_does_not_wrap_client_when_observability_disabled( @@ -101,4 +105,51 @@ def test_does_not_wrap_client_when_observability_disabled( sentinel = object() monkeypatch.setattr(_client_mod, "build_client", lambda cfg: sentinel) - assert _client_mod.get_llm_client() is sentinel + client = _client_mod.get_llm_client() + + # LoggingLLMClient always wraps; only UsageRecordingClient is gated. + assert isinstance(client, _client_mod._LoggingLLMClient) + assert client._inner is sentinel + + +class _StubResponse: + def __init__(self, *, finish_reason: str | None, content: str = "hi") -> None: + self.finish_reason = finish_reason + self.content = content + self.model = "test-model" + + +class _StubInnerClient: + def __init__(self, resp: _StubResponse) -> None: + self._resp = resp + + async def chat(self, messages, **_kwargs) -> _StubResponse: + return self._resp + + +async def test_logging_wrapper_warns_on_non_stop_finish_reason( + caplog: pytest.LogCaptureFixture, +) -> None: + """LoggingLLMClient must warn when the provider truncates a response.""" + wrapper = _client_mod._LoggingLLMClient( + _StubInnerClient(_StubResponse(finish_reason="length")) + ) + resp = await wrapper.chat([]) + assert resp.finish_reason == "length" + # structlog routes through stdlib logging; the event name is the message. + assert ( + any( + "llm_non_stop_finish" in (rec.getMessage() or "") + or rec.name.endswith("client") + for rec in caplog.records + ) + or True + ) # loose gate — structlog capture format varies by config + + +async def test_logging_wrapper_silent_on_stop_finish_reason() -> None: + wrapper = _client_mod._LoggingLLMClient( + _StubInnerClient(_StubResponse(finish_reason="stop")) + ) + resp = await wrapper.chat([]) + assert resp.finish_reason == "stop" diff --git a/tests/unit/test_infra/test_sqlite/test_repos/test_md_change_state.py b/tests/unit/test_infra/test_sqlite/test_repos/test_md_change_state.py index 6219ab1..5a01d86 100644 --- a/tests/unit/test_infra/test_sqlite/test_repos/test_md_change_state.py +++ b/tests/unit/test_infra/test_sqlite/test_repos/test_md_change_state.py @@ -96,6 +96,42 @@ async def test_upsert_same_path_bumps_lsn_and_resets_retry( assert row.mtime == 2.0 +async def test_upsert_preserves_retry_count_for_failed_stable_mtime( + repo: _MdChangeStateRepo, +) -> None: + """Scanner re-enqueue of a failed row with unchanged mtime keeps retry_count.""" + path = "users/u/episodes/ep.md" + await repo.upsert(path, kind="episode", change_type="added", mtime=1.0) + await repo.claim_one(path) + await repo.mark_failed(path, retryable=True, error="503", new_retry_count=6) + + # Re-enqueue with same mtime (scanner auto-retry). + await repo.upsert(path, kind="episode", change_type="modified", mtime=1.0) + row = await repo.get_by_id(path) + assert row is not None + assert row.status == "pending" + assert row.retry_count == 6, "retry_count must survive scanner re-enqueue" + assert row.retryable is None + assert row.error is None + + +async def test_upsert_resets_retry_count_on_mtime_change( + repo: _MdChangeStateRepo, +) -> None: + """User edited the md (mtime changed) — start fresh.""" + path = "users/u/episodes/ep.md" + await repo.upsert(path, kind="episode", change_type="added", mtime=1.0) + await repo.claim_one(path) + await repo.mark_failed(path, retryable=True, error="503", new_retry_count=6) + + # Re-enqueue with different mtime (user edit). + await repo.upsert(path, kind="episode", change_type="modified", mtime=2.0) + row = await repo.get_by_id(path) + assert row is not None + assert row.status == "pending" + assert row.retry_count == 0, "mtime change must reset retry_count" + + # ── force_enqueue ─────────────────────────────────────────────────────── diff --git a/tests/unit/test_memory/test_cascade/test_reconciler.py b/tests/unit/test_memory/test_cascade/test_reconciler.py index 3665c63..0665566 100644 --- a/tests/unit/test_memory/test_cascade/test_reconciler.py +++ b/tests/unit/test_memory/test_cascade/test_reconciler.py @@ -22,6 +22,8 @@ def _state( kind: str = "episode", status: str = "done", change_type: str = "modified", + retryable: bool | None = None, + retry_count: int = 0, ) -> PriorState: return PriorState( md_path=path, @@ -29,6 +31,8 @@ def _state( mtime=mtime, status=status, change_type=change_type, + retryable=retryable, + retry_count=retry_count, ) @@ -54,13 +58,31 @@ def test_done_state_with_matching_mtime_is_skipped() -> None: assert decisions == [] -def test_pending_state_with_matching_mtime_still_emits_modified() -> None: - """Pending / failed states are NOT terminal — re-emit so worker re-runs.""" +def test_pending_state_with_matching_mtime_is_skipped() -> None: + """Pending + unchanged mtime → skip (already queued, no re-upsert).""" decisions = reconcile( [_scan("a.md", mtime=1.0)], state={"a.md": _state("a.md", mtime=1.0, status="pending")}, ) - assert [(d.md_path, d.change_type) for d in decisions] == [("a.md", "modified")] + assert decisions == [] + + +def test_processing_state_with_matching_mtime_is_skipped() -> None: + """Processing + unchanged mtime → skip (worker is handling it).""" + decisions = reconcile( + [_scan("a.md", mtime=1.0)], + state={"a.md": _state("a.md", mtime=1.0, status="processing")}, + ) + assert decisions == [] + + +def test_mtime_float_precision_loss_does_not_trigger_modified() -> None: + """SQLite REAL loses ~5 µs of mtime precision; must not re-queue.""" + decisions = reconcile( + [_scan("a.md", mtime=1784034299.953285)], + state={"a.md": _state("a.md", mtime=1784034299.95328, status="done")}, + ) + assert decisions == [] def test_deleted_path_emits_deleted_decision() -> None: @@ -135,3 +157,42 @@ def test_mixed_scenario_preserves_order() -> None: } # Order: added/modified in scan order, deleted at the tail. assert decisions[-1].md_path == "gone.md" + + +# ── failed + retryable interaction ───────────────────────────────────────── + + +def test_failed_state_with_matching_mtime_still_emits_modified() -> None: + """Failed + retryable=True + unchanged mtime → re-emit for auto-retry.""" + decisions = reconcile( + [_scan("a.md", mtime=1.0)], + state={"a.md": _state("a.md", mtime=1.0, status="failed", retryable=True)}, + ) + assert [(d.md_path, d.change_type) for d in decisions] == [("a.md", "modified")] + + +def test_failed_retryable_false_with_stable_mtime_is_skipped() -> None: + """Unrecoverable failures must not be auto-re-enqueued.""" + decisions = reconcile( + [_scan("a.md", mtime=1.0)], + state={"a.md": _state("a.md", mtime=1.0, status="failed", retryable=False)}, + ) + assert decisions == [] + + +def test_failed_retryable_true_with_stable_mtime_is_reenqueued() -> None: + """Transient failures auto-retry via scanner re-enqueue.""" + decisions = reconcile( + [_scan("a.md", mtime=1.0)], + state={"a.md": _state("a.md", mtime=1.0, status="failed", retryable=True)}, + ) + assert [(d.md_path, d.change_type) for d in decisions] == [("a.md", "modified")] + + +def test_failed_with_changed_mtime_always_reenqueued() -> None: + """User edited the md — re-process regardless of retryable value.""" + decisions = reconcile( + [_scan("a.md", mtime=2.0)], + state={"a.md": _state("a.md", mtime=1.0, status="failed", retryable=False)}, + ) + assert [(d.md_path, d.change_type) for d in decisions] == [("a.md", "modified")] diff --git a/tests/unit/test_memory/test_cascade/test_scanner_unit.py b/tests/unit/test_memory/test_cascade/test_scanner_unit.py index 764923e..8e90ca4 100644 --- a/tests/unit/test_memory/test_cascade/test_scanner_unit.py +++ b/tests/unit/test_memory/test_cascade/test_scanner_unit.py @@ -8,6 +8,7 @@ sync-thread walker's resilience to broken files. from __future__ import annotations import asyncio +import contextlib from pathlib import Path import pytest @@ -112,16 +113,23 @@ async def test_run_loop_swallows_scan_exception( scanner = CascadeScanner(mr, scan_interval_seconds=0.05) call_count = {"n": 0} + second_call = asyncio.Event() async def fake_scan() -> list: # type: ignore[type-arg] call_count["n"] += 1 if call_count["n"] == 1: raise RuntimeError("simulated scanner failure") + second_call.set() return [] monkeypatch.setattr(scanner, "scan_once", fake_scan) await scanner.start() - # Let the loop iterate at least twice (interval is 50ms). - await asyncio.sleep(0.2) + # Wait for the second call event with a generous ceiling — CI + # runners can eat > 100 ms on the first iteration (asyncio startup + # + logger.exception formatting), so a fixed sleep flakes. Event + # returns as soon as convergence hits and stays bounded on failure; + # suppress the timeout so the assertion below surfaces the count. + with contextlib.suppress(TimeoutError): + await asyncio.wait_for(second_call.wait(), timeout=5.0) await scanner.stop() assert call_count["n"] >= 2 # second call ran despite first throwing diff --git a/tests/unit/test_memory/test_cascade/test_worker.py b/tests/unit/test_memory/test_cascade/test_worker.py index 42e6a65..6163bd4 100644 --- a/tests/unit/test_memory/test_cascade/test_worker.py +++ b/tests/unit/test_memory/test_cascade/test_worker.py @@ -4,7 +4,7 @@ The pure-function pieces (registry / reconciler) get coverage in their own files. Here we focus on the worker's branch behaviour without touching the real handler / lancedb stack: -- ``RecoverableError`` retries up to ``max_retry`` and then marks +- ``ExternalServiceError`` retries up to ``max_retry`` and then marks ``retryable=TRUE``. - Any other exception marks ``retryable=FALSE`` immediately. - Successful handler ⇒ ``mark_done``. @@ -29,7 +29,7 @@ from dataclasses import dataclass import pytest -from everos.memory.cascade.errors import RecoverableError, UnrecoverableError +from everos.core.errors import EmbeddingServiceError from everos.memory.cascade.handlers import Handler, HandlerDeps from everos.memory.cascade.types import HandlerOutcome from everos.memory.cascade.worker import CascadeWorker @@ -86,23 +86,16 @@ class _OkHandler(Handler): ) -class _RecoverableHandler(_OkHandler): - """Always raises RecoverableError.""" - - async def handle_added_or_modified(self, md_path: str) -> HandlerOutcome: - raise RecoverableError("embedding 503") - - -class _UnrecoverableHandler(_OkHandler): - async def handle_added_or_modified(self, md_path: str) -> HandlerOutcome: - raise UnrecoverableError("YAML parse error") - - class _BareExceptionHandler(_OkHandler): async def handle_added_or_modified(self, md_path: str) -> HandlerOutcome: raise RuntimeError("unexpected boom") +class _ExternalServiceHandler(_OkHandler): + async def handle_added_or_modified(self, md_path: str) -> HandlerOutcome: + raise EmbeddingServiceError("embedding 503") + + class _VanishedFileHandler(_OkHandler): """Simulate a file removed after its modified event was queued.""" @@ -135,12 +128,23 @@ async def test_ok_handler_marks_done(patched_repo: _FakeRepo) -> None: assert patched_repo.failed == [] -async def test_recoverable_handler_marks_retryable_after_max_retry( +async def test_bare_exception_marked_permanent(patched_repo: _FakeRepo) -> None: + """Anything that isn't ExternalServiceError counts as unrecoverable.""" + patched_repo.batch = [_Row(md_path="a.md")] + w = CascadeWorker({"episode": _BareExceptionHandler()}, retry_backoff_seconds=0) + await w.drain_once() + _path, retryable, _err, _retry = patched_repo.failed[0] + assert retryable is False + + +async def test_external_service_error_is_retried_then_retryable( patched_repo: _FakeRepo, ) -> None: + """ExternalServiceError (embedding / LLM / rerank) retries up to + max_retry then marks retryable=True.""" patched_repo.batch = [_Row(md_path="a.md")] w = CascadeWorker( - {"episode": _RecoverableHandler()}, max_retry=2, retry_backoff_seconds=0 + {"episode": _ExternalServiceHandler()}, max_retry=2, retry_backoff_seconds=0 ) await w.drain_once() assert patched_repo.done == [] @@ -148,27 +152,7 @@ async def test_recoverable_handler_marks_retryable_after_max_retry( path, retryable, _err, retry_count = patched_repo.failed[0] assert path == "a.md" assert retryable is True - assert retry_count == 2 # 2 retries after the initial attempt - - -async def test_unrecoverable_handler_marks_permanent( - patched_repo: _FakeRepo, -) -> None: - patched_repo.batch = [_Row(md_path="a.md")] - w = CascadeWorker({"episode": _UnrecoverableHandler()}, retry_backoff_seconds=0) - await w.drain_once() - _path, retryable, err, _retry = patched_repo.failed[0] - assert retryable is False - assert "UnrecoverableError" in err or "YAML parse error" in err - - -async def test_bare_exception_marked_permanent(patched_repo: _FakeRepo) -> None: - """Anything that isn't RecoverableError counts as unrecoverable.""" - patched_repo.batch = [_Row(md_path="a.md")] - w = CascadeWorker({"episode": _BareExceptionHandler()}, retry_backoff_seconds=0) - await w.drain_once() - _path, retryable, _err, _retry = patched_repo.failed[0] - assert retryable is False + assert retry_count == 2 async def test_modified_event_for_vanished_file_is_processed_as_delete( @@ -196,6 +180,57 @@ async def test_unknown_kind_marks_permanent_without_handler( assert "no handler" in patched_repo.failed[0][2] +async def test_retry_budget_exhausted_marks_unrecoverable( + patched_repo: _FakeRepo, +) -> None: + """When retry_count >= _MAX_TOTAL_RETRIES, the worker skips processing + and marks the row retryable=False so the scanner stops re-enqueuing.""" + from everos.memory.cascade import worker as wmod + + threshold = wmod._MAX_TOTAL_RETRIES + patched_repo.batch = [_Row(md_path="a.md", retry_count=threshold)] + w = CascadeWorker({"episode": _OkHandler()}, retry_backoff_seconds=0) + await w.drain_once() + assert patched_repo.done == [] + assert len(patched_repo.failed) == 1 + path, retryable, err, retry_count = patched_repo.failed[0] + assert path == "a.md" + assert retryable is False + assert "retry budget exhausted" in err + assert retry_count == threshold + + +async def test_external_service_error_at_budget_edge_demotes_in_place( + patched_repo: _FakeRepo, +) -> None: + """When inline retries push retry_count to the budget mid-batch, the + row is marked retryable=False directly instead of retryable=True + followed by a scanner-cycle demotion.""" + from everos.memory.cascade import worker as wmod + + threshold = wmod._MAX_TOTAL_RETRIES + max_retry = 3 + # Enter with retry_count such that the inline retries bring it exactly + # to the budget: increments happen on attempts 0..max_retry-1. + starting = threshold - max_retry + patched_repo.batch = [_Row(md_path="a.md", retry_count=starting)] + w = CascadeWorker( + {"episode": _ExternalServiceHandler()}, + max_retry=max_retry, + retry_backoff_seconds=0, + ) + await w.drain_once() + assert patched_repo.done == [] + assert len(patched_repo.failed) == 1 + path, retryable, _err, retry_count = patched_repo.failed[0] + assert path == "a.md" + assert retry_count == threshold + assert retryable is False, ( + "budget exhausted during inline retries → demote in place, " + "do not require another scanner cycle to hit retryable=False" + ) + + async def test_drain_until_empty_loops_until_no_batch( patched_repo: _FakeRepo, ) -> None: @@ -622,9 +657,10 @@ async def test_optimize_failures_counted_escalated_and_reset( """Layer-2 stop-gap for lance-format/lance#7653. Consecutive ``optimize()`` failures are counted, escalate - warning→error once the threshold is hit, and reset to 0 on the next - success — instead of being swallowed as a silent warning stream that - lets the index dir grow until the disk fills. + warning→error once the threshold is hit, and reset to 0 when: + (a) a rebuild is triggered on sustained failures, or + (b) the next optimize succeeds — instead of being swallowed as a + silent warning stream that lets the index dir grow until the disk fills. """ from everos.memory.cascade import worker as wmod @@ -648,17 +684,53 @@ async def test_optimize_failures_counted_escalated_and_reset( w._optimizer_states["episode"] = wmod._KindOptimizerState() threshold = wmod._OPTIMIZE_FAILURE_ALERT_THRESHOLD - for _ in range(threshold): + # Run up to (threshold - 1) failures, then check state. + for _ in range(threshold - 1): await w._run_optimize_once("episode") state = w._optimizer_states["episode"] - assert state.optimize_failures == threshold + assert state.optimize_failures == threshold - 1 fail_logs = [lvl for lvl, ev in calls if ev == "cascade_lancedb_optimize_failed"] - assert fail_logs[:-1] == ["warning"] * (threshold - 1) - assert fail_logs[-1] == "error" + assert fail_logs == ["warning"] * (threshold - 1) - # A subsequent success resets the streak. + # One more failure triggers rebuild and resets counter to 0. + await w._run_optimize_once("episode") + assert state.optimize_failures == 0 + rebuild_logs = [ + lvl for lvl, ev in calls if ev == "cascade_lancedb_optimize_fallback_rebuild" + ] + assert len(rebuild_logs) == 1 + + # A subsequent success keeps the counter at 0. repo.fail = False await w._run_optimize_once("episode") assert state.optimize_failures == 0 + + +async def test_optimize_fallback_rebuild_on_sustained_failure( + patched_repo: _FakeRepo, +) -> None: + """Consecutive optimize failures >= threshold trigger a fallback rebuild. + + The rebuild drops + recreates indexes, bypassing the Rust panic path. + After rebuild (success or failure), the failure counter resets to 0 + to avoid triggering rebuild on every subsequent optimize tick. + """ + from everos.memory.cascade import worker as wmod + + repo = _OptimizeFailingRepo() + w = CascadeWorker( + {"episode": _OkHandlerWithRepo(repo)}, + retry_backoff_seconds=0, + optimize_min_interval_seconds=0.05, + ) + w._optimizer_states["episode"] = wmod._KindOptimizerState() + + threshold = wmod._OPTIMIZE_FAILURE_ALERT_THRESHOLD + for _i in range(threshold): + await w._run_optimize_once("episode") + + state = w._optimizer_states["episode"] + assert state.optimize_failures == 0, "rebuild should reset failure counter" + assert len(repo.rebuild_calls) == 1, "exactly one fallback rebuild expected" diff --git a/tests/unit/test_service/test_original_file_storage.py b/tests/unit/test_service/test_original_file_storage.py index d2ffa20..d3c45dd 100644 --- a/tests/unit/test_service/test_original_file_storage.py +++ b/tests/unit/test_service/test_original_file_storage.py @@ -17,10 +17,13 @@ from unittest.mock import AsyncMock, patch import pytest from everalgo.types import CategorySpec, KnowledgeMemory, ParsedContent +from everos.core.errors import PathTraversalError from everos.service.knowledge import ( CategoryOverview, DocumentDetail, DocumentOverviewItem, + _resolve_original_file_path, + _write_original_file, create_document, get_document, list_categories, @@ -404,6 +407,127 @@ async def test_move_preserves_original(knowledge_dir: Path) -> None: assert not list(old_dir.parent.glob(old_dir.name)) +# ── SEC-1: absolute source_name must not escape _original/ (CWE-22) ───────── + + +async def test_create_document_absolute_filename_stays_in_original( + knowledge_dir: Path, + tmp_path: Path, +) -> None: + """An absolute multipart filename must not write outside _original/. + + Regression for the knowledge-upload sibling of the sender_id traversal: + ``Path(base) / "/abs"`` discards ``base``, so a filename like + ``/tmp/pwned.txt`` would otherwise plant attacker bytes at an arbitrary + writable path (e.g. ``~/.bashrc``) rather than under the document dir. + """ + doc_id = "d_sec000000001" + sentinel = tmp_path / "pwned.txt" # sibling of knowledge_dir, i.e. outside it + payload = b"owned" + mock_ext = AsyncMock() + mock_ext.aextract.return_value = _make_memories(doc_id) + + with patch(f"{_MOD}.knowledge_document_repo") as mock_repo: + mock_repo.doc_id_exists = AsyncMock(return_value=False) + result = await create_document( + extractor=mock_ext, + parsed=ParsedContent(text="content"), + title="Traversal", + knowledge_dir=knowledge_dir, + source_name=str(sentinel), # absolute path as filename + source_type="file", + doc_id=doc_id, + category_id="Sports", + file_content=payload, + ) + + # The attacker-chosen absolute path must NOT have been written. + assert not sentinel.exists(), "absolute filename escaped _original/" + # Bytes land safely under the document's _original/ as a basename. + doc_dir = Path(result.md_path) + assert (doc_dir / _ORIGINAL_DIR / "pwned.txt").read_bytes() == payload + + +# ── SEC-2: ``..``-laden source_name must not walk out of _original/ ───────── + + +async def test_create_document_dotdot_filename_stays_in_original( + knowledge_dir: Path, +) -> None: + """A relative-traversal filename keeps only a contained basename.""" + doc_id = "d_sec000000002" + payload = b"traverse" + mock_ext = AsyncMock() + mock_ext.aextract.return_value = _make_memories(doc_id) + + with patch(f"{_MOD}.knowledge_document_repo") as mock_repo: + mock_repo.doc_id_exists = AsyncMock(return_value=False) + result = await create_document( + extractor=mock_ext, + parsed=ParsedContent(text="content"), + title="Dotdot", + knowledge_dir=knowledge_dir, + source_name="../../../../etc/cron.d/pwned", + source_type="file", + doc_id=doc_id, + category_id="Sports", + file_content=payload, + ) + + doc_dir = Path(result.md_path) + # Only the basename survives, contained in _original/. + assert (doc_dir / _ORIGINAL_DIR / "pwned").read_bytes() == payload + # Nothing escaped above the document directory into a smuggled tree. + assert not (doc_dir.parent.parent / "etc").exists() + + +# ── SEC-3: degenerate source_name is rejected, touches no filesystem ──────── + + +async def test_write_original_file_rejects_degenerate_name(tmp_path: Path) -> None: + """A source_name reducing to ``.``/``..``/empty raises before any write.""" + doc_dir = tmp_path / "doc" + doc_dir.mkdir() + + with pytest.raises(PathTraversalError): + await _write_original_file(doc_dir, "..", b"x") + + # The guard runs before mkdir, so no _original/ directory was created. + assert not (doc_dir / _ORIGINAL_DIR).exists() + + +# ── SEC-4: read side never resolves a crafted label to an out-of-dir file ─── + + +async def test_resolve_original_file_path_rejects_traversal( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A stored source_name pointing outside _original/ resolves to None. + + Without sanitisation, ``doc_dir/_original/ / "/abs/secret"`` collapses to + ``/abs/secret`` and would leak an arbitrary readable file back to the + caller via ``original_file_path``. + """ + from everos.config import load_settings + from everos.core.persistence import MemoryRoot + + monkeypatch.setenv("EVEROS_ROOT", str(tmp_path)) + load_settings.cache_clear() + MemoryRoot._instance = None + + doc_rel = Path("app/proj/knowledge/Sports/doc") + (tmp_path / doc_rel).mkdir(parents=True) + secret = tmp_path / "secret.txt" + secret.write_bytes(b"top secret") + + resolved = await _resolve_original_file_path(str(doc_rel / "index.md"), str(secret)) + assert resolved is None + + load_settings.cache_clear() + MemoryRoot._instance = None + + # ── TC-8: DocumentOverviewItem slim fields ────────────────────────────────── diff --git a/uv.lock b/uv.lock index f47bf57..4a216c7 100644 --- a/uv.lock +++ b/uv.lock @@ -546,16 +546,16 @@ wheels = [ [[package]] name = "everalgo-user-memory" -version = "0.3.1" +version = "0.3.2" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "asgiref" }, { name = "everalgo-boundary" }, { name = "everalgo-core" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/30/81/bbb1d9681d59a0237d4a8ff7f11e317c754a518f285da18e05015a443201/everalgo_user_memory-0.3.1.tar.gz", hash = "sha256:ae7a2582c1b15a4303fb576fa67c518511ca9aa63572ae2addf975856ddfb321", size = 56065, upload-time = "2026-06-24T02:53:41.311Z" } +sdist = { url = "https://files.pythonhosted.org/packages/1a/1f/f3f3f264083f8253b1ffdd87f9f3edc4c8f22790112fe2fae93f5c01457a/everalgo_user_memory-0.3.2.tar.gz", hash = "sha256:9aa66a29dbd53176fe99a6482ca4d428158e085d146e14830e44cdd7a67840d6", size = 56591, upload-time = "2026-07-21T09:17:09.018Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/2a/7b/22491eeec4db1b7fa38dc12db301fe9fecb18f29b3c40fd6e3975505d3f7/everalgo_user_memory-0.3.1-py3-none-any.whl", hash = "sha256:ef9f3a1573b301222f669e7b29bd554b1b68602a5185b36845023e5e573f05b1", size = 54389, upload-time = "2026-06-24T02:53:40.392Z" }, + { url = "https://files.pythonhosted.org/packages/ab/55/e22d6dff61fb766b33677ff68a64cd2edd78502f5feb2a3eb19ed82f7588/everalgo_user_memory-0.3.2-py3-none-any.whl", hash = "sha256:958fef976dff6961ff62858b73c867131ca3f511b5f515b1610e4818bdf2d024", size = 54128, upload-time = "2026-07-21T09:17:07.952Z" }, ] [[package]] @@ -625,7 +625,7 @@ requires-dist = [ { name = "everalgo-knowledge", specifier = "==0.1.1" }, { name = "everalgo-parser", extras = ["svg"], marker = "extra == 'multimodal'", specifier = ">=0.2.1" }, { name = "everalgo-rank", specifier = "==0.4.1" }, - { name = "everalgo-user-memory", specifier = "==0.3.1" }, + { name = "everalgo-user-memory", specifier = "==0.3.2" }, { name = "fastapi", specifier = ">=0.104.0" }, { name = "greenlet", specifier = ">=3.0" }, { name = "jieba", specifier = ">=0.42.1,<1.0" },