fix: backfill missing 1.1.4 fixes into main (#366)
* fix(knowledge): contain original-file write path (CWE-22) The multipart upload filename was joined into ``_original/`` verbatim. An attacker-controlled filename such as ``/tmp/pwned`` or ``../../.bashrc`` would let ``POST /knowledge/documents`` write outside the document directory (the ``/`` operator discards the left operand for absolute paths; ``..`` walks upward). The read side had the symmetric issue. Fix, mirroring the sender_id containment shipped in 1.0.1 (GHSA-c795-2g9c-j48m): - Add ``_safe_original_filename`` reducing the untrusted filename to a single POSIX/Windows-basename component; reject degenerate residuals (``""``, ``"."``, ``".."``) with PathTraversalError. - ``_write_original_file`` asserts ``target.resolve()`` stays inside ``original_dir.resolve()`` before any filesystem touch (mkdir/write). - ``_resolve_original_file_path`` sanitises symmetrically so a stored provenance label can never resolve to an out-of-directory file. Four SEC regression tests cover: absolute filename, ``..`` traversal, degenerate filename rejection, and read-side sanitisation. Backport from GitLab release/v1.1.4 (commit 40f19de) — 1.1.4 shipped this fix; the GitLab -> GitHub sync stopped at 1.1.3, so 1.2.0 regressed the containment. This commit alone restores the fix; the 1.2.1 release PR ships it to PyPI. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * fix(cascade): retry classification, budget, and reconcile races Backport the cascade reliability work that shipped in GitLab 1.1.4 (MR !49 / commit 95db2f5) — six interlocking changes the reviewer should read in order: 1. Worker retry classification uses ExternalServiceError (embedding / LLM / rerank transient failures) as the "retry inline" signal. The legacy RecoverableError hierarchy under cascade/errors.py is removed; the retry contract now lives in the domain error tree (core/errors.py). Docstrings in handlers/base.py and sqlite/tables/md_change_state.py updated to match. 2. Cross-cycle retry budget: _MAX_TOTAL_RETRIES = 12. Once total attempts across scanner cycles exhaust the budget, the worker marks retryable=False in place instead of looping forever on a sustained upstream outage. 3. md_change_state upsert preserves retry_count on scanner re-enqueue when mtime is unchanged (previously reset to 0 every sweep, defeating the budget). mtime change (user edit) still resets the counter. 4. Reconciler no longer re-enqueues pending / processing rows on stable mtime — that was overwriting the worker's mark_done. It also skips failed rows with retryable=False on stable mtime so the entry-check demote path is stable. 5. mtime tolerance (10 ms, MTIME_TOLERANCE_SECONDS) absorbs the SQLite REAL float precision loss that previously flapped the reconcile decision when the same md was rewritten without a real content change. The constant is defined once in the sqlite repo and imported by the reconciler so both sides use the same tol. 6. Worker _run_rebuild_once carries an explicit `state.task is not asyncio.current_task()` guard before awaiting the optimize task — the previous contextlib.suppress was silently swallowing self-await RuntimeError. Kept intact from the GitHub 1.2.0 baseline: - The `except FileNotFoundError → handle_deleted` branch in the worker (delete/modify race — see test_modified_event_for_vanished_file_is_processed_as_delete). Test coverage added: - test_retry_budget_exhausted_marks_unrecoverable - test_external_service_error_at_budget_edge_demotes_in_place - test_upsert_preserves_retry_count_for_failed_stable_mtime - test_upsert_resets_retry_count_on_mtime_change - test_optimize_fallback_rebuild_on_sustained_failure - reconciler mtime-tolerance / stable-mtime skip suite Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * fix(embedding): raise on empty API data; forward MRL dimensions Backport from GitLab 1.1.4 (MR !49 / commit 95db2f5). Two behavioural changes on ``OpenAIEmbeddingProvider._embed_chunk``: 1. ``response.data == []`` now raises ``EmbeddingServiceError`` instead of returning an empty list. Some upstream providers (observed on DeepInfra under load) return HTTP 200 with an empty ``data`` array; the silent zero-vector path was corrupting search indexes without any signal. 2. When ``[embedding] dimensions = N`` is set in ``everos.toml``, the parameter is forwarded to the API so MRL-capable models (OpenAI text-embedding-3-*, Qwen3-Embedding, ...) do server-side truncation with proper re-normalization. Client-side truncation to ``dim`` remains as a fallback for backends that ignore the param. ``openai.NOT_GIVEN`` is used as the sentinel so the request omits the field when the setting is left at the default ``None``. Config plumbing: - ``EmbeddingSettings.dimensions: int | None = None`` - factory forwards ``dimensions=settings.dimensions`` to the provider The provider stays inside the existing ``memory_span`` OTel wrapper and continues to report input-only tokens via ``set_generation_usage`` - both are GitHub 1.2.0 native tracing behaviours preserved intact. Test coverage: - test_empty_response_data_raises_embedding_error (new) - test_usage_span._FakeEmbeddings.create signature updated to accept ``dimensions`` kwarg so the OTel token-recording tests still exercise the same call path Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * fix(extract): retry episode extraction on malformed LLM output Backport from GitLab 1.1.4 (MR !49 / commit 95db2f5). The ``/flush`` synchronous path called ``EpisodeExtractor.aextract`` exactly once. everalgo raises ``ValueError`` when the LLM returns malformed JSON (observed with OpenRouter partial responses where finish_reason=stop but the body is truncated) — the caller was surfaced a 500 for a transient upstream hiccup. ``_extract_with_retry`` wraps the call with two extra attempts at 1s and 2s backoff (final attempt propagates untouched), typed as ``AlgoEpisode`` so the caller path stays annotated. Retry stays inside the existing GitHub 1.2.0 ``memory_span("everos.extract", ...)`` OTel wrapper — the OTel token capture and the retry loop are orthogonal. TODO in the code notes we should catch a typed everalgo ``ExtractionError`` once that type is introduced (currently ValueError is broad). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * chore: log non-stop finish_reason; bump everalgo-user-memory 0.3.2 Two loosely coupled changes from GitLab 1.1.4 (MR !49 / commit 95db2f5) that arrive together as a housekeeping commit. 1. ``_LoggingLLMClient`` diagnostic wrapper (new, file-private). Wraps the raw everalgo LLM client and, on every ``chat()``, warns when ``resp.finish_reason != "stop"`` — logging the reason, ``content_len``, the last 200 chars of ``content``, and ``model``. Aims at OpenRouter/DeepSeek truncation triage where the provider silently caps output length and returns finish_reason=length / filter / etc. Non-invasive: one branch per call, no config gate. Wrapper stack in ``get_llm_client``: LoggingLLMClient(UsageRecordingClient(build_client(...))) LoggingLLMClient(build_client(...)) # observability off ``UsageRecordingClient`` (GitHub 1.2.0 native OTel token capture) stays gated by ``settings.observability.enabled`` — this commit preserves that. LoggingLLMClient is always outermost so the reason it observes is exactly the reason the underlying provider reported. 2. ``everalgo-user-memory`` 0.3.1 -> 0.3.2 (pyproject + uv.lock). Same bump the GitLab 1.1.4 release lane took; unblocks the episode-extract retry work in commit 4 seeing the upstream improvements. Verified via ``uv sync``. No functional API changes. Test coverage: - test_returns_singleton_when_configured now asserts the outer LoggingLLMClient wrapper. - test_wraps_client_when_observability_enabled asserts the two-layer Logging(UsageRecording(...)) stack. - test_does_not_wrap_client_when_observability_disabled asserts Logging still wraps when tracing is off. - test_logging_wrapper_warns_on_non_stop_finish_reason (new). - test_logging_wrapper_silent_on_stop_finish_reason (new). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * docs(changelog): restore [1.1.4] section to match published sdist The 1.1.4 changelog entry on this branch previously listed three items (Langfuse example, delete/modify race, live-server telemetry). The 1.1.4 sdist on PyPI, however, was built from the internal release lane and includes the CWE-22 containment, the cascade retry-budget / mtime-tolerance / reconcile-guard work, the embedding empty-data raise, the episode-extract retry, MRL dimensions, and the LLM finish_reason diagnostic — none of which were represented here when the tag was cut. Rewrite the [1.1.4] section so it matches the wheel a user actually installs from PyPI: - Add a header note explaining the retroactive restoration. - Fixed: CWE-22, cascade reliability bundle, delete/modify race (unchanged wording), embedding empty-data, episode extract retry, Langfuse live-server telemetry (unchanged wording). - Added: MRL dimensions, LLM finish_reason diagnostic, Langfuse example (unchanged wording). - Changed: everalgo-user-memory 0.3.1 -> 0.3.2. The GitLab-side `.gitlab-ci.yml` in-house-runner entry is dropped — open-source CI runs on GitHub Actions and the internal runner switch is not visible to public users. Date stays 2026-07-20 (the GitHub v1.1.4 tag date / PyPI upload timestamp) rather than the internal 2026-07-23 code-freeze date, so the timeline of what shipped where remains internally consistent. The corresponding code fixes are all backported by earlier commits in this PR; this commit only aligns the changelog surface. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Kendrick-Song
GitHub
parent
6dcd3ebda2
commit
42629dfd4d
65
CHANGELOG.md
65
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
|
||||
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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]
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
},
|
||||
)
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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).
|
||||
"""
|
||||
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -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."""
|
||||
|
|
@ -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.
|
||||
"""
|
||||
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
@ -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
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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 ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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")]
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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 ──────────────────────────────────
|
||||
|
||||
|
||||
|
|
|
|||
8
uv.lock
8
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" },
|
||||
|
|
|
|||
Loading…
Reference in New Issue