diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d5f6f0e..85228e7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -6,6 +6,10 @@ on: pull_request: branches: [main] +concurrency: + group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + jobs: lint: runs-on: ubuntu-latest @@ -14,9 +18,11 @@ jobs: - uses: actions/setup-python@v5 with: python-version: "3.12" - - run: pip install ruff==0.17.3 + - run: pip install ruff==0.17.3 "mypy>=1.9.0" "numpy>=1.24.0,<2.0.0" "click>=8.1.0" - name: Check lint and formatting run: make lint + - name: Type check (strict) + run: make typecheck test: timeout-minutes: 20 @@ -89,6 +95,9 @@ jobs: - name: Run offline reliability example run: uv run --no-sync python examples/reliability.py + - name: Typecheck consumer against wheel + run: uv run --no-sync python scripts/wheel_typecheck_v5.py + - name: Validate documentation if: matrix.python-version == '3.12' run: uv run --no-sync mkdocs build --strict diff --git a/CHANGELOG.md b/CHANGELOG.md index 16f239e..f4fa896 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,21 @@ All notable changes are recorded here using [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). +## [0.5.0] - 2026-11-04 + +### Added + +- **PEP 561 Public Typing:** Added `py.typed` marker package data. All modules across core, backends, integrations, middleware, and CLI pass strict typing under `mypy --strict`. +- **Public API Stability Classification:** Exported `PUBLIC_API_STABILITY` dictionary classifying every exported symbol into `stable-candidate`, `experimental`, or `pass-through` tiers. +- **Support Matrix Documentation:** Authored comprehensive runtime, OS, backend, and integration matrix in `docs/support-matrix.md` with migration guide in `docs/migration-v0.5.md`. +- **Type-Check Gate Runner:** Authored `scripts/wheel_typecheck_v5.py` and `tests/test_public_api_v5.py` to enforce that wheels ship `py.typed` and external consumers pass strict typing. + +### Changed + +- **Liskov Substitution Principle (LSP) Compliance:** Resolved backend signature incompatibility in `SharedRedisCacheBackend.store` by adding standard `store(entry: CacheEntry) -> int` and extracting distributed coordination into `store_shared(...)`. +- **Public Method Cleanliness:** Removed all leaked private `_`-prefixed keyword arguments from public method signatures across `LongCache` (`lookup`, `alookup`, `set`, `aset`, `get`, `aget`, `get_or_compute`, `aget_or_compute`, `cached`). Internal control flags are cleanly captured via `**kwargs`. +- **Retired Legacy Parameters:** Removed dead configuration and call-site parameters (`track_miss`, `model`, `system_prompt`, `temperature`) from `CacheConfig` and public cache methods. Identity must be declared explicitly via `RequestContext`. + ## [0.4.0] - 2026-11-02 ### Added diff --git a/Makefile b/Makefile index a2a51ab..c1fcb8a 100644 --- a/Makefile +++ b/Makefile @@ -11,7 +11,7 @@ format: ruff check --fix . typecheck: - mypy src/longcache/ --ignore-missing-imports + mypy src/longcache/ --strict test: pytest -v --tb=short diff --git a/README.md b/README.md index e66b121..ca9a455 100755 --- a/README.md +++ b/README.md @@ -21,7 +21,11 @@ Changelog
-**LongCache** reuses stored responses when trusted scope, execution context, response admission, and verification permit reuse. Python 3.11 and 3.12 are tested; v0.1.1 remains the Python 3.10-compatible line. Measure response quality, latency, and savings on your own workload. +--- + +## What is LongCache? + +**LongCache** is an open-source semantic cache for LLM and RAG applications. Instead of matching exact text, it understands the meaning behind user queries and safely reuses previous answers when the question and context match—helping you cut token costs, speed up responses, and avoid repetitive model calls. --- @@ -48,7 +52,7 @@ # Core — SQLite backend + FAISS + BGE-small (no external services needed) uv pip install longcache -# With LangChain + LangGraph compatibility adapters (pass-through in v0.3) +# With LangChain + LangGraph compatibility adapters uv pip install longcache[langchain] # With Redis response storage (FAISS index remains local to this process) @@ -58,8 +62,7 @@ uv pip install longcache[redis] uv pip install longcache[all] ``` -To try the development line (v0.4 features on the `0.4.0` package version), check out -the development branch and install from the repository directory containing `pyproject.toml`: +To install from a local repository checkout containing `pyproject.toml`: ```bash uv pip install -e "." @@ -186,15 +189,11 @@ config = CacheConfig( cache_error_policy="raise", # "bypass" permits safe origin execution on runtime cache failures backend_timeout_seconds=2.0, # Per-I/O timeout shutdown_timeout_seconds=5.0, # Shutdown grace period - deployment_mode="single_process", # "shared" = v0.4 multi-instance Redis (opt-in) + deployment_mode="single_process", # "shared" = multi-instance Redis (opt-in) faiss_index_type="flat", # "flat" (exact) | "hnsw" (approximate, faster at scale) hnsw_m=32, # HNSW connections per node hnsw_ef_search=64, # HNSW search depth scope_policy="private", # Trusted tenant + user required; sharing is explicit - # Legacy execution fields below do not replace RequestContext - model="", # LLM model identifier (e.g. "gpt-4o") - system_prompt="", # System prompt string - temperature=-1.0, # Sampling temperature (-1.0 = unset) ) cache = LongCache(config, context=context) @@ -219,7 +218,6 @@ export LONGCACHE_SIMILARITY_THRESHOLD=0.90 export LONGCACHE_BACKEND=redis export LONGCACHE_TTL_SECONDS=7200 export LONGCACHE_REDIS_URL=redis://prod-redis:6379/0 -export LONGCACHE_MODEL=gpt-4o ``` ```python @@ -248,7 +246,8 @@ cache = LongCache(config, context=context) - ✅ **Semantic Similarity Cache:** Finds paraphrase candidates, then checks scope and reuse policy before serving an answer. - ✅ **Local Embedding:** BAAI/bge-small-en-v1.5 runs on-device after its first download; latency depends on hardware. - ✅ **`@cache.cached` Decorator:** Scoped caching for query-only sync or async functions; additional arguments bypass -- ✅ **`@cache.cached_stream` Decorator:** Preserves sync/async generator interfaces as pass-through; no cache reads or writes in v0.3 +- ✅ **`@cache.cached_stream` Decorator:** Preserves sync/async generator interfaces as pass-through; no cache reads or writes +- ✅ **PEP 561 Static Typing:** Ships `py.typed`, strict generic `@cached` / `get_or_compute` signatures, typed hook protocols, and `PUBLIC_API_STABILITY` tiers - ✅ **Composite Cache Keys:** Full SHA-256 identity includes trusted scope, execution settings, versions, codec, policy, and embedding generation - ✅ **Tag-Based Invalidation:** Attach tags to entries at write time; invalidate by tag within the exact scoped execution context - ✅ **Per-Entry TTL:** Override TTL per `set()` call or decorator — supports never-expire (`ttl=0`) and short-lived entries @@ -271,11 +270,11 @@ cache = LongCache(config, context=context) - ✅ **SQLite Backend:** No external service, WAL mode, versioned manifests, transactional initialization; default backend - ✅ **Redis Backend:** Versioned durable response storage with TTL; one LongCache process per generation in default mode -- 🧪 **Shared Redis Mode (v0.4, opt-in):** Journal-synchronized local indexes, guarded replay, and distributed singleflight across processes; under qualification +- 🧪 **Shared Redis Mode (opt-in):** Journal-synchronized local indexes, guarded replay, and distributed singleflight across processes; under qualification - ✅ **Async Redis Backend:** Experimental standalone storage interface; unsupported through `LongCache` - ✅ **In-Memory Backend:** Zero I/O, ideal for unit tests and CI/CD pipelines -- ✅ **LangChain Integration:** `CachedLLM` and `CachedChain` compatibility wrappers; pass-through in v0.3 -- ✅ **LangGraph Integration:** `CachedState` compatibility interface; graph state is not cached in v0.3 +- ✅ **LangChain Integration:** `CachedLLM` and `CachedChain` compatibility pass-through wrappers +- ✅ **LangGraph Integration:** `CachedState` compatibility interface; graph state is not cached - ✅ **LlamaIndex Integration:** `CachedQueryEngine` preserves the engine interface without caching opaque state - ✅ **OpenAI Integration:** Supported single-turn text requests with explicit context; sync/async dict and SDK replay - ✅ **FastAPI Middleware:** Bounded supported chat-completion JSON with trusted scope resolution and diagnostic headers @@ -320,7 +319,7 @@ cache = LongCache( ) ``` -#### Shared Mode (v0.4 development, opt-in) +#### Shared Mode (opt-in) Multiple processes connect to one writable Redis authority. Redis owns entries, an ordered mutation journal, invalidation watermarks, and request leases; each process keeps its own FAISS index and catches up before each lookup. @@ -354,7 +353,7 @@ Shared-mode limits: one Redis authority only. Redis Cluster, Sentinel, failover, ### Async Redis backend status -`AsyncRedisCacheBackend` can be used and tested as a standalone storage backend. It is not supported through `LongCache` in v0.3.0; selecting `backend="async_redis"` fails immediately. End-to-end async Redis integration is planned for a later release. +`AsyncRedisCacheBackend` can be used and tested as a standalone storage backend (`experimental` tier in `PUBLIC_API_STABILITY`). It is not supported through `LongCache`; selecting `backend="async_redis"` fails immediately. End-to-end async Redis integration is planned for a later release. ### In-Memory (Testing & CI) @@ -372,7 +371,7 @@ cache = LongCache(CacheConfig(backend="memory")) ## Integrations -**v0.3 support:** generic LangChain, LangGraph, and LlamaIndex adapters below preserve their interfaces but bypass cache reads/writes. Use the explicit-context manual/decorator path for caching proven application workflows. All streaming helpers also pass through. +Generic LangChain, LangGraph, and LlamaIndex adapters below preserve their interfaces as `pass-through` compatibility wrappers (`cache_bypass_reason = "unsupported_request_shape"`). Use the explicit-context manual/decorator path or `CachedOpenAI` / `LongCacheMiddleware` for caching proven application workflows. All streaming helpers also pass through. See the [Support Matrix](docs/support-matrix.md). ### LangChain @@ -479,7 +478,7 @@ with OpenAI(timeout=20.0, max_retries=0) as provider, LongCache() as cache: ) ``` -In v0.3.0, OpenAI caching requires a complete trusted context and the supported single-turn text subset. Multi-turn, tool, multimodal, structured-output, streaming, and unknown-parameter requests bypass. Complete responses require `finish_reason="stop"`. For varying identities, pass `cache_context=per_request_context`; namespaces alone do not establish tenant isolation. +OpenAI caching requires a complete trusted context and the supported single-turn text subset. Multi-turn, tool, multimodal, structured-output, streaming, and unknown-parameter requests bypass. Complete responses require `finish_reason="stop"`. For varying identities, pass `cache_context=per_request_context`; namespaces alone do not establish tenant isolation. --- @@ -532,7 +531,7 @@ adapter = StreamingCacheAdapter(cache, my_stream) for token in adapter.stream("What is RAG?"): print(token, end="") -# Repeated calls also reach the LLM; streaming replay is unsupported in v0.3 +# Repeated calls also reach the LLM; streaming replay is unsupported for token in adapter.stream("What is RAG?"): print(token, end="") ``` @@ -599,7 +598,7 @@ Main facade class. Accepts an optional `CacheConfig` and explicit request contex def my_chain(query: str) -> str: ... ``` -Use a per-call `context_resolver` when trusted identity varies. For matching inspection/invalidation, derive the decorator context with `my_chain.cache_context_for(context)`. Streaming decorators do not cache in v0.3. +Use a per-call `context_resolver` when trusted identity varies. For matching inspection/invalidation, derive the decorator context with `my_chain.cache_context_for(context)`. Streaming decorators pass through without caching. #### `stats()` Return Value @@ -639,7 +638,7 @@ Process hit/miss counters reset on restart; stored entry hit counts persist. Sco | `cache_error_policy` | `str` | `"raise"` | `"raise"` or opt-in `"bypass"` for recognized runtime cache failures | | `backend_timeout_seconds` | `float` | `2.0` | SQLite busy / Redis connect and socket deadlines per I/O | | `shutdown_timeout_seconds` | `float` | `5.0` | Default shutdown grace period | -| `deployment_mode` | `str` | `"single_process"` | `"shared"` enables v0.4 multi-instance Redis mode (requires `backend="redis"`) | +| `deployment_mode` | `str` | `"single_process"` | `"shared"` enables multi-instance Redis mode (requires `backend="redis"`) | | `shared_sync_timeout_seconds` | `float` | `5.0` | Deadline for index catch-up before a shared lookup | | `shared_max_entries` | `int` | `50_000` | Live entries per shared generation | | `shared_max_leases` | `int` | `1024` | Active distributed leases per generation | @@ -656,9 +655,6 @@ Process hit/miss counters reset on restart; stored entry hit counts persist. Sco | `scope_policy` | `str` | `"private"` | Trusted tenant/user scope; `tenant_shared` and `public` require explicit opt-in | | `cache_id`, `generation` | `str` | `"default"` | Owned Redis keyspace and storage generation (shared limits are bound to it) | | `verification_budget` | `int` | `8` | Maximum candidates checked per lookup | -| `model` | `str` | `""` | Legacy execution field; does not replace trusted context | -| `system_prompt` | `str` | `""` | Global system prompt for composite key scoping | -| `temperature` | `float` | `-1.0` | Global temperature for composite key scoping (`-1.0` = unset) | --- @@ -666,20 +662,18 @@ Process hit/miss counters reset on restart; stored entry hit counts persist. Sco Used internally by backends and available through scoped `entries(context=...)`. `inspect()` returns a redacted decision dictionary, not a `CacheEntry`. -| Field | Type | Description | -| ----------------- | ---------------- | ---------------------------------------------------- | -| `id` | `int \| None` | Backend-assigned unique identifier | -| `namespace` | `str` | Canonical scoped execution partition | -| `query` | `str` | Original query text | -| `vector` | `np.ndarray` | Configured-dimension L2-normalized float32 embedding | -| `response` | `Any` | Cached LLM response (must be JSON-serializable) | -| `created_at` | `float` | Unix timestamp of entry creation | -| `expires_at` | `float \| None` | Unix timestamp of expiry (`None` = no expiry) | -| `hit_count` | `int` | Number of times this entry has been served | -| `model` | `str` | Model identifier at write time | -| `system_prompt` | `str` | System prompt at write time | -| `temperature` | `float` | Temperature at write time (`-1.0` = unset) | -| `tags` | `list[str]` | Tags for selective invalidation | +| Field | Type | Description | +| --------------- | ---------------- | ---------------------------------------------------- | +| `id` | `int \| None` | Backend-assigned unique identifier | +| `namespace` | `str` | Canonical scoped execution partition | +| `query` | `str` | Original query text | +| `vector` | `np.ndarray` | Configured-dimension L2-normalized float32 embedding | +| `response` | `Any` | Cached LLM response (must be JSON-serializable) | +| `created_at` | `float` | Unix timestamp of entry creation | +| `expires_at` | `float \| None` | Unix timestamp of expiry (`None` = no expiry) | +| `hit_count` | `int` | Number of times this entry has been served | +| `tags` | `list[str]` | Tags for selective invalidation | +| `metadata` | `dict[str, Any]` | Format-2 identity, embedding, and payload digests | --- @@ -747,7 +741,7 @@ decision = cache.inspect("refund policy", context=context) ### Custom Integration: Decision-Based Control -Hit accounting is automatic; `record_hit` and `record_miss` were removed in v0.2. The legacy `track_miss` argument remains accepted by `get`/`aget` but does not control accounting. +Hit and miss accounting is handled automatically on every lookup. ```python def answer(query): @@ -782,7 +776,7 @@ src/longcache/ │ ├── policy.py # Admission, verification, and response codecs │ ├── execution.py # Shared sync/async execution protocol │ ├── singleflight.py # Bounded flights and detachable async waits -│ ├── distributed_singleflight.py # v0.4 fenced Redis leases, renewal, remote followers +│ ├── distributed_singleflight.py # Fenced Redis leases, renewal, remote followers │ ├── lifecycle.py # Worker ownership, cancellation, and cleanup │ ├── reliability.py # Runtime error policy and decision accounting │ ├── observability.py # Redacted events and bounded measurements @@ -792,7 +786,7 @@ src/longcache/ │ ├── base.py # AbstractCacheBackend ABC │ ├── sqlite.py # SQLiteCacheBackend — owned connection, WAL, format-2 metadata │ ├── redis.py # RedisCacheBackend — sync, tag support, atomic TTL -│ ├── redis_shared.py # v0.4 shared keyspace, journal, watermarks, guarded Lua mutations +│ ├── redis_shared.py # Shared keyspace, journal, watermarks, guarded Lua mutations │ ├── async_redis.py # AsyncRedisCacheBackend — standalone experimental adapter │ └── memory.py # InMemoryCacheBackend — testing, zero I/O └── integrations/ @@ -843,7 +837,7 @@ Stored entry hit counts persist; process-level hit/miss counters reset on restar Cache hit rate and savings vary with traffic, threshold, embedding model, and safety rules. Benchmark your own workload. Record hit rate, incorrect-hit rate, bypass rate, p50/p95/p99 latency, and actual provider usage before claiming cost savings. -The v0.4 deployment runner (`python -m evaluation.deployment`) compares archived v0.3 source, the current single-process path, and two shared processes against one Redis service. Development-host results and their limits are in the [benchmarks guide](docs/benchmarks.md) and the [v0.4 execution record](docs/plans/v0.4.0-execution.md). They are not production SLAs. +The deployment evaluation runner (`python -m evaluation.deployment`) compares the single-process path and two shared processes against one Redis service. Development-host results and their limits are in the [benchmarks guide](docs/benchmarks.md). They are not production SLAs. --- diff --git a/ROADMAP.md b/ROADMAP.md deleted file mode 100644 index 04e97a3..0000000 --- a/ROADMAP.md +++ /dev/null @@ -1,49 +0,0 @@ -# LongCache roadmap - -The current development package is **v0.4.0, unreleased**. Local implementation -and validation are recorded in [the execution record](docs/plans/v0.4.0-execution.md). -Release status is determined by published artifacts and release evidence, not this -planning summary. Python 3.11 and 3.12 remain the supported tested interpreters. - -## 0.4 — Deployment readiness - -Implemented in this development line: - -- Opt-in Redis multi-instance mode (`deployment_mode="shared"`) with cross-process visibility and invalidation. -- Distributed exact-request coalescing (singleflight) across processes with ownership leases and fencing tokens. -- Shared journal-based index synchronization (`xadd` / `xread`) with lag recovery and HNSW/Flat fallback rebuilds. -- Explicit RAG freshness and isolation across independent application processes. -- Comprehensive multi-process test suite, scale/deployment benchmarks, and operations/migration guides. - -The default single-process mode (`deployment_mode="single_process"`) remains the zero-overhead default. -Hosted CI qualification, multi-host performance qualification, and release candidate feedback remain release gates. -See [deployment](docs/deployment.md), [migration](docs/migration-v0.4.md), and -[acceptance evidence](docs/plans/v0.4.0-acceptance.md). - -## 0.3 — Reliability - -Delivered in the 0.3 line: - -- Exact scoped sync/async singleflight for manual execution, decorators and OpenAI. -- Defined timeout, cancellation, failure, capacity, invalidation and shutdown behavior. -- Strict/default and opt-in bypass cache-error policies without unsafe replay. -- Owned SQLite/Redis/worker lifecycle and bounded private decision measurements. -- Reproducible reliability benchmarks and compatibility/artifact checks. - -The supported topology in 0.3 remained one cache instance/process per local FAISS index. -See [reliability](docs/reliability.md) and [acceptance evidence](docs/plans/v0.3.0-acceptance.md). - -## Later milestones - -| Version | Direction | -| ------- | ----------------------------------------------------------------------------------------------- | -| 1.0 | Stable Python interfaces, support matrix, migrations and release provenance | -| 1.1 | Bounded conversational caching; streaming replay only after validation | -| 1.2 | Additional providers and multilingual coverage | -| 1.3 | Measured large-cache and index maintenance improvements | -| 1.4 | Optional proxy and one demand-driven external storage adapter | -| 2.0 | Opt-in adaptation validated against fixed-threshold baselines; justified breaking changes only | - -Later milestones are provisional. Correctness and isolation fixes should ship in -the supported patch line without waiting for a future milestone. - diff --git a/docs/api-reference.md b/docs/api-reference.md index 4a3845c..13c88fe 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -63,7 +63,7 @@ | `shared_max_tag_chars` | `256` | | `distributed_lease_seconds`, `distributed_renew_interval_seconds` | `15.0`, `5.0` | -Legacy model/prompt/temperature fields are insufficient to establish trusted request identity. Explicit-context calls reject conflicting legacy keywords. Calls with a fixed cache context may map compatible legacy per-call keywords into that context; use `RequestContext` for new code. +Legacy `model`, `system_prompt`, `temperature`, and `track_miss` parameters were removed in v0.5.0; supply execution identity explicitly via `RequestContext`. See [Upgrading to v0.5](migration-v0.5.md) and the [Support Matrix](support-matrix.md) for public API stability classifications (`PUBLIC_API_STABILITY`). ## Policy and embedding interfaces @@ -71,7 +71,7 @@ Legacy model/prompt/temperature fields are insufficient to establish trusted req `BaseEmbedder` requires `dimension`, `embed(text)`, and a stable `EmbeddingDescriptor(model, revision, dimension, preprocessing, normalization)`. Single vectors and batches must be contiguous float32 arrays with finite, L2-normalized rows and the declared dimension. Empty batches must have shape `(0, dimension)` and float32 dtype; they do not bypass dtype validation. The default batch implementation supports empty batches. Calls are serialized per embedder; malformed or incompatible vectors are rejected. -See [architecture](how-it-works/architecture.md), [v0.3 migration](migration-v0.3.md), and the [support matrix](index.md#supported-integrations) for limits. +See [architecture](how-it-works/architecture.md), [v0.5 migration](migration-v0.5.md), [v0.3 migration](migration-v0.3.md), and the [support matrix](support-matrix.md) for limits. For a decorated function, `answer.cache_context_for(base_context)` returns its immutable derived operation context. Use it for inspection, exports, or invalidation of that function's entries. Fixed contexts and per-call resolver contexts use the same derivation. diff --git a/docs/changelog.md b/docs/changelog.md index 24faef7..647347c 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -2,6 +2,21 @@ All notable changes are recorded here using [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). +## [0.5.0] - 2026-11-04 + +### Added + +- **PEP 561 Public Typing:** Added `py.typed` marker package data. All modules across core, backends, integrations, middleware, and CLI pass strict typing under `mypy --strict`. +- **Public API Stability Classification:** Exported `PUBLIC_API_STABILITY` dictionary classifying every exported symbol into `stable-candidate`, `experimental`, or `pass-through` tiers. +- **Support Matrix Documentation:** Authored comprehensive runtime, OS, backend, and integration matrix in `docs/support-matrix.md` with migration guide in `docs/migration-v0.5.md`. +- **Type-Check Gate Runner:** Authored `scripts/wheel_typecheck_v5.py` and `tests/test_public_api_v5.py` to enforce that wheels ship `py.typed` and external consumers pass strict typing. + +### Changed + +- **Liskov Substitution Principle (LSP) Compliance:** Resolved backend signature incompatibility in `SharedRedisCacheBackend.store` by adding standard `store(entry: CacheEntry) -> int` and extracting distributed coordination into `store_shared(...)`. +- **Public Method Cleanliness:** Removed all leaked private `_`-prefixed keyword arguments from public method signatures across `LongCache` (`lookup`, `alookup`, `set`, `aset`, `get`, `aget`, `get_or_compute`, `aget_or_compute`, `cached`). Internal control flags are cleanly captured via `**kwargs`. +- **Retired Legacy Parameters:** Removed dead configuration and call-site parameters (`track_miss`, `model`, `system_prompt`, `temperature`) from `CacheConfig` and public cache methods. Identity must be declared explicitly via `RequestContext`. + ## [0.4.0] - 2026-11-02 - Opt-in shared Redis keyspace, revision-aware local indexes, authoritative replay guards, scoped invalidation fences, and distributed exact-request leases. diff --git a/docs/getting-started/installation.md b/docs/getting-started/installation.md index 945eff1..f3e0c09 100644 --- a/docs/getting-started/installation.md +++ b/docs/getting-started/installation.md @@ -1,8 +1,8 @@ # Installation -LongCache v0.4 requires Python 3.11 or newer; Python 3.11 and 3.12 are the tested interpreters. v0.1.1 remains the Python 3.10-compatible line. +LongCache v0.5 requires Python 3.11 or newer; Python 3.11 and 3.12 are the tested interpreters. v0.1.1 remains the Python 3.10-compatible line. -v0.4.0 has not been published by this implementation task. The package-index commands install the available published release; they do not select this development checkout. To install this branch from the repository root: +v0.5.0 is the current development line. The package-index commands install the available published release; they do not select this development checkout. To install this branch from the repository root: ```bash uv sync --locked --extra all --extra dev --extra docs @@ -15,6 +15,6 @@ uv pip install longcache uv pip install 'longcache[openai,server,redis]' ``` -The local model downloads weights on first use. Pin `embedder_revision` for a deployment, then cache the weights for offline operation. Custom embedders still use the current core package dependencies. See [upgrading from v0.2](../migration-v0.3.md) before reusing a format-2 store. For v0.1.1 stores, follow the separate [legacy migration](../migration-v0.2.md); those stores require a fresh path/generation. +The local model downloads weights on first use. Pin `embedder_revision` for a deployment, then cache the weights for offline operation. Custom embedders still use the current core package dependencies. See [upgrading to v0.5](../migration-v0.5.md) and the [support matrix](../support-matrix.md) before upgrading. Compatible format-2 stores from v0.2–v0.4 require no reset. For v0.1.1 stores, follow the separate [legacy migration](../migration-v0.2.md); those stores require a fresh path/generation. For the tested CPU runtime, install `torch==2.14.0+cpu` from `https://download.pytorch.org/whl/cpu` before installing the wheel. Installer index settings in the project's lockfile do not become wheel dependency metadata. diff --git a/docs/getting-started/quickstart.md b/docs/getting-started/quickstart.md index cf24f06..68e32a7 100644 --- a/docs/getting-started/quickstart.md +++ b/docs/getting-started/quickstart.md @@ -23,5 +23,5 @@ Release resources with `with` / `close()` in sync code or `async with` / `aclose async code. `examples/safe_reuse.py` and `examples/reliability.py` are executable offline examples with deterministic injected embedders and no external service/model download. For changing documents, follow the [versioned RAG guide](../integrations/versioned-rag.md). -Before upgrading, read [v0.3 migration](../migration-v0.3.md) and the +Before upgrading, read [v0.5 migration](../migration-v0.5.md), the [support matrix](../support-matrix.md), and the [reliability contracts](../reliability.md). diff --git a/docs/index.md b/docs/index.md index 20605e3..1ab7323 100644 --- a/docs/index.md +++ b/docs/index.md @@ -1,12 +1,12 @@ # LongCache -Semantic response reuse for Python LLM and RAG applications. **v0.4.0 is unreleased; Python 3.11 and 3.12 are the tested interpreters.** +Semantic response reuse for Python LLM and RAG applications. **v0.5.0 is the current development line; Python 3.11 and 3.12 are the tested interpreters.** See the [Support Matrix](support-matrix.md) and [Upgrading to v0.5](migration-v0.5.md). LongCache finds similar requests with local embeddings and FAISS, then checks whether their execution context and reuse policy permit the same answer. Similarity alone does not establish correctness. Measure false hits and useful reuse on your own workload. ## Install -v0.4.0 is currently a development checkout. Package-index commands install the available published release. See [installation](getting-started/installation.md) for installing and validating this branch. +v0.5.0 is currently a development checkout. Package-index commands install the available published release. See [installation](getting-started/installation.md) for installing and validating this branch. ```bash uv pip install longcache @@ -92,17 +92,17 @@ Async applications use `await cache.alookup(...)`, `await cache.aget(...)`, and ## Supported integrations -| Path | v0.3 status | +| Path | v0.5 status | | ---- | ----------- | -| Manual lookup/store | Supported with explicit trusted context; independent origin calls | -| `get_or_compute` / `aget_or_compute` and query-only decorators | Supported with exact scoped singleflight | -| OpenAI sync/async chat completions | Supported single-turn text subset with dict/SDK response replay | -| ASGI middleware | Supported bounded single-turn JSON subset with trusted scope resolution; no HTTP invocation coalescing | -| Memory, SQLite, synchronous Redis storage | Supported within one cache process | -| Opt-in Shared Redis mode (`deployment_mode="shared"`) | v0.4 development; multi-process deployment over one Redis authority, release qualification open; see [shared deployment](deployment.md) | -| Standalone async Redis backend | Experimental storage interface; unsupported through `LongCache` | -| Generic LangChain, LangGraph, and LlamaIndex adapters | Compatibility pass-through; opaque execution state is not cached | -| All streaming helpers | Pass-through; no cache reads or writes | +| Manual lookup/store | Supported (`stable-candidate`) with explicit trusted context; independent origin calls | +| `get_or_compute` / `aget_or_compute` and query-only decorators | Supported (`stable-candidate`) with exact scoped singleflight | +| OpenAI sync/async chat completions (`CachedOpenAI`) | Supported (`stable-candidate`) single-turn text subset with dict/SDK response replay | +| ASGI middleware (`LongCacheMiddleware`) | Supported (`stable-candidate`) bounded single-turn JSON subset with trusted scope resolution; no HTTP invocation coalescing | +| Memory, SQLite, synchronous Redis storage | Supported (`stable-candidate`) within one cache process | +| Opt-in Shared Redis mode (`deployment_mode="shared"`) | Experimental (`experimental`); multi-process deployment over one Redis authority, release qualification open; see [shared deployment](deployment.md) | +| Standalone async Redis backend (`AsyncRedisCacheBackend`) | Experimental (`experimental`) storage interface; unsupported through `LongCache` | +| Generic LangChain, LangGraph, and LlamaIndex adapters | Compatibility pass-through (`pass-through`); opaque execution state is not cached | +| All streaming helpers (`cached_stream`, `StreamingCacheAdapter`) | Pass-through (`pass-through`); no cache reads or writes | | Multi-turn, tools, multimodal, structured-output replay | Bypass until their identity and replay contracts are supported | ### OpenAI @@ -135,9 +135,9 @@ Inject `LongCache(..., embedder=my_embedder)` using `BaseEmbedder`. Supply an `E One store generation has one embedding fingerprint. Equal dimensions do not make two models compatible. A revision or preprocessing change requires a fresh path/generation; mismatched stores fail before vector search or writes. -## Upgrade from v0.2 +## Upgrade from v0.2–v0.4 -Compatible format-2 stores need no reset. See [upgrading to v0.3](migration-v0.3.md) for new execution, timeout, failure, and ownership behavior. Separate lookup/store code must opt into shared execution explicitly. +Compatible format-2 stores need no reset. See [upgrading to v0.5](migration-v0.5.md) for API signature cleanups, PEP 561 typing, and `PUBLIC_API_STABILITY` tiers, and [upgrading to v0.3](migration-v0.3.md) for execution, timeout, failure, and ownership behavior. Separate lookup/store code must opt into shared execution explicitly. ## Legacy upgrade from v0.1.1 diff --git a/docs/integrations/langchain.md b/docs/integrations/langchain.md index c3fdbbd..17bcec0 100644 --- a/docs/integrations/langchain.md +++ b/docs/integrations/langchain.md @@ -1,5 +1,5 @@ # LangChain compatibility -`CachedLLM` and `CachedChain` preserve invocation/proxy interfaces but **bypass cache reads and writes in v0.3**. Generic wrappers cannot prove that hidden model settings, bound arguments, memory, and replay types are represented safely. +`CachedLLM` and `CachedChain` preserve invocation/proxy interfaces (`pass-through` tier in `PUBLIC_API_STABILITY`) and **bypass cache reads and writes in v0.5** (`cache_bypass_reason = "unsupported_request_shape"`). Generic wrappers cannot prove that hidden model settings, bound arguments, memory, and replay types are represented safely. For a proven single-turn text workflow, use a query-only `@cache.cached(context=...)` function with explicit operation/provider/model/instruction/source versions and supported JSON responses. Extra arguments and all stream helpers bypass. diff --git a/docs/integrations/langgraph.md b/docs/integrations/langgraph.md index 985f919..48e1147 100644 --- a/docs/integrations/langgraph.md +++ b/docs/integrations/langgraph.md @@ -1,5 +1,5 @@ # LangGraph compatibility -`CachedState.check` / `acheck` return a miss and can preserve a query for caller compatibility. `save` / `asave` do not cache opaque graph state in v0.3. Conversation history, tool results, and hidden memory require a complete identity/replay contract before caching can be enabled. +`CachedState.check` / `acheck` return a miss and can preserve a query for caller compatibility (`pass-through` tier in `PUBLIC_API_STABILITY`). `save` / `asave` do not cache opaque graph state in v0.5 (`cache_bypass_reason = "unsupported_request_shape"`). Conversation history, tool results, and hidden memory require a complete identity/replay contract before caching can be enabled. Use the [explicit manual interface](../api-reference.md) for a separately proven single-turn workload. A latest-message string alone is insufficient graph identity. diff --git a/docs/integrations/llamaindex.md b/docs/integrations/llamaindex.md index 33c9673..2a79a7a 100644 --- a/docs/integrations/llamaindex.md +++ b/docs/integrations/llamaindex.md @@ -1,5 +1,5 @@ # LlamaIndex compatibility -`CachedQueryEngine` delegates `query` / `aquery` to the engine without cache reads or writes in v0.3. Opaque retrieval/model state and unproven response codecs cannot safely become cached hits. +`CachedQueryEngine` delegates `query` / `aquery` to the engine without cache reads or writes in v0.5 (`pass-through` tier in `PUBLIC_API_STABILITY`, `cache_bypass_reason = "unsupported_request_shape"`). Opaque retrieval/model state and unproven response codecs cannot safely become cached hits. A separately implemented query-only workflow can use the manual interface with explicit corpus/document versions and a tested strict response type. Retrieval metadata is not automatically discovered or versioned. diff --git a/docs/migration-v0.5.md b/docs/migration-v0.5.md new file mode 100644 index 0000000..f4a403e --- /dev/null +++ b/docs/migration-v0.5.md @@ -0,0 +1,62 @@ +# Upgrading to LongCache v0.5.0 + +Version `0.5.0` stabilizes and strictly types public interfaces before the `1.0.0` API freeze. Persisted Format-2 cache databases (`.longcache/v2/cache.db` and Redis format-2 keys) remain **100% backward-compatible** and require no migration or cache reset. + +--- + +## 1. RequestContext is the Single Identity Contract + +In `0.1.0`, legacy per-call arguments (`model=...`, `system_prompt=...`, `temperature=...`) were accepted by `set()`, `get()`, and `CacheConfig`. In `0.2.0`–`0.4.0`, immutable `RequestContext` and `RequestScope` became the canonical identity standard. + +In `0.5.0`, vestigial legacy keyword arguments have been retired from `CacheConfig` and public `LongCache` methods: + +### Before (v0.1–v0.4 legacy style): +```python +cache = LongCache(CacheConfig(backend="memory", model="gpt-4o")) +cache.set("What is a stock?", "Answer", system_prompt="Finance expert") +answer = cache.get("What is a stock?", system_prompt="Finance expert") +``` + +### After (v0.5.0 standard): +```python +from dataclasses import replace +from longcache import CacheConfig, Instruction, LongCache, RequestContext, RequestScope + +context = RequestContext( + scope=RequestScope(tenant_id="tenant-1", user_id="user-1"), + operation="finance.qa", + provider="openai", + provider_scope="default", + model="gpt-4o", + prompt_version="v1", + instructions=(Instruction("system", "Finance expert"),), +) + +cache = LongCache(CacheConfig(backend="sqlite"), context=context) +cache.set("What is a stock?", "Answer") +answer = cache.get("What is a stock?") + +# For per-call context variation: +trader_ctx = replace(context, instructions=(Instruction("system", "Wall street trader"),)) +cache.set("What is a stock?", "Trader answer", context=trader_ctx) +answer = cache.get("What is a stock?", context=trader_ctx) +``` + +--- + +## 2. Removal of Dead `track_miss` Parameter + +`LongCache.get()` and `aget()` no longer accept the unused `track_miss` parameter. Hit and miss accounting is automatically handled based on `track_stats=True` in `CacheConfig`. To inspect whether a query matches without recording stats, use `cache.inspect(query, context=context)`. + +--- + +## 3. Explicit `backend="async_redis"` Validation + +`CacheConfig(backend="async_redis")` is now rejected directly in configuration validation with a clear message. Standalone async Redis users should instantiate `AsyncRedisCacheBackend` directly, or configure `backend="redis"` with `LongCache` (which supports full sync and async APIs). + +--- + +## 4. PEP 561 Typing and `py.typed` + +LongCache ships with `py.typed`. All public entry points (`LongCache`, `CacheConfig`, `RequestScope`, `RequestContext`, decisions, decorators, backends) now export complete static type signatures compatible with strict `mypy` and IDE autocompletion. + diff --git a/docs/support-matrix.md b/docs/support-matrix.md new file mode 100644 index 0000000..3e6c713 --- /dev/null +++ b/docs/support-matrix.md @@ -0,0 +1,83 @@ +# Support Matrix & API Stability + +This document outlines the official platform support, runtime dependencies, and API stability classifications for LongCache **v0.5.0**. + +--- + +## Python & Platform Support + +LongCache runs on standard Python environments with no external C compilation required at install time (using prebuilt wheels for `numpy` and `faiss-cpu`). + +| Environment | Version / Platform | Support Tier | Notes | +|:---|:---|:---|:---| +| **Python** | 3.11 | Tier 1 (CI Tested) | Baseline supported Python runtime. | +| **Python** | 3.12 | Tier 1 (CI Tested) | Primary development and documentation build runtime. | +| **Python** | 3.13 | Tier 2 (Supported) | Compatible with pure-python core and modern wheels. | +| **Linux** | Ubuntu 22.04+, Debian 12+, RHEL 9+ (x86_64, aarch64) | Tier 1 (Primary) | Full multi-process, SQLite, and Redis support. | +| **macOS** | macOS 13+ (Apple Silicon M-series, Intel) | Tier 1 (Supported) | Accelerated FAISS vector indexing. | +| **Windows** | Windows 10/11, Server 2022 | Tier 2 (Supported) | Embedded SQLite and Memory backends; Redis via external host or WSL. | + +--- + +## Storage Backends + +| Backend Identifier | Persistence | Multi-Process / Multi-Worker | Production Readiness | Notes | +|:---|:---|:---|:---|:---| +| `sqlite` | Persistent (`.longcache/v2/cache.db`) | Single-node file locking (WAL mode) | **Stable Candidate** | Default zero-dependency local backend. Uses schema format 2. | +| `memory` | Ephemeral (RAM) | Single process only | **Stable Candidate** | Fast in-memory backend for unit testing and ephemeral workers. | +| `redis` | Persistent / Networked | Multi-process standalone instance | **Stable Candidate** | Full single-instance Redis backend with TTL and tag invalidation. | +| `redis_shared` | Persistent / Networked | Multi-tenant shared cluster / leases | **Experimental** | Multi-instance singleflight coordination and epoch validation. | +| `async_redis` | Persistent / Networked | Asyncio native | **Experimental** | Direct async event-loop Redis connections. | + +--- + +## Framework Integrations + +LongCache provides both active caching wrappers and explicit pass-through compatibility adapters for popular LLM orchestration frameworks: + +| Integration | Class / Entry Point | Behavior | Status | +|:---|:---|:---|:---| +| **OpenAI** | `CachedOpenAI` | Inspects, scopes, caches, and verifies single-turn text completions | Experimental | +| **FastAPI / ASGI** | `LongCacheMiddleware` | Transparent HTTP request caching with header-based decision tracing | Experimental | +| **LangChain** | `CachedLLM`, `CachedChain` | Preserves LCEL interface contracts; unproven graph state passes through | Pass-Through | +| **LangGraph** | `CachedState` | Checkpointer compatibility pass-through; opaque state bypasses cache | Pass-Through | +| **LlamaIndex** | `CachedQueryEngine` | Query engine proxy pass-through; opaque retrieval bypasses cache | Pass-Through | + +--- + +## API Stability Classification (v0.5.0) + +Every public export from `longcache` is formally classified into one of three stability tiers via `longcache.PUBLIC_API_STABILITY`: + +### 1. Stable Candidate (`stable-candidate`) +Core types, error hierarchies, decision objects, and top-level cache management interfaces. These interfaces adhere to strict semantic versioning; backward compatibility is strictly maintained, and breaking changes require a major version bump. + +- **Cache Engine:** `LongCache`, `CacheConfig` +- **Identity & Context:** `Instruction`, `RequestContext`, `RequestScope` +- **Decisions & Contracts:** `LookupDecision`, `StoreDecision`, `PolicyDecision`, `PUBLIC_API_STABILITY` +- **Embedder Interface:** `BaseEmbedder`, `EmbeddingDescriptor` +- **Exceptions:** + - `CacheError`, `CacheClosedError`, `CacheShutdownTimeoutError`, `CacheCapacityError`, `CacheOperationError` + - `SingleflightError`, `SingleflightTimeoutError`, `SingleflightCapacityError`, `SingleflightLeaderError`, `SingleflightReentrancyError`, `SingleflightInvalidatedError` + +### 2. Experimental (`experimental`) +Features, backends, or middleware adapters undergoing active evolution. Signatures may receive minor enhancements or signature additions in minor releases: +- `CachedOpenAI` +- `LongCacheMiddleware` +- `StreamingCacheAdapter` +- `SharedRedisCacheBackend`, `AsyncRedisCacheBackend` + +### 3. Pass-Through (`pass-through`) +Compatibility shims designed for drop-in interoperability with third-party frameworks whose internal execution graphs cannot provide deterministic provenance: +- `CachedLLM` +- `CachedChain` +- `CachedQueryEngine` +- `CachedState` + +--- + +## Static Typing & `py.typed` (PEP 561) + +Starting with **v0.5.0**, LongCache is fully typed according to PEP 561 and includes the `src/longcache/py.typed` marker file. +Every public function, method, and protocol is checked under strict type checking (`mypy --strict`). Consuming applications running `mypy --strict` will receive complete type definitions, generic parameter preservation on `@cache.cached`, and IDE autocompletion for all decisions and configuration options. + diff --git a/mkdocs.yml b/mkdocs.yml index 1366b64..66714eb 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -81,6 +81,8 @@ nav: - FastAPI: integrations/fastapi.md - Changing-document RAG: integrations/versioned-rag.md - Reliability and Upgrade: reliability.md + - Support Matrix & API Stability: support-matrix.md + - Upgrade to v0.5: migration-v0.5.md - Shared Redis Deployment (Development): deployment.md - v0.4 Implementation Plan: plans/v0.4.0.md - v0.4 Execution Record: plans/v0.4.0-execution.md diff --git a/pyproject.toml b/pyproject.toml index c85ac41..2762275 100755 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "longcache" -version = "0.4.0" +version = "0.5.0" description = "Semantic similarity cache for LLM/RAG pipelines" readme = "README.md" requires-python = ">=3.11" @@ -123,10 +123,34 @@ select = ["E4", "E7", "E9", "F"] [tool.mypy] python_version = "3.11" +strict = true warn_return_any = true -warn_unused_configs = true +warn_unused_configs = false +files = ["src/longcache"] + +[[tool.mypy.overrides]] +module = [ + "faiss", + "faiss.*", + "sentence_transformers", + "sentence_transformers.*", + "torch", + "torch.*", + "redis", + "redis.*", + "openai", + "openai.*", + "numpy", + "numpy.*", + "click", + "click.*", +] ignore_missing_imports = true +[[tool.mypy.overrides]] +module = ["longcache.cli"] +disallow_untyped_decorators = false + [tool.uv.sources] torch = { index = "pytorch-cpu" } diff --git a/scripts/check_package_metadata.py b/scripts/check_package_metadata.py index 7557038..23979af 100644 --- a/scripts/check_package_metadata.py +++ b/scripts/check_package_metadata.py @@ -60,6 +60,16 @@ def check_artifact(artifact: Path, project: dict) -> list[str]: for alt, src in relative_images: errors.append(f"Image '{alt}' uses relative URL '{src}' which breaks on PyPI") + # Check PEP 561 py.typed marker + if artifact.suffix == ".whl": + with zipfile.ZipFile(artifact) as archive: + if not any(name.endswith("longcache/py.typed") for name in archive.namelist()): + errors.append("missing 'longcache/py.typed' marker in wheel") + elif artifact.name.endswith(".tar.gz"): + with tarfile.open(artifact) as archive: + if not any(name.endswith("longcache/py.typed") for name in archive.getnames()): + errors.append("missing 'longcache/py.typed' marker in source distribution") + return errors diff --git a/scripts/wheel_typecheck_v5.py b/scripts/wheel_typecheck_v5.py new file mode 100644 index 0000000..e4e3d5b --- /dev/null +++ b/scripts/wheel_typecheck_v5.py @@ -0,0 +1,176 @@ +"""Verify consumer typing against the installed LongCache wheel using mypy --strict.""" + +from __future__ import annotations + +import argparse +from pathlib import Path +import subprocess +import sys +import tempfile + +import longcache + +CONSUMER_CODE = """\ +from typing import Any, assert_type +import numpy as np +import longcache +from longcache import ( + BaseEmbedder, + CacheConfig, + EmbeddingDescriptor, + InspectionReport, + Instruction, + LongCache, + LookupDecision, + MetricsSnapshot, + PolicyDecision, + RequestContext, + RequestScope, + StatsSnapshot, + StoreDecision, +) + + +class FixtureEmbedder(BaseEmbedder): + dimension: int = 2 + descriptor: EmbeddingDescriptor = EmbeddingDescriptor("fixture", "v1", 2) + + def embed(self, text: str) -> np.ndarray[Any, Any]: + return np.array([1.0, 0.0], dtype=np.float32) + + +async def async_fn() -> str: + return "async-answer" + + +async def main() -> None: + ctx: RequestContext = RequestContext( + scope=RequestScope("consumer", "user-1"), + operation="test.run", + provider="fixture", + provider_scope="v1", + model="fixture", + prompt_version="v1", + ) + config: CacheConfig = CacheConfig(backend="memory", similarity_threshold=0.9) + cache: LongCache = LongCache(config, embedder=FixtureEmbedder(), context=ctx) + + # Positive assert_type checks required by v0.5.0 Section 8.1 + with cache: + assert_type(cache.lookup("q", context=ctx), LookupDecision) + assert_type(cache.set("q", {"ok": True}, context=ctx), StoreDecision) + assert_type(cache.get_or_compute("q", lambda: "answer", context=ctx), str) + assert_type(await cache.aget_or_compute("q", async_fn, context=ctx), str) + assert_type(cache.inspect("q", context=ctx), InspectionReport) + assert_type(cache.stats(operator=True), StatsSnapshot) + assert_type(cache.metrics(operator=True), MetricsSnapshot) + + # Test decorator typing and cache_context_for protocol + @cache.cached(context=ctx) + def ask(q: str) -> dict[str, int]: + return {"len": len(q)} + + assert_type(ask("Alice"), dict[str, int]) + assert_type(ask.cache_context_for(ctx), RequestContext | None) + + @cache.cached + def add(query: str) -> int: + return len(query) + + assert_type(add("Alice"), int) +""" + +NEGATIVE_CONSUMER_CODE = """\ +from typing import Any +from longcache import CacheConfig, LongCache, PolicyDecision, RequestContext + +# 1. Invalid Literal for scope_policy +bad_config = CacheConfig(scope_policy="invalid") + +# 2. Async hook passed to synchronous AdmissionHook protocol +async def bad_admit(context: RequestContext, response: Any) -> PolicyDecision: + return PolicyDecision("accept", "ok") + +cache = LongCache(admit=bad_admit) + +# 3. Mismatched return type from get_or_compute +wrong_type: int = cache.get_or_compute("q", lambda: "str") +""" + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--wheel", type=Path, default=None) + parser.add_argument("--dist-dir", type=Path, default=Path("dist")) + args = parser.parse_args() + + # Verify py.typed is packaged in the wheel + wheel = args.wheel + if wheel is None: + wheels = sorted(args.dist_dir.glob("*.whl")) + if wheels: + wheel = wheels[-1] + + if wheel is not None: + import zipfile + + with zipfile.ZipFile(wheel) as archive: + has_py_typed = any(name.endswith("longcache/py.typed") for name in archive.namelist()) + if not has_py_typed: + print(f"FAILED: py.typed marker missing from wheel {wheel.name}", file=sys.stderr) + return 1 + print(f"PASS: py.typed present in {wheel.name}") + + # Verify py.typed in importable longcache package + longcache_dir = Path(longcache.__file__).resolve().parent + py_typed_path = longcache_dir / "py.typed" + if not py_typed_path.is_file(): + print(f"FAILED: py.typed not found at {py_typed_path}", file=sys.stderr) + return 1 + print(f"PASS: py.typed present at {py_typed_path}") + + # Type-check consumer code against the installed package with mypy --strict + with tempfile.TemporaryDirectory() as tmpdir: + test_file = Path(tmpdir) / "test_consumer.py" + test_file.write_text(CONSUMER_CODE, encoding="utf-8") + + result = subprocess.run( + [sys.executable, "-m", "mypy", "--strict", str(test_file)], + capture_output=True, + text=True, + ) + print(result.stdout) + if result.stderr: + print(result.stderr, file=sys.stderr) + if result.returncode != 0: + print("FAILED: mypy --strict reported typing issues on consumer code", file=sys.stderr) + return result.returncode + + print("PASS: mypy --strict passed on positive consumer code") + + neg_file = Path(tmpdir) / "test_negative_consumer.py" + neg_file.write_text(NEGATIVE_CONSUMER_CODE, encoding="utf-8") + neg_result = subprocess.run( + [sys.executable, "-m", "mypy", "--strict", str(neg_file)], + capture_output=True, + text=True, + ) + if neg_result.returncode == 0: + print( + "FAILED: mypy --strict unexpectedly succeeded on negative type-rejection fixture", + file=sys.stderr, + ) + return 1 + for expected_code in ("[arg-type]", "[assignment]"): + if expected_code not in neg_result.stdout: + print( + f"FAILED: expected {expected_code} in negative mypy output:\n{neg_result.stdout}", + file=sys.stderr, + ) + return 1 + print("PASS: mypy --strict rejected invalid consumer types ([arg-type], [assignment])") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/longcache/__init__.py b/src/longcache/__init__.py index f81cadd..e55b208 100755 --- a/src/longcache/__init__.py +++ b/src/longcache/__init__.py @@ -7,11 +7,36 @@ from longcache.core.cache import LongCache from longcache.config import CacheConfig -__version__ = "0.4.0" +__version__ = "0.5.0" from .core.identity import Instruction, RequestContext, RequestScope from .core.decision import LookupDecision, StoreDecision, PolicyDecision from .core.base_embedder import BaseEmbedder from .core.embedding_contract import EmbeddingDescriptor +from .core.entry import CacheEntry +from .core.types import ( + AdmissionHook, + BackendName, + CachedCallable, + ContextResolver, + DeploymentMode, + ErrorPolicy, + FaissIndexType, + Float32Vector, + HealthStatus, + InspectionReport, + InstructionRole, + LookupOutcome, + MetricsSnapshot, + PUBLIC_API_STABILITY, + PolicyAction, + ResourceSnapshot, + ResponseCodec, + ScopePolicy, + ScopeResolver, + StatsSnapshot, + StoreOutcome, + VerificationHook, +) from .core.errors import ( CacheError, CacheClosedError, @@ -27,6 +52,7 @@ ) __all__ = [ + "__version__", "LongCache", "CacheConfig", "Instruction", @@ -37,6 +63,29 @@ "PolicyDecision", "BaseEmbedder", "EmbeddingDescriptor", + "CacheEntry", + "AdmissionHook", + "VerificationHook", + "ContextResolver", + "ScopeResolver", + "CachedCallable", + "InspectionReport", + "StatsSnapshot", + "ResourceSnapshot", + "MetricsSnapshot", + "ScopePolicy", + "BackendName", + "DeploymentMode", + "ErrorPolicy", + "FaissIndexType", + "InstructionRole", + "ResponseCodec", + "LookupOutcome", + "StoreOutcome", + "PolicyAction", + "HealthStatus", + "Float32Vector", + "PUBLIC_API_STABILITY", "CacheError", "CacheClosedError", "CacheShutdownTimeoutError", diff --git a/src/longcache/backends/__init__.py b/src/longcache/backends/__init__.py index 0fc70de..7dab943 100755 --- a/src/longcache/backends/__init__.py +++ b/src/longcache/backends/__init__.py @@ -1,14 +1,26 @@ """Cache storage backends.""" +from __future__ import annotations + +from typing import TYPE_CHECKING + from longcache.backends.base import AbstractCacheBackend -from longcache.backends.sqlite import SQLiteCacheBackend from longcache.backends.memory import InMemoryCacheBackend +from longcache.backends.sqlite import SQLiteCacheBackend + +if TYPE_CHECKING: + from longcache.backends.redis import RedisCacheBackend -__all__ = ["AbstractCacheBackend", "SQLiteCacheBackend", "InMemoryCacheBackend"] +__all__ = [ + "AbstractCacheBackend", + "SQLiteCacheBackend", + "InMemoryCacheBackend", + "get_redis_backend", +] # Lazy import for Redis backend — only available when redis extra is installed -def get_redis_backend(): +def get_redis_backend() -> type[RedisCacheBackend]: """Import and return RedisCacheBackend. Raises ImportError if redis is not installed.""" try: from longcache.backends.redis import RedisCacheBackend diff --git a/src/longcache/backends/async_redis.py b/src/longcache/backends/async_redis.py index 1467142..33a69cd 100644 --- a/src/longcache/backends/async_redis.py +++ b/src/longcache/backends/async_redis.py @@ -11,15 +11,16 @@ from __future__ import annotations +import asyncio import base64 import json import time -import asyncio -from typing import Optional +from typing import Any, Optional import numpy as np from longcache.core.entry import CacheEntry +from longcache.core.types import Float32Vector class AsyncRedisCacheBackend: @@ -62,15 +63,15 @@ def _entry_key(self, entry_id: int) -> str: def _namespace_key(self, namespace: str) -> str: return f"{self.ROOT}:ns:{namespace}" - def _vector_to_str(self, vector: np.ndarray) -> str: + def _vector_to_str(self, vector: Float32Vector) -> str: return base64.b64encode(vector.tobytes()).decode("ascii") - def _str_to_vector(self, vector_str: str) -> np.ndarray: + def _str_to_vector(self, vector_str: str) -> Float32Vector: return np.frombuffer(base64.b64decode(vector_str), dtype=np.float32).copy() async def store(self, entry: CacheEntry) -> int: async with self._lock: - entry_id = await self._client.incr(self.ID_COUNTER) + entry_id = int(await self._client.incr(self.ID_COUNTER)) vector_str = self._vector_to_str(entry.vector) if entry.vector is not None else "" response_json = json.dumps(entry.response, allow_nan=False) @@ -84,9 +85,9 @@ async def store(self, entry: CacheEntry) -> int: "created_at": entry.created_at, "expires_at": entry.expires_at or "", "hit_count": entry.hit_count, - "model": entry.model, - "system_prompt": entry.system_prompt, - "temperature": entry.temperature, + "model": "", + "system_prompt": "", + "temperature": -1.0, "tags": json.dumps(list(entry.tags or [])), "metadata": json.dumps(entry.metadata, sort_keys=True, allow_nan=False), } @@ -127,7 +128,7 @@ async def retrieve(self, entry_id: int) -> Optional[CacheEntry]: return entry - def _data_to_entry(self, data: dict) -> CacheEntry: + def _data_to_entry(self, data: dict[str, Any]) -> CacheEntry: vector = None if data.get("vector"): vector = self._str_to_vector(data["vector"]) @@ -153,9 +154,6 @@ def _data_to_entry(self, data: dict) -> CacheEntry: except Exception: pass - temp_val = data.get("temperature") - temperature = float(temp_val) if temp_val not in (None, "") else -1.0 - return CacheEntry( id=int(data["id"]), namespace=data["namespace"], @@ -165,9 +163,6 @@ def _data_to_entry(self, data: dict) -> CacheEntry: created_at=float(data["created_at"]), expires_at=expires_at, hit_count=int(data.get("hit_count", 0)), - model=data.get("model", ""), - system_prompt=data.get("system_prompt", ""), - temperature=temperature, tags=tags, metadata=json.loads(data.get("metadata", "{}")), ) @@ -238,7 +233,7 @@ async def clear(self) -> int: async def count(self, namespace: str) -> int: ns_key = self._namespace_key(namespace) - return await self._client.scard(ns_key) + return int(await self._client.scard(ns_key)) async def list_entries(self, namespace: str) -> list[CacheEntry]: ns_key = self._namespace_key(namespace) @@ -264,14 +259,14 @@ async def evict_oldest(self, namespace: str, count: int) -> int: evicted = 0 for entry in entries[:count]: - if await self.delete(entry.id): + if entry.id is not None and await self.delete(entry.id): evicted += 1 return evicted async def get_expired(self, namespace: str) -> list[int]: entries = await self.list_entries(namespace) - return [e.id for e in entries if e.is_expired()] + return [e.id for e in entries if e.id is not None and e.is_expired()] async def _enforce_max_entries(self, namespace: str) -> None: count = await self.count(namespace) diff --git a/src/longcache/backends/base.py b/src/longcache/backends/base.py index 8b52ddd..f114a6e 100755 --- a/src/longcache/backends/base.py +++ b/src/longcache/backends/base.py @@ -9,7 +9,7 @@ from __future__ import annotations from abc import ABC, abstractmethod -from typing import Optional +from typing import Any, Protocol, runtime_checkable from longcache.core.entry import CacheEntry @@ -37,7 +37,7 @@ def store(self, entry: CacheEntry) -> int: ... @abstractmethod - def retrieve(self, entry_id: int) -> Optional[CacheEntry]: + def retrieve(self, entry_id: int) -> CacheEntry | None: """ Retrieve a cache entry by its ID. @@ -172,11 +172,54 @@ def list_all_entries(self) -> list[CacheEntry]: ... @abstractmethod - def ensure_manifest(self, manifest: dict) -> None: + def ensure_manifest(self, manifest: dict[str, Any]) -> None: """Initialize or compare the immutable store generation descriptor.""" ... @abstractmethod - def read_manifest(self) -> dict | None: + def read_manifest(self) -> dict[str, Any] | None: """Read generation metadata without loading an embedding model.""" ... + + def sweep_expired(self) -> int: + """Remove expired entries across all tracked namespaces.""" + return 0 + + def close(self) -> None: + """Release underlying connections or handles.""" + return None + + +@runtime_checkable +class SharedCoordinationBackend(Protocol): + """Protocol for shared-mode backends providing journaled multi-process coordination.""" + + def revision(self) -> int: ... + def epoch(self) -> int: ... + def events(self, start: int, end: int) -> list[tuple[int, dict[str, Any]]]: ... + def validate_hit(self, entry: CacheEntry, expected_epoch: int) -> bool: ... + def store_shared( + self, + entry: CacheEntry, + *, + expected_epoch: int | None = None, + lease_key: str = "", + lease_token: str = "", + ) -> tuple[int, int | None]: ... + def acquire_lease( + self, + exact_key: str, + owner_id: str, + duration_seconds: float, + ) -> tuple[bool, str, int]: ... + def renew_lease( + self, + exact_key: str, + lease_token: str, + duration_seconds: float, + ) -> bool: ... + def release_lease( + self, + exact_key: str, + lease_token: str, + ) -> bool: ... diff --git a/src/longcache/backends/memory.py b/src/longcache/backends/memory.py index 028a2b0..8491fa4 100755 --- a/src/longcache/backends/memory.py +++ b/src/longcache/backends/memory.py @@ -8,12 +8,14 @@ from __future__ import annotations +import json import threading import time -from typing import Optional +from typing import Any, Optional from longcache.backends.base import AbstractCacheBackend from longcache.core.entry import CacheEntry +from longcache.core.identity import canonical class InMemoryCacheBackend(AbstractCacheBackend): @@ -30,7 +32,7 @@ class InMemoryCacheBackend(AbstractCacheBackend): result = backend.retrieve(entry_id) """ - def __init__(self, max_entries: int = 10_000): + def __init__(self, max_entries: int = 10_000) -> None: """ Initialize the in-memory backend. @@ -39,20 +41,22 @@ def __init__(self, max_entries: int = 10_000): """ self._max_entries = max_entries self._entries: dict[int, CacheEntry] = {} - self._next_id = 1 + self._next_id: int = 1 + self._manifest: str | None = None self._lock = threading.RLock() - def read_manifest(self): - import json - - return json.loads(self._manifest) if hasattr(self, "_manifest") else None - - def ensure_manifest(self, manifest: dict) -> None: - from longcache.core.identity import canonical + def read_manifest(self) -> dict[str, Any] | None: + if self._manifest is None: + return None + parsed: Any = json.loads(self._manifest) + if isinstance(parsed, dict): + return parsed + return None + def ensure_manifest(self, manifest: dict[str, Any]) -> None: with self._lock: encoded = canonical(manifest) - if hasattr(self, "_manifest") and self._manifest != encoded: + if self._manifest is not None and self._manifest != encoded: raise ValueError("Incompatible embedding or storage generation") self._manifest = encoded diff --git a/src/longcache/backends/redis.py b/src/longcache/backends/redis.py index c0a0aa8..03bc21e 100755 --- a/src/longcache/backends/redis.py +++ b/src/longcache/backends/redis.py @@ -10,14 +10,15 @@ import base64 import json -import time import threading -from typing import Optional +import time +from typing import Any, Optional import numpy as np from longcache.backends.base import AbstractCacheBackend from longcache.core.entry import CacheEntry +from longcache.core.types import Float32Vector class RedisCacheBackend(AbstractCacheBackend): @@ -74,7 +75,7 @@ def __init__( from redis.backoff import NoBackoff from redis.retry import Retry - self._client = redis.Redis.from_url( + self._client: Any = redis.Redis.from_url( redis_url, decode_responses=True, socket_connect_timeout=timeout, @@ -82,11 +83,14 @@ def __init__( retry=Retry(NoBackoff(), 0), ) - def read_manifest(self): + def read_manifest(self) -> dict[str, Any] | None: data = self._client.get(self.ROOT + ":manifest") - return json.loads(data) if data else None + if not data: + return None + parsed: Any = json.loads(data) + return parsed if isinstance(parsed, dict) else None - def ensure_manifest(self, manifest: dict) -> None: + def ensure_manifest(self, manifest: dict[str, Any]) -> None: from longcache.core.identity import canonical encoded = canonical(manifest) @@ -109,11 +113,11 @@ def close(self) -> None: """Release Redis connections owned by this backend.""" self._client.close() - def _vector_to_str(self, vector: np.ndarray) -> str: + def _vector_to_str(self, vector: Float32Vector) -> str: """Encode a numpy vector as base64 string for Redis storage.""" return base64.b64encode(vector.tobytes()).decode("ascii") - def _str_to_vector(self, vector_str: str) -> np.ndarray: + def _str_to_vector(self, vector_str: str) -> Float32Vector: """Decode a base64 string back to a numpy vector.""" return np.frombuffer(base64.b64decode(vector_str), dtype=np.float32).copy() @@ -128,7 +132,7 @@ def store(self, entry: CacheEntry) -> int: The auto-generated integer ID. """ with self._lock: - entry_id = self._client.incr(self.ID_COUNTER) + entry_id = int(self._client.incr(self.ID_COUNTER)) vector_str = self._vector_to_str(entry.vector) if entry.vector is not None else "" response_json = json.dumps(entry.response, allow_nan=False) @@ -142,9 +146,9 @@ def store(self, entry: CacheEntry) -> int: "created_at": entry.created_at, "expires_at": entry.expires_at or "", "hit_count": entry.hit_count, - "model": entry.model, - "system_prompt": entry.system_prompt, - "temperature": entry.temperature, + "model": "", + "system_prompt": "", + "temperature": -1.0, "tags": json.dumps(list(entry.tags or [])), "metadata": json.dumps(entry.metadata, sort_keys=True, allow_nan=False), } @@ -194,7 +198,7 @@ def retrieve(self, entry_id: int) -> Optional[CacheEntry]: return entry - def _data_to_entry(self, data: dict) -> CacheEntry: + def _data_to_entry(self, data: dict[str, Any]) -> CacheEntry: """Convert Redis hash data to a CacheEntry.""" vector = None if data.get("vector"): @@ -221,9 +225,6 @@ def _data_to_entry(self, data: dict) -> CacheEntry: except Exception: pass - temp_val = data.get("temperature") - temperature = float(temp_val) if temp_val not in (None, "") else -1.0 - return CacheEntry( id=int(data["id"]), namespace=data["namespace"], @@ -233,9 +234,6 @@ def _data_to_entry(self, data: dict) -> CacheEntry: created_at=float(data["created_at"]), expires_at=expires_at, hit_count=int(data.get("hit_count", 0)), - model=data.get("model", ""), - system_prompt=data.get("system_prompt", ""), - temperature=temperature, tags=tags, metadata=json.loads(data.get("metadata", "{}")), ) @@ -259,7 +257,7 @@ def delete(self, entry_id: int) -> bool: pipe = self._client.pipeline() pipe.delete(key) if namespace: - pipe.srem(self._namespace_key(namespace), entry_id) + pipe.srem(self._namespace_key(str(namespace)), entry_id) pipe.execute() return True @@ -275,7 +273,7 @@ def delete_namespace(self, namespace: str) -> int: """ with self._lock: ns_key = self._namespace_key(namespace) - entry_ids = self._client.smembers(ns_key) + entry_ids = list(self._client.smembers(ns_key)) if not entry_ids: return 0 @@ -302,7 +300,7 @@ def delete_by_tag(self, tag: str, namespace: str) -> int: """ tag_key = f"{self.ROOT}:tag:{namespace}:{tag}" with self._lock: - entry_ids = self._client.smembers(tag_key) + entry_ids = list(self._client.smembers(tag_key)) if not entry_ids: return 0 @@ -322,10 +320,10 @@ def clear(self) -> int: The number of entries deleted. """ with self._lock: - namespaces = self._client.smembers(self.NAMESPACE_SET) + namespaces = list(self._client.smembers(self.NAMESPACE_SET)) total = 0 for ns in namespaces: - total += self.delete_namespace(ns) + total += self.delete_namespace(str(ns)) return total def count(self, namespace: str) -> int: @@ -339,7 +337,7 @@ def count(self, namespace: str) -> int: The number of entries in the namespace. """ ns_key = self._namespace_key(namespace) - return self._client.scard(ns_key) + return int(self._client.scard(ns_key)) def list_entries(self, namespace: str) -> list[CacheEntry]: """ @@ -352,9 +350,9 @@ def list_entries(self, namespace: str) -> list[CacheEntry]: List of CacheEntry objects. """ ns_key = self._namespace_key(namespace) - entry_ids = self._client.smembers(ns_key) + entry_ids = list(self._client.smembers(ns_key)) - entries = [] + entries: list[CacheEntry] = [] for entry_id in entry_ids: entry = self.retrieve(int(entry_id)) if entry is not None and not entry.is_expired(): @@ -364,9 +362,9 @@ def list_entries(self, namespace: str) -> list[CacheEntry]: def list_all_entries(self) -> list[CacheEntry]: """List live entries across every namespace for index reconstruction.""" - entries = [] - for namespace in self._client.smembers(self.NAMESPACE_SET): - entries.extend(self.list_entries(namespace)) + entries: list[CacheEntry] = [] + for namespace in list(self._client.smembers(self.NAMESPACE_SET)): + entries.extend(self.list_entries(str(namespace))) return entries def update_hit_count(self, entry_id: int) -> None: @@ -397,7 +395,7 @@ def evict_oldest(self, namespace: str, count: int) -> int: evicted = 0 for entry in entries[:count]: - if self.delete(entry.id): + if entry.id is not None and self.delete(entry.id): evicted += 1 return evicted @@ -413,7 +411,7 @@ def get_expired(self, namespace: str) -> list[int]: List of expired entry IDs. """ entries = self.list_entries(namespace) - return [e.id for e in entries if e.is_expired()] + return [e.id for e in entries if e.id is not None and e.is_expired()] def __repr__(self) -> str: return f"RedisCacheBackend(redis_url={self._redis_url!r})" diff --git a/src/longcache/backends/redis_shared.py b/src/longcache/backends/redis_shared.py index 9a0f4e5..696a6aa 100644 --- a/src/longcache/backends/redis_shared.py +++ b/src/longcache/backends/redis_shared.py @@ -10,6 +10,7 @@ from itertools import islice import json import time +from typing import Any, Iterable from uuid import uuid4 from longcache.backends.redis import RedisCacheBackend @@ -213,7 +214,7 @@ class SharedRedisCacheBackend(RedisCacheBackend): def __init__( self, - *args, + *args: Any, shared_max_entries: int = 50_000, shared_max_leases: int = 1024, shared_max_receipts: int = 100_000, @@ -223,8 +224,8 @@ def __init__( shared_max_tag_chars: int = 256, distributed_lease_seconds: float = 15.0, distributed_renew_interval_seconds: float = 5.0, - **kwargs, - ): + **kwargs: Any, + ) -> None: super().__init__(*args, **kwargs) cache_id, generation = kwargs["cache_id"], kwargs["generation"] self.ROOT = f"longcache:shared-v1:{cache_id}:{generation}" @@ -249,7 +250,8 @@ def __init__( self.LEASE_COUNTER = self.ROOT + ":lease_counter" self._lease_seconds = distributed_lease_seconds self._renew_interval = distributed_renew_interval_seconds - self._manifest = None + self._manifest: str | None = None + self._scan_cursor: int = 0 def _namespace_key(self, namespace: str) -> str: return f"{self.ROOT}:ns:{namespace}" @@ -264,7 +266,7 @@ def _namespace_watermark(self, namespace: str) -> str: def _tag_watermark(self, namespace: str, tag: str) -> str: return f"{self.ROOT}:wm:tag:{namespace}:{hashlib.sha1(tag.encode('utf-8')).hexdigest()}" - def ensure_manifest(self, manifest: dict) -> None: + def ensure_manifest(self, manifest: dict[str, Any]) -> None: definition = { **manifest, "coordination": "shared-v1", @@ -310,9 +312,12 @@ def ensure_manifest(self, manifest: dict) -> None: self._client.set(self.EPOCH, 0, nx=True) self._manifest = actual - def read_manifest(self): + def read_manifest(self) -> dict[str, Any] | None: raw = self._client.get(self.ROOT + ":manifest") - return json.loads(raw) if raw else None + if not raw: + return None + parsed: Any = json.loads(raw) + return parsed if isinstance(parsed, dict) else None def revision(self) -> int: self._healthy() @@ -327,11 +332,11 @@ def _healthy(self) -> None: ): raise ValueError("Shared generation changed or lost its manifest") - def events(self, start: int, end: int, limit: int = 256): + def events(self, start: int, end: int, limit: int = 256) -> list[tuple[int, dict[str, Any]]]: """Return a bounded, ordered slice of authoritative mutations.""" self._healthy() return [ - (int(event_id.split("-", 1)[0]), fields) + (int(str(event_id).split("-", 1)[0]), dict(fields)) for event_id, fields in self._client.xrange( self.JOURNAL, min=f"{start}-0", max=f"{end}-0", count=limit ) @@ -341,7 +346,7 @@ def epoch(self) -> int: self._healthy() return int(self._client.get(self.EPOCH) or 0) - def store( + def store_shared( self, entry: CacheEntry, *, @@ -364,9 +369,9 @@ def store( "created_at": str(entry.created_at), "expires_at": str(entry.expires_at) if entry.expires_at is not None else "", "hit_count": str(entry.hit_count), - "model": entry.model, - "system_prompt": entry.system_prompt, - "temperature": str(entry.temperature), + "model": "", + "system_prompt": "", + "temperature": "-1.0", "tags": json.dumps(list(entry.tags or [])), "metadata": json.dumps(entry.metadata, sort_keys=True, allow_nan=False), } @@ -375,7 +380,7 @@ def store( receipt_key = self.ROOT + ":receipt:" + uuid4().hex payload_digest = hashlib.sha256(payload.encode()).hexdigest() - def commit(): + def commit() -> Any: return self._client.eval( _STORE, 14, @@ -414,22 +419,26 @@ def commit(): import redis try: - result = commit() + result: Any = commit() except (redis.exceptions.ConnectionError, redis.exceptions.TimeoutError): try: result = commit() except (redis.exceptions.ConnectionError, redis.exceptions.TimeoutError) as exc: raise CacheOperationError("store_indeterminate") from exc - if isinstance(result, list): + if isinstance(result, (list, tuple)): return int(result[0]), int(result[1]) if int(result) == -2: return -1, None return int(result), None + def store(self, entry: CacheEntry) -> int: + identifier, _ = self.store_shared(entry) + return identifier + def acquire_flight(self, exact_key: str, ttl_ms: int) -> tuple[str, str | None]: self._healthy() key = f"{self.ROOT}:flight:{exact_key}" - result = self._client.eval( + result: Any = self._client.eval( _ACQUIRE_FLIGHT, 6, self.ROOT + ":manifest", @@ -445,7 +454,7 @@ def acquire_flight(self, exact_key: str, ttl_ms: int) -> tuple[str, str | None]: ) if int(result[0]) == -1: raise SingleflightCapacityError("Shared lease capacity reached") - return key, result[1] if int(result[0]) == 1 else None + return key, str(result[1]) if int(result[0]) == 1 else None def finish_flight(self, key: str, token: str, outcome: str) -> bool: if outcome not in ("error", "cancelled", "non_shareable"): @@ -463,7 +472,8 @@ def finish_flight(self, key: str, token: str, outcome: str) -> bool: ) def flight_outcome(self, key: str) -> str | None: - return self._client.get(key + ":terminal") + val = self._client.get(key + ":terminal") + return str(val) if val is not None else None def renew_flight(self, key: str, token: str, ttl_ms: int) -> bool: return bool( @@ -511,6 +521,8 @@ def retrieve(self, entry_id: int) -> CacheEntry | None: return entry def validate_hit(self, entry: CacheEntry, *, expected_epoch: int) -> bool: + if entry.id is None: + return False return bool( self._client.eval( _VALIDATE, @@ -530,7 +542,7 @@ def validate_hit(self, entry: CacheEntry, *, expected_epoch: int) -> bool: ) ) - def _remove(self, ids): + def _remove(self, ids: Iterable[int | str]) -> int: total = 0 iterator = iter(ids) while batch := list(islice(iterator, 256)): @@ -578,7 +590,7 @@ def _invalidate( f"{membership}:{watermark}:{before}:{count}".encode() ).hexdigest() - def commit(): + def commit() -> Any: return self._client.eval( _INVALIDATE, 9, @@ -651,22 +663,20 @@ def update_hit_count(self, entry_id: int) -> None: def count(self, namespace: str) -> int: return sum( self.retrieve(int(identifier)) is not None - for identifier in self._client.smembers(self._namespace_key(namespace)) + for identifier in list(self._client.smembers(self._namespace_key(namespace))) ) def evict_oldest(self, namespace: str, count: int) -> int: entries = sorted(self.list_entries(namespace), key=lambda item: item.created_at) - return sum(self.delete(entry.id) for entry in entries[:count]) + return sum(self.delete(entry.id) for entry in entries[:count] if entry.id is not None) def get_expired(self, namespace: str) -> list[int]: return [] def sweep_expired(self) -> int: """Bound cleanup of expired and logically invalidated shared entries.""" - cursor, ids = self._client.sscan( - self.ALL_IDS, cursor=getattr(self, "_scan_cursor", 0), count=256 - ) - self._scan_cursor = cursor + cursor, ids = self._client.sscan(self.ALL_IDS, cursor=self._scan_cursor, count=256) + self._scan_cursor = int(cursor) invalid = [identifier for identifier in ids if self.retrieve(int(identifier)) is None] self._remove(invalid) return len(invalid) diff --git a/src/longcache/backends/sqlite.py b/src/longcache/backends/sqlite.py index 1d555b2..8c3dff6 100755 --- a/src/longcache/backends/sqlite.py +++ b/src/longcache/backends/sqlite.py @@ -13,7 +13,7 @@ import sqlite3 import threading import time -from typing import Optional +from typing import Any, Optional import numpy as np @@ -151,16 +151,19 @@ def __init__( self.close() raise - def read_manifest(self): + def read_manifest(self) -> dict[str, Any] | None: with self._lock: row = ( self._get_connection() .execute("SELECT data FROM cache_manifest WHERE id=1") .fetchone() ) - return json.loads(row[0]) if row else None + if row is None: + return None + parsed: Any = json.loads(row[0]) + return parsed if isinstance(parsed, dict) else None - def ensure_manifest(self, manifest: dict) -> None: + def ensure_manifest(self, manifest: dict[str, Any]) -> None: from longcache.core.identity import canonical encoded = canonical(manifest) @@ -168,7 +171,8 @@ def ensure_manifest(self, manifest: dict) -> None: conn.execute( "INSERT OR IGNORE INTO cache_manifest (id, data) VALUES (1, ?)", (encoded,) ) - if conn.execute("SELECT data FROM cache_manifest WHERE id=1").fetchone()[0] != encoded: + row = conn.execute("SELECT data FROM cache_manifest WHERE id=1").fetchone() + if row is None or row[0] != encoded: raise ValueError("Incompatible embedding or storage generation; use a new store") def _get_connection(self) -> sqlite3.Connection: @@ -190,7 +194,7 @@ def close(self) -> None: self._connection.close() self._connection = None - def _entry_to_row(self, entry: CacheEntry) -> dict: + def _entry_to_row(self, entry: CacheEntry) -> dict[str, Any]: """Convert a CacheEntry to a row dict for database insertion.""" vector_bytes = entry.vector.tobytes() if entry.vector is not None else b"" response_json = json.dumps(entry.response, allow_nan=False) @@ -203,13 +207,13 @@ def _entry_to_row(self, entry: CacheEntry) -> dict: "created_at": entry.created_at, "expires_at": entry.expires_at, "hit_count": entry.hit_count, - "model": entry.model, - "system_prompt": entry.system_prompt, - "temperature": entry.temperature, + "model": "", + "system_prompt": "", + "temperature": -1.0, "metadata": json.dumps(entry.metadata, sort_keys=True, allow_nan=False), } - def _row_to_entry(self, row) -> CacheEntry: + def _row_to_entry(self, row: sqlite3.Row) -> CacheEntry: """Convert a database row to a CacheEntry.""" vector = None if row["vector"]: @@ -222,14 +226,6 @@ def _row_to_entry(self, row) -> CacheEntry: except (json.JSONDecodeError, TypeError): pass # Keep as string if not valid JSON - model = row["model"] if "model" in row.keys() else "" - system_prompt = row["system_prompt"] if "system_prompt" in row.keys() else "" - temperature = ( - float(row["temperature"]) - if "temperature" in row.keys() and row["temperature"] is not None - else -1.0 - ) - tags = [] try: conn = self._get_connection() @@ -249,9 +245,6 @@ def _row_to_entry(self, row) -> CacheEntry: created_at=row["created_at"], expires_at=row["expires_at"], hit_count=row["hit_count"], - model=model, - system_prompt=system_prompt, - temperature=temperature, tags=tags, metadata=json.loads(row["metadata"]), ) @@ -277,7 +270,9 @@ def store(self, entry: CacheEntry) -> int: """, row, ) - entry_id = cursor.lastrowid + if cursor.lastrowid is None: + raise RuntimeError("Failed to obtain lastrowid after inserting cache entry") + entry_id = int(cursor.lastrowid) if entry.tags: tag_rows = [(entry_id, tag) for tag in entry.tags] conn.executemany( @@ -402,7 +397,8 @@ def clear(self) -> int: with self._lock: conn = self._get_connection() conn.execute("DELETE FROM cache_tags") - count = conn.execute("SELECT COUNT(*) FROM cache_entries").fetchone()[0] + row = conn.execute("SELECT COUNT(*) FROM cache_entries").fetchone() + count = int(row[0]) if row else 0 conn.execute("DELETE FROM cache_entries") conn.commit() return count @@ -429,7 +425,7 @@ def count(self, namespace: str) -> int: """, (namespace, now), ).fetchone() - return row[0] + return int(row[0]) if row else 0 def list_entries(self, namespace: str) -> list[CacheEntry]: """ @@ -510,7 +506,7 @@ def evict_oldest(self, namespace: str, count: int) -> int: (namespace, count), ) conn.commit() - return cursor.rowcount + return int(cursor.rowcount) def get_expired(self, namespace: str) -> list[int]: """ @@ -532,9 +528,9 @@ def get_expired(self, namespace: str) -> list[int]: """, (namespace, now), ).fetchall() - return [row[0] for row in rows] + return [int(row[0]) for row in rows] - def export_all(self) -> list[dict]: + def export_all(self) -> list[dict[str, Any]]: """ Export all cache entries as dictionaries. @@ -574,7 +570,7 @@ def sweep_expired(self) -> int: "DELETE FROM cache_entries WHERE expires_at IS NOT NULL AND expires_at <= ?", (now,) ) conn.commit() - return cursor.rowcount + return int(cursor.rowcount) def increment_hits(self) -> None: with self._lock: @@ -592,10 +588,10 @@ def increment_misses(self, saved_ms: int = 0) -> None: ) conn.commit() - def load_stats(self) -> dict: + def load_stats(self) -> dict[str, int]: with self._lock: rows = self._get_connection().execute("SELECT key, value FROM cache_stats").fetchall() - return {row["key"]: row["value"] for row in rows} + return {str(row["key"]): int(row["value"]) for row in rows} def __repr__(self) -> str: return f"SQLiteCacheBackend(db_path={self._db_path!r})" diff --git a/src/longcache/cli.py b/src/longcache/cli.py index 6cef6fb..4ac97e0 100755 --- a/src/longcache/cli.py +++ b/src/longcache/cli.py @@ -1,17 +1,22 @@ """Scoped cache inspection and explicit local operator maintenance.""" +from __future__ import annotations + from dataclasses import asdict import json from pathlib import Path +from typing import Any, Callable, TypeVar, cast import click from longcache import CacheConfig, Instruction, LongCache, RequestContext, RequestScope, __version__ +FC = TypeVar("FC", bound=Callable[..., Any]) + -def _get_cache(config=None, *, metadata_only=False): +def _get_cache(config: CacheConfig | None = None, *, metadata_only: bool = False) -> LongCache: return LongCache(config or CacheConfig(), metadata_only=metadata_only) -def _context(path): +def _context(path: str | Path | None) -> RequestContext | None: if path is None: return None try: @@ -24,7 +29,7 @@ def _context(path): raise click.ClickException("Invalid context file") from exc -def common(fn): +def common(fn: FC) -> FC: for option in [ click.option("--db-path", default=".longcache/v2/cache.db", show_default=True), click.option("--namespace", default="default"), @@ -44,22 +49,33 @@ def common(fn): return fn -def _open(db_path, namespace, scope_policy, metadata_only): +def _open( + db_path: str, + namespace: str, + scope_policy: str, + metadata_only: bool, +) -> LongCache: return _get_cache( - CacheConfig(db_path=db_path, namespace=namespace, scope_policy=scope_policy), + CacheConfig(db_path=db_path, namespace=namespace, scope_policy=cast(Any, scope_policy)), metadata_only=metadata_only, ) @click.group() @click.version_option(version=__version__, prog_name="longcache") -def main(): +def main() -> None: """Inspect scoped cache decisions or explicitly maintain an owned v0.2 store.""" @main.command() @common -def stats(db_path, namespace, scope_policy, context_file, operator): +def stats( + db_path: str, + namespace: str, + scope_policy: str, + context_file: str | None, + operator: bool, +) -> None: """Show counts for the declared scope; process hit/miss counters reset on restart.""" with _open(db_path, namespace, scope_policy, True) as cache: result = cache.stats(context=_context(context_file), operator=operator) @@ -70,7 +86,14 @@ def stats(db_path, namespace, scope_policy, context_file, operator): @main.command() @common @click.option("--query", required=True) -def inspect(db_path, namespace, scope_policy, context_file, operator, query): +def inspect( + db_path: str, + namespace: str, + scope_policy: str, + context_file: str | None, + operator: bool, + query: str, +) -> None: """Explain a lookup without recording a hit. Context is always required.""" if operator: raise click.UsageError( @@ -91,7 +114,15 @@ def inspect(db_path, namespace, scope_policy, context_file, operator, query): @common @click.option("--all", "all_partitions", is_flag=True) @click.option("--yes", is_flag=True, help="Confirm deletion of the selected owned entries.") -def clear(db_path, namespace, scope_policy, context_file, operator, all_partitions, yes): +def clear( + db_path: str, + namespace: str, + scope_policy: str, + context_file: str | None, + operator: bool, + all_partitions: bool, + yes: bool, +) -> None: """Delete one exact context, or explicitly delete every owned partition.""" ctx = _context(context_file) if all_partitions and not operator: @@ -111,7 +142,15 @@ def clear(db_path, namespace, scope_policy, context_file, operator, all_partitio @common @click.option("--format", "output_format", type=click.Choice(["json"]), default="json") @click.option("--output", "-o", type=click.Path(dir_okay=False)) -def export(db_path, namespace, scope_policy, context_file, operator, output_format, output): +def export( + db_path: str, + namespace: str, + scope_policy: str, + context_file: str | None, + operator: bool, + output_format: str, + output: str | None, +) -> None: """Export raw owned payloads only for an explicit context or trusted operator.""" ctx = _context(context_file) if ctx is None and not operator: diff --git a/src/longcache/config.py b/src/longcache/config.py index 0bccb34..afea3cd 100755 --- a/src/longcache/config.py +++ b/src/longcache/config.py @@ -7,8 +7,14 @@ from __future__ import annotations +from collections.abc import Mapping +from dataclasses import asdict, dataclass +import math import os -from dataclasses import dataclass, asdict +import re +from typing import Any + +from .core.types import BackendName, DeploymentMode, ErrorPolicy, FaissIndexType, ScopePolicy @dataclass @@ -37,7 +43,7 @@ class CacheConfig: max_entries: int = 10_000 namespace: str = "default" embedder_model: str = "BAAI/bge-small-en-v1.5" - backend: str = "sqlite" + backend: BackendName = "sqlite" db_path: str = ".longcache/v2/cache.db" redis_url: str = "redis://localhost:6379/0" track_stats: bool = True @@ -45,10 +51,10 @@ class CacheConfig: singleflight_wait_timeout_seconds: float = 30.0 singleflight_max_inflight: int = 1024 singleflight_max_waiters_per_key: int = 256 - cache_error_policy: str = "raise" + cache_error_policy: ErrorPolicy = "raise" backend_timeout_seconds: float = 2.0 shutdown_timeout_seconds: float = 5.0 - deployment_mode: str = "single_process" + deployment_mode: DeploymentMode = "single_process" shared_sync_timeout_seconds: float = 5.0 shared_max_entries: int = 50_000 shared_max_leases: int = 1024 @@ -59,10 +65,10 @@ class CacheConfig: shared_max_tag_chars: int = 256 distributed_lease_seconds: float = 15.0 distributed_renew_interval_seconds: float = 5.0 - faiss_index_type: str = "flat" + faiss_index_type: FaissIndexType = "flat" hnsw_m: int = 32 hnsw_ef_search: int = 64 - scope_policy: str = "private" + scope_policy: ScopePolicy = "private" cache_id: str = "default" generation: str = "default" policy_version: str = "safe-reuse-v1" @@ -70,15 +76,9 @@ class CacheConfig: max_query_chars: int = 16384 max_capture_bytes: int = 1048576 embedder_revision: str = "" - model: str = "" - system_prompt: str = "" - temperature: float = -1.0 - def __post_init__(self): + def __post_init__(self) -> None: """Validate configuration values after initialization.""" - import re - import math - for name in ("singleflight_enabled", "track_stats"): if type(getattr(self, name)) is not bool: raise ValueError(f"{name} must be a boolean") @@ -139,10 +139,13 @@ def __post_init__(self): raise ValueError(f"ttl_seconds must be >= 0, got {self.ttl_seconds}") if self.max_entries < 1: raise ValueError(f"max_entries must be >= 1, got {self.max_entries}") - if self.backend not in ("sqlite", "redis", "async_redis", "memory"): + if str(self.backend) == "async_redis": raise ValueError( - f"backend must be one of 'sqlite', 'redis', 'async_redis', 'memory', " - f"got '{self.backend}'" + "async_redis is not supported through LongCache; use AsyncRedisCacheBackend directly or configure backend='redis'" + ) + if self.backend not in ("sqlite", "redis", "memory"): + raise ValueError( + f"backend must be one of 'sqlite', 'redis', 'memory', got '{self.backend}'" ) if self.faiss_index_type not in ("flat", "hnsw"): raise ValueError("faiss_index_type must be 'flat' or 'hnsw'") @@ -150,7 +153,7 @@ def __post_init__(self): raise ValueError("hnsw_m and hnsw_ef_search must be positive") @classmethod - def from_dict(cls, data: dict) -> CacheConfig: + def from_dict(cls, data: Mapping[str, Any]) -> CacheConfig: """ Create a CacheConfig from a dictionary. @@ -166,7 +169,7 @@ def from_dict(cls, data: dict) -> CacheConfig: unknown = set(data) - valid_keys if unknown: raise ValueError(f"Unknown configuration keys: {sorted(unknown)}") - return cls(**data) + return cls(**dict(data)) @classmethod def from_env(cls, prefix: str = "LONGCACHE_") -> CacheConfig: @@ -184,28 +187,53 @@ def from_env(cls, prefix: str = "LONGCACHE_") -> CacheConfig: CacheConfig instance with values from environment variables. """ field_types = {f.name: f.type for f in cls.__dataclass_fields__.values()} - env_values = {} + env_values: dict[str, Any] = {} for field_name in cls.__dataclass_fields__: env_key = f"{prefix}{field_name.upper()}" env_val = os.environ.get(env_key) if env_val is not None: - # Cast to appropriate type field_type = field_types[field_name] - if field_type == "float": + if field_type == "float" or field_name == "similarity_threshold": env_values[field_name] = float(env_val) - elif field_type == "int": + elif field_type == "int" or field_name in ( + "ttl_seconds", + "max_entries", + "verification_budget", + "max_query_chars", + "max_capture_bytes", + "hnsw_m", + "hnsw_ef_search", + "shared_max_entries", + "shared_max_leases", + "shared_max_receipts", + "shared_max_partitions", + "shared_max_distinct_tags", + "shared_max_tags_per_entry", + "shared_max_tag_chars", + "singleflight_max_inflight", + "singleflight_max_waiters_per_key", + ): env_values[field_name] = int(env_val) - elif field_type == "bool": + elif field_type == "bool" or field_name in ("track_stats", "singleflight_enabled"): if env_val.lower() not in ("true", "1", "yes", "false", "0", "no"): raise ValueError(f"Invalid boolean for {env_key}") env_values[field_name] = env_val.lower() in ("true", "1", "yes") + elif field_name in ( + "backend_timeout_seconds", + "shutdown_timeout_seconds", + "shared_sync_timeout_seconds", + "distributed_lease_seconds", + "distributed_renew_interval_seconds", + "singleflight_wait_timeout_seconds", + ): + env_values[field_name] = float(env_val) else: env_values[field_name] = env_val return cls(**env_values) - def to_dict(self) -> dict: + def to_dict(self) -> dict[str, Any]: """ Convert the configuration to a dictionary. diff --git a/src/longcache/core/base_embedder.py b/src/longcache/core/base_embedder.py index 4be6b4b..63bfa21 100644 --- a/src/longcache/core/base_embedder.py +++ b/src/longcache/core/base_embedder.py @@ -1,9 +1,15 @@ """Abstract base class for all embedder implementations.""" from __future__ import annotations + from abc import ABC, abstractmethod +from typing import TYPE_CHECKING import numpy as np +if TYPE_CHECKING: + from .embedding_contract import EmbeddingDescriptor + from .types import Float32Vector + class BaseEmbedder(ABC): """ @@ -14,10 +20,10 @@ class BaseEmbedder(ABC): the FAISS index and must not change between calls. """ - thread_safe = False + thread_safe: bool = False @property - def descriptor(self): + def descriptor(self) -> EmbeddingDescriptor: """Declare a stable model/revision contract for persisted vector compatibility.""" raise ValueError("Custom embedders must supply an EmbeddingDescriptor") @@ -28,7 +34,7 @@ def dimension(self) -> int: ... @abstractmethod - def embed(self, text: str) -> np.ndarray: + def embed(self, text: str) -> Float32Vector: """ Embed a single text string. @@ -37,7 +43,7 @@ def embed(self, text: str) -> np.ndarray: """ ... - def embed_batch(self, texts: list[str]) -> np.ndarray: + def embed_batch(self, texts: list[str]) -> Float32Vector: """ Embed multiple texts. Default: loop over embed(). Override for batched efficiency. @@ -45,8 +51,6 @@ def embed_batch(self, texts: list[str]) -> np.ndarray: Returns: float32 array of shape (len(texts), dimension). """ - import numpy as np - if not texts: return np.empty((0, self.dimension), dtype=np.float32) return np.stack([self.embed(t) for t in texts]) diff --git a/src/longcache/core/cache.py b/src/longcache/core/cache.py index 167554e..4edbf0f 100755 --- a/src/longcache/core/cache.py +++ b/src/longcache/core/cache.py @@ -1,45 +1,54 @@ """Scoped semantic reuse with shared policy across sync and async entry points.""" from __future__ import annotations + import asyncio from collections import Counter, OrderedDict +from collections.abc import Sequence from dataclasses import asdict, replace from functools import wraps import inspect import threading import time +from typing import Any, Awaitable, Callable, ParamSpec, TypeVar, cast, overload -from longcache.config import CacheConfig from longcache.backends.memory import InMemoryCacheBackend from longcache.backends.sqlite import SQLiteCacheBackend +from longcache.config import CacheConfig + from .base_embedder import BaseEmbedder -from .decision import LookupDecision, StoreDecision -from .embedding_contract import EmbeddingGuard, validate_vector +from .decision import InspectionReport, LookupDecision, StoreDecision +from .distributed_singleflight import DistributedExecution from .embedder import LocalEmbedder +from .embedding_contract import EmbeddingGuard, validate_vector from .entry import CacheEntry -from .identity import Instruction, RequestContext, canonical, digest +from .errors import CacheClosedError, CacheOperationError +from .execution import Execution +from .identity import RequestContext, canonical, digest from .index import SimilarityIndex from .lifecycle import Lifecycle, managed -from .errors import CacheClosedError -from .errors import CacheOperationError from .observability import Observability -from .reliability import dependency, decision_operation, operational, execution_operation -from .execution import Execution -from .distributed_singleflight import DistributedExecution from .policy import admit, decode_response, verify +from .reliability import decision_operation, dependency, execution_operation, operational +from .types import AdmissionHook, CachedCallable, MetricsSnapshot, StatsSnapshot, VerificationHook + +T = TypeVar("T") +F = TypeVar("F", bound=Callable[..., Any]) +P = ParamSpec("P") +R = TypeVar("R") class LongCache: def __init__( self, - config=None, + config: CacheConfig | None = None, *, embedder: BaseEmbedder | None = None, - admit=None, - verifier=None, + admit: AdmissionHook | None = None, + verifier: VerificationHook | None = None, context: RequestContext | None = None, metadata_only: bool = False, - ): + ) -> None: self.config = config or CacheConfig() for hook in (admit, verifier): if hook is not None and ( @@ -52,10 +61,11 @@ def __init__( raise ValueError("context must be a RequestContext") self._metadata_only = metadata_only self.context = context - self._admit, self._verifier = admit, verifier + self._admit: Callable[..., Any] | None = admit + self._verifier: Callable[..., Any] | None = verifier self._mutation_lock = threading.RLock() - self._stats = Counter() - self._partition_stats = OrderedDict() + self._stats: Counter[str] = Counter() + self._partition_stats: OrderedDict[str, Counter[str]] = OrderedDict() self._stats_started_at = time.time() self._stats_evictions = 0 self._closed = False @@ -67,21 +77,22 @@ def __init__( self._observability = Observability(self.config.track_stats, self.config.backend) self._ready = False self._stop_event = threading.Event() - self.embedder = ( + self.embedder: BaseEmbedder = ( embedder if embedder is not None else LocalEmbedder(self.config.embedder_model, self.config.embedder_revision) ) self._embedding = EmbeddingGuard(self.embedder) - self.backend = self._init_backend() + self.backend: Any = self._init_backend() self.index = SimilarityIndex(self.backend, self.config) self._lifecycle = Lifecycle( self._finalize_resources, self._revoke_flights, self.config.shutdown_timeout_seconds ) self._execution = Execution(self) - self._distributed_execution = ( + self._distributed_execution: DistributedExecution | None = ( DistributedExecution(self) if self.config.deployment_mode == "shared" else None ) + self._fingerprint: str = "" # Metadata-only construction does not load/download model weights. try: if self._dependency("backend", self.backend.list_all_entries): @@ -94,9 +105,10 @@ def __init__( if hasattr(self.backend, "close"): self.backend.close() raise - self._sweep_thread = None + self._sweep_thread: threading.Thread | None = None if ( not metadata_only + and self.config.backend != "memory" and (self.config.ttl_seconds or self.config.deployment_mode == "shared") and hasattr(self.backend, "sweep_expired") ): @@ -105,7 +117,7 @@ def __init__( ) self._sweep_thread.start() - def _init_backend(self): + def _init_backend(self) -> Any: c = self.config if c.backend == "memory": return InMemoryCacheBackend(c.max_entries) @@ -142,13 +154,13 @@ def _init_backend(self): ) raise ValueError("async_redis is not supported through LongCache; use a supported backend") - def _ensure_open(self): + def _ensure_open(self) -> None: self._lifecycle.ensure_open() - def _dependency(self, stage, function, *args, **kwargs): + def _dependency(self, stage: str, function: Callable[..., T], *args: Any, **kwargs: Any) -> T: return dependency(self, stage, function, *args, **kwargs) - def _ensure_ready(self): + def _ensure_ready(self) -> None: with self._mutation_lock: self._ensure_open() if not self._coherent: @@ -191,10 +203,10 @@ def _ensure_ready(self): self._shared_revision = before self._ready = True - def _rebuild_index_from_backend(self): + def _rebuild_index_from_backend(self) -> None: if self._metadata_only: return - entries = [] + entries: list[CacheEntry] = [] for entry in self._dependency("backend", self.backend.list_all_entries): metadata = entry.metadata if metadata.get("format") != 2 or metadata.get("embedding") != self._fingerprint: @@ -206,7 +218,7 @@ def _rebuild_index_from_backend(self): entries.append(entry) self._dependency("index", self.index.rebuild, entries) - def _synchronize_shared(self): + def _synchronize_shared(self) -> None: """Never search a local index without checking authoritative Redis state.""" if self.config.deployment_mode != "shared": return @@ -216,7 +228,7 @@ def _synchronize_shared(self): before = self._dependency("backend", self.backend.revision) if self._shared_revision == before: return - if self._shared_revision >= 0 and self._shared_revision < before: + if 0 <= self._shared_revision < before: cursor = self._shared_revision complete = True while cursor < before and complete: @@ -260,24 +272,18 @@ def _synchronize_shared(self): if time.monotonic() >= deadline: raise CacheOperationError("synchronization") - def _prepare(self, query, context, namespace, legacy, *, _initialize=True): + def _prepare( + self, + query: str, + context: RequestContext | None, + namespace: str | None, + *, + _initialize: bool = True, + ) -> tuple[tuple[RequestContext, str, str] | None, str | None]: self._ensure_open() - if context is not None and any(value is not None for value in legacy.values()): - raise ValueError("Legacy context keywords cannot override RequestContext") ctx = context if context is not None else self.context if not isinstance(ctx, RequestContext): return None, "missing_identity" - if any(value is not None for value in legacy.values()): - updates = {} - if legacy.get("model") is not None: - updates["model"] = legacy["model"] - if legacy.get("system_prompt") is not None: - updates["instructions"] = (Instruction("system", legacy["system_prompt"]),) - if legacy.get("temperature") is not None: - updates["generation"] = dict(ctx.generation) | { - "temperature": legacy["temperature"] - } - ctx = replace(ctx, **updates) policy = self.config.scope_policy if policy != "public" and ( ctx.scope is None or (policy == "private" and not ctx.scope.user_id) @@ -306,49 +312,52 @@ def _prepare(self, query, context, namespace, legacy, *, _initialize=True): partition = "v2:" + digest(record) return (ctx, partition, identity), None - def _account(self, partition, outcome): + def _account(self, partition: str, outcome: str) -> None: if self.config.track_stats: self._stats[outcome] += 1 if partition not in self._partition_stats: if len(self._partition_stats) >= 1024: self._partition_stats.popitem(last=False) self._stats_evictions += 1 - self._partition_stats[partition] = Counter(_started_at=time.time()) + self._partition_stats[partition] = Counter() + self._partition_stats[partition]["_started_at"] = int(time.time()) self._partition_stats.move_to_end(partition) self._partition_stats[partition][outcome] += 1 + def lookup( + self, + query: str, + namespace: str | None = None, + *, + context: RequestContext | None = None, + ) -> LookupDecision: + return cast(LookupDecision, self._lookup(query, namespace, context=context)) + @managed @decision_operation("lookup") - def lookup( + def _lookup( self, - query, - namespace=None, + query: str, + namespace: str | None = None, *, - context=None, - model=None, - system_prompt=None, - temperature=None, - _inspect=False, - _record=True, - _cancel=None, - ): + context: RequestContext | None = None, + _inspect: bool = False, + _record: bool = True, + _cancel: Any = None, + _internal: bool = False, + ) -> LookupDecision: _cancel = _cancel or self._lifecycle.current if self._metadata_only: raise ValueError("Vector operations require a normal cache instance") - prepared, reason = self._prepare( - query, - context, - namespace, - dict(model=model, system_prompt=system_prompt, temperature=temperature), - ) + prepared, reason = self._prepare(query, context, namespace) threshold = self.config.similarity_threshold - if reason: - return LookupDecision("BYPASS", reason, threshold=threshold) + if reason or prepared is None: + return LookupDecision("BYPASS", reason or "missing_identity", threshold=threshold) ctx, partition, identity = prepared self._synchronize_shared() with self._mutation_lock: if self.index.size == 0: - if not _inspect and _record and not _cancel.is_set(): + if not _inspect and _record and not (_cancel and _cancel.is_set()): self._account(partition, "misses") return LookupDecision("MISS", "no_candidate", threshold=threshold) vector = self._dependency("embedding", self._embedding.embed, query) @@ -452,74 +461,97 @@ def lookup( self._account(partition, "misses") return LookupDecision("MISS", reason, similarity=last_score, threshold=threshold) - async def _async(self, method, *args, **kwargs): + async def _async(self, method: Callable[..., T], *args: Any, **kwargs: Any) -> T: return await self._lifecycle.call(method, *args, **kwargs) - async def alookup(self, query, namespace=None, **kwargs): - return await self._async(self.lookup, query, namespace, **kwargs) + async def alookup( + self, + query: str, + namespace: str | None = None, + *, + context: RequestContext | None = None, + ) -> LookupDecision: + return await self._async(self.lookup, query, namespace, context=context) + + async def _alookup( + self, + query: str, + namespace: str | None = None, + *, + context: RequestContext | None = None, + _inspect: bool = False, + _record: bool = True, + _cancel: Any = None, + _internal: bool = False, + ) -> LookupDecision: + kwargs: dict[str, Any] = { + "context": context, + "_inspect": _inspect, + "_record": _record, + "_internal": _internal, + } + if _cancel is not None: + kwargs["_cancel"] = _cancel + return await self._async(self._lookup, query, namespace, **kwargs) def get( self, - query, - namespace=None, - model=None, - system_prompt=None, - temperature=None, - track_miss=True, + query: str, + namespace: str | None = None, *, - context=None, - ): - return self.lookup( - query, - namespace, - context=context, - model=model, - system_prompt=system_prompt, - temperature=temperature, - ).response + context: RequestContext | None = None, + ) -> Any: + return self.lookup(query, namespace, context=context).response async def aget( self, - query, - namespace=None, - model=None, - system_prompt=None, - temperature=None, - track_miss=True, + query: str, + namespace: str | None = None, *, - context=None, - ): - return ( - await self.alookup( + context: RequestContext | None = None, + ) -> Any: + return (await self.alookup(query, namespace, context=context)).response + + def set( + self, + query: str, + response: Any, + namespace: str | None = None, + ttl: int | None = None, + tags: Sequence[str] | None = None, + *, + context: RequestContext | None = None, + ) -> StoreDecision: + return cast( + StoreDecision, + self._set( query, + response, namespace, + ttl=ttl, + tags=tags, context=context, - model=model, - system_prompt=system_prompt, - temperature=temperature, - ) - ).response + ), + ) @managed @decision_operation("store") - def set( + def _set( self, - query, - response, - namespace=None, - ttl=None, - tags=None, - model=None, - system_prompt=None, - temperature=None, + query: str, + response: Any, + namespace: str | None = None, + ttl: int | None = None, + tags: Any = None, *, - context=None, - _cancel=None, - _expected_epoch=None, - _expected_shared_epoch=None, - _lease_key="", - _lease_token="", - ): + context: RequestContext | None = None, + _cancel: Any = None, + _expected_epoch: int | None = None, + _expected_shared_epoch: int | None = None, + _lease_key: str = "", + _lease_token: str = "", + _internal: bool = False, + ) -> StoreDecision: _cancel = _cancel or self._lifecycle.current if _expected_epoch is None: with self._mutation_lock: @@ -528,14 +560,9 @@ def set( _expected_shared_epoch = self._dependency("backend", self.backend.epoch) if self._metadata_only: raise ValueError("Vector operations require a normal cache instance") - prepared, reason = self._prepare( - query, - context, - namespace, - dict(model=model, system_prompt=system_prompt, temperature=temperature), - ) - if reason: - return StoreDecision("REJECTED", reason) + prepared, reason = self._prepare(query, context, namespace) + if reason or prepared is None: + return StoreDecision("REJECTED", reason or "missing_identity") ctx, partition, identity = prepared payload, decision = self._dependency("admission", admit, response, ctx, self._admit) self._observability.count( @@ -587,7 +614,7 @@ def set( if self.config.deployment_mode == "shared": identifier, commit_revision = self._dependency( "store", - self.backend.store, + self.backend.store_shared, entry, expected_epoch=_expected_shared_epoch, lease_key=_lease_key, @@ -624,86 +651,208 @@ def set( raise return StoreDecision("STORED", "accepted", identifier) - async def aset(self, query, response, namespace=None, **kwargs): - return await self._async(self.set, query, response, namespace, **kwargs) + async def aset( + self, + query: str, + response: Any, + namespace: str | None = None, + ttl: int | None = None, + tags: Sequence[str] | None = None, + *, + context: RequestContext | None = None, + ) -> StoreDecision: + return await self._async( + self.set, + query, + response, + namespace, + ttl=ttl, + tags=tags, + context=context, + ) + + async def _aset( + self, + query: str, + response: Any, + namespace: str | None = None, + ttl: int | None = None, + tags: Any = None, + *, + context: RequestContext | None = None, + _cancel: Any = None, + _expected_epoch: int | None = None, + _expected_shared_epoch: int | None = None, + _lease_key: str = "", + _lease_token: str = "", + _internal: bool = False, + ) -> StoreDecision: + kwargs: dict[str, Any] = { + "ttl": ttl, + "tags": tags, + "context": context, + "_expected_epoch": _expected_epoch, + "_expected_shared_epoch": _expected_shared_epoch, + "_lease_key": _lease_key, + "_lease_token": _lease_token, + "_internal": _internal, + } + if _cancel is not None: + kwargs["_cancel"] = _cancel + return await self._async(self._set, query, response, namespace, **kwargs) - @execution_operation def get_or_compute( self, - query, - compute, + query: str, + compute: Callable[[], T], *, - context=None, - namespace=None, - ttl=None, - tags=None, - _entry_point="manual", - ): + context: RequestContext | None = None, + namespace: str | None = None, + ttl: int | None = None, + tags: Sequence[str] | None = None, + ) -> T: """Reuse a verified answer or coordinate one exact scoped computation.""" + return self._run_compute( + query, + compute, + context=context, + namespace=namespace, + ttl=ttl, + tags=tags, + _entry_point="manual", + ) + + @execution_operation + def _run_compute( + self, + query: str, + compute: Callable[[], T], + *, + context: RequestContext | None = None, + namespace: str | None = None, + ttl: int | None = None, + tags: Any = None, + _entry_point: str = "manual", + ) -> T: if self._distributed_execution is not None: - return self._distributed_execution.run( + return cast( + T, + self._distributed_execution.run( + query, + compute, + context=context, + namespace=namespace, + ttl=ttl, + tags=tags, + entry_point=_entry_point, + ), + ) + return cast( + T, + self._execution.run( query, compute, context=context, namespace=namespace, ttl=ttl, tags=tags, - entry_point=_entry_point, - ) - return self._execution.run( + _entry_point=_entry_point, + ), + ) + + async def aget_or_compute( + self, + query: str, + compute: Callable[[], Awaitable[T]], + *, + context: RequestContext | None = None, + namespace: str | None = None, + ttl: int | None = None, + tags: Sequence[str] | None = None, + ) -> T: + """Async equivalent; followers await without occupying worker threads.""" + return await self._arun_compute( query, compute, context=context, namespace=namespace, ttl=ttl, tags=tags, - _entry_point=_entry_point, + _entry_point="manual", ) @execution_operation - async def aget_or_compute( + async def _arun_compute( self, - query, - compute, + query: str, + compute: Callable[[], Awaitable[T]], *, - context=None, - namespace=None, - ttl=None, - tags=None, - _entry_point="manual", - ): - """Async equivalent; followers await without occupying worker threads.""" + context: RequestContext | None = None, + namespace: str | None = None, + ttl: int | None = None, + tags: Any = None, + _entry_point: str = "manual", + ) -> T: if self._distributed_execution is not None: - return await self._distributed_execution.arun( + return cast( + T, + await self._distributed_execution.arun( + query, + compute, + context=context, + namespace=namespace, + ttl=ttl, + tags=tags, + entry_point=_entry_point, + ), + ) + return cast( + T, + await self._execution.arun( query, compute, context=context, namespace=namespace, ttl=ttl, tags=tags, - entry_point=_entry_point, - ) - return await self._execution.arun( - query, - compute, - context=context, - namespace=namespace, - ttl=ttl, - tags=tags, - _entry_point=_entry_point, + _entry_point=_entry_point, + ), ) + @overload def cached( self, - func=None, + func: Callable[P, R], *, - namespace=None, - ttl=None, - tags=None, - context=None, - context_resolver=None, - **legacy, - ): + namespace: str | None = None, + ttl: int | None = None, + tags: Sequence[str] | None = None, + context: RequestContext | None = None, + context_resolver: Callable[..., RequestContext | None] | None = None, + ) -> CachedCallable[P, R]: ... + + @overload + def cached( + self, + func: None = None, + *, + namespace: str | None = None, + ttl: int | None = None, + tags: Sequence[str] | None = None, + context: RequestContext | None = None, + context_resolver: Callable[..., RequestContext | None] | None = None, + ) -> Callable[[Callable[P, R]], CachedCallable[P, R]]: ... + + def cached( + self, + func: Callable[..., Any] | None = None, + *, + namespace: str | None = None, + ttl: int | None = None, + tags: Sequence[str] | None = None, + context: RequestContext | None = None, + context_resolver: Callable[..., RequestContext | None] | None = None, + ) -> Any: if context_resolver is not None and ( not callable(context_resolver) or inspect.iscoroutinefunction(context_resolver) @@ -711,8 +860,8 @@ def cached( ): raise ValueError("Context resolvers must be synchronous callables") - def decorate(fn): - def context_for(base): + def decorate(fn: Callable[..., Any]) -> Any: + def context_for(base: RequestContext | None) -> RequestContext | None: if not isinstance(base, RequestContext): if inspect.iscoroutine(base): base.close() @@ -721,7 +870,7 @@ def context_for(base): base, operation=base.operation + ":" + fn.__module__ + "." + fn.__qualname__ ) - def resolve(args, kwargs): + def resolve(args: tuple[Any, ...], kwargs: dict[str, Any]) -> RequestContext | None: if context_resolver is None: return context_for(context or self.context) try: @@ -732,8 +881,8 @@ def resolve(args, kwargs): if inspect.iscoroutinefunction(fn): @wraps(fn) - async def async_wrapper(query, *args, **kwargs): - if args or kwargs or legacy: + async def async_wrapper(query: str, *args: Any, **kwargs: Any) -> Any: + if args or kwargs: return await self._apass_through( lambda: fn(query, *args, **kwargs), "unsupported_parameter", "decorator" ) @@ -742,7 +891,7 @@ async def async_wrapper(query, *args, **kwargs): return await self._apass_through( lambda: fn(query), "missing_identity", "decorator" ) - return await self.aget_or_compute( + return await self._arun_compute( query, lambda: fn(query), context=ctx, @@ -752,19 +901,19 @@ async def async_wrapper(query, *args, **kwargs): _entry_point="decorator", ) - async_wrapper.cache_context_for = context_for + async_wrapper.cache_context_for = context_for # type: ignore[attr-defined] return async_wrapper @wraps(fn) - def wrapper(query, *args, **kwargs): - if args or kwargs or legacy: + def wrapper(query: str, *args: Any, **kwargs: Any) -> Any: + if args or kwargs: return self._pass_through( lambda: fn(query, *args, **kwargs), "unsupported_parameter", "decorator" ) ctx = resolve((query,), {}) if ctx is None: return self._pass_through(lambda: fn(query), "missing_identity", "decorator") - return self.get_or_compute( + return self._run_compute( query, lambda: fn(query), context=ctx, @@ -774,13 +923,13 @@ def wrapper(query, *args, **kwargs): _entry_point="decorator", ) - wrapper.cache_context_for = context_for + wrapper.cache_context_for = context_for # type: ignore[attr-defined] return wrapper return decorate(func) if func is not None else decorate @execution_operation - def _pass_through(self, compute, reason, entry_point): + def _pass_through(self, compute: Callable[[], Any], reason: str, entry_point: str) -> Any: with self._lifecycle.operation(): result = self._execution.origin(compute) return self._execution.finish( @@ -796,7 +945,9 @@ def _pass_through(self, compute, reason, entry_point): ) @execution_operation - async def _apass_through(self, compute, reason, entry_point): + async def _apass_through( + self, compute: Callable[[], Any], reason: str, entry_point: str + ) -> Any: with self._lifecycle.operation(): result = await self._execution.aorigin(compute) return await self._execution.afinish( @@ -811,25 +962,39 @@ async def _apass_through(self, compute, reason, entry_point): reason=reason, ) - def cached_stream(self, func=None, **kwargs): + def cached_stream(self, func: Any = None, **kwargs: Any) -> Any: # Preserve the generator interface while unsupported replay remains pass-through. return func if func is not None else lambda fn: fn - def inspect(self, query, namespace=None, *, context=None, **legacy): - return self.lookup( - query, namespace, context=context, _inspect=True, _internal=True, **legacy - ).inspection() + def inspect( + self, + query: str, + namespace: str | None = None, + *, + context: RequestContext | None = None, + ) -> InspectionReport: + return cast( + InspectionReport, + self._lookup( + query, namespace, context=context, _inspect=True, _internal=True + ).inspection(), + ) @managed - def invalidate(self, namespace=None, *, context=None, **legacy): - prepared, reason = self._prepare("maintenance", context, namespace, legacy) - if reason: + def invalidate( + self, + namespace: str | None = None, + *, + context: RequestContext | None = None, + ) -> int: + prepared, reason = self._prepare("maintenance", context, namespace) + if reason or prepared is None: return 0 with self._mutation_lock: try: self._epoch += 1 self._execution.registry.abort() - count = self._dependency("store", self.backend.delete_namespace, prepared[1]) + count = int(self._dependency("store", self.backend.delete_namespace, prepared[1])) revision = ( self._dependency("backend", self.backend.revision) if self.config.deployment_mode == "shared" @@ -846,15 +1011,21 @@ def invalidate(self, namespace=None, *, context=None, **legacy): raise @managed - def invalidate_by_tag(self, tag, namespace=None, *, context=None, **legacy): - prepared, reason = self._prepare("maintenance", context, namespace, legacy) - if reason: + def invalidate_by_tag( + self, + tag: str, + namespace: str | None = None, + *, + context: RequestContext | None = None, + ) -> int: + prepared, reason = self._prepare("maintenance", context, namespace) + if reason or prepared is None: return 0 with self._mutation_lock: try: self._epoch += 1 self._execution.registry.abort() - count = self._dependency("store", self.backend.delete_by_tag, tag, prepared[1]) + count = int(self._dependency("store", self.backend.delete_by_tag, tag, prepared[1])) revision = ( self._dependency("backend", self.backend.revision) if self.config.deployment_mode == "shared" @@ -871,7 +1042,7 @@ def invalidate_by_tag(self, tag, namespace=None, *, context=None, **legacy): raise @managed - def clear(self, *, operator=False): + def clear(self, *, operator: bool = False) -> int: self._ensure_open() if not operator: raise ValueError("Clearing all partitions requires explicit operator=True") @@ -879,7 +1050,7 @@ def clear(self, *, operator=False): try: self._epoch += 1 self._execution.registry.abort() - count = self._dependency("store", self.backend.clear) + count = int(self._dependency("store", self.backend.clear)) revision = ( self._dependency("backend", self.backend.revision) if self.config.deployment_mode == "shared" @@ -896,29 +1067,43 @@ def clear(self, *, operator=False): raise @managed - def entries(self, namespace=None, *, context=None, operator=False): + def entries( + self, + namespace: str | None = None, + *, + context: RequestContext | None = None, + operator: bool = False, + ) -> list[CacheEntry]: self._ensure_open() with self._mutation_lock: if operator: - return self.backend.list_all_entries() - prepared, reason = self._prepare("maintenance", context, namespace, {}) - return [] if reason else self.backend.list_entries(prepared[1]) + return list(self.backend.list_all_entries()) + prepared, reason = self._prepare("maintenance", context, namespace) + return ( + [] if reason or prepared is None else list(self.backend.list_entries(prepared[1])) + ) @managed - def stats(self, *, context=None, namespace=None, operator=False): + def stats( + self, + *, + context: RequestContext | None = None, + namespace: str | None = None, + operator: bool = False, + ) -> StatsSnapshot: self._ensure_open() with self._mutation_lock: if operator: counts = self._stats.copy() total = len(self.backend.list_all_entries()) else: - prepared, reason = self._prepare("maintenance", context, namespace, {}) + prepared, reason = self._prepare("maintenance", context, namespace) counts = ( Counter() - if reason + if reason or prepared is None else self._partition_stats.get(prepared[1], Counter()).copy() ) - total = 0 if reason else self.backend.count(prepared[1]) + total = 0 if reason or prepared is None else self.backend.count(prepared[1]) hits, misses = counts["hits"], counts["misses"] return { "hits": hits, @@ -934,8 +1119,8 @@ def stats(self, *, context=None, namespace=None, operator=False): "stats_retention_evictions": self._stats_evictions, } - def _sweep(self): - delay, failures = 60, 0 + def _sweep(self) -> None: + delay, failures = 60.0, 0 while not self._stop_event.wait(delay): try: with self._lifecycle.operation(), self._mutation_lock: @@ -957,40 +1142,49 @@ def _sweep(self): self._background_failed = True return failures += 1 - delay = min(60, 2 ** min(failures - 1, 6)) + delay = float(min(60, 2 ** min(failures - 1, 6))) else: - delay, failures = 60, 0 + delay, failures = 60.0, 0 - def metrics(self, *, operator=False): + def metrics(self, *, operator: bool = False) -> MetricsSnapshot: """Return process-wide redacted measurements without backend I/O.""" if not operator: raise ValueError("Process-wide metrics require operator=True") + obs = self._observability.snapshot() + life = self._lifecycle.snapshot() + reg = self._execution.registry.snapshot() return { - **self._observability.snapshot(), + "enabled": bool(obs["enabled"]), + "counters": obs["counters"], + "latency": obs["latency"], "resources": { - **self._lifecycle.snapshot(), - **self._execution.registry.snapshot(), + "state": str(life["state"]), + "active_operations": int(life["active_operations"]), + "cache_jobs": int(life["cache_jobs"]), + "cleanup_failed": bool(life["cleanup_failed"]), + "cleanup_workers": int(life["cleanup_workers"]), + "active_flights": int(reg["active_flights"]), + "followers": int(reg["followers"]), + "retained_replay_bytes": int(reg["retained_replay_bytes"]), "sqlite_connections": int(getattr(self.backend, "_connection", None) is not None), }, "health": "degraded" - if not self._coherent - or self._background_failed - or self._lifecycle.snapshot()["cleanup_failed"] + if not self._coherent or self._background_failed or life["cleanup_failed"] else "healthy", } - def _revoke_flights(self): + def _revoke_flights(self) -> None: if hasattr(self, "_execution"): self._execution.shutdown() - def _finalize_resources(self): + def _finalize_resources(self) -> None: if self._sweep_thread is not None and self._sweep_thread is not threading.current_thread(): self._sweep_thread.join() if hasattr(self.backend, "close"): self.backend.close() self._closed = True - def close(self, timeout=None): + def close(self, timeout: float | None = None) -> None: if self._lifecycle.current is not None: raise CacheClosedError("Cannot close the cache from its own active operation") try: @@ -1003,17 +1197,17 @@ def close(self, timeout=None): self._stop_event.set() self._lifecycle.close(timeout) - async def aclose(self, timeout=None): + async def aclose(self, timeout: float | None = None) -> None: if self._lifecycle.current is not None: raise CacheClosedError("Cannot close the cache from its own active operation") self._stop_event.set() await self._lifecycle.aclose(timeout) - def __enter__(self): + def __enter__(self) -> LongCache: self._ensure_open() return self - def __exit__(self, exc_type, exc, traceback): + def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None: try: self.close() except Exception: @@ -1021,11 +1215,11 @@ def __exit__(self, exc_type, exc, traceback): if exc is None: raise - async def __aenter__(self): + async def __aenter__(self) -> LongCache: self._ensure_open() return self - async def __aexit__(self, exc_type, exc, traceback): + async def __aexit__(self, exc_type: Any, exc: Any, traceback: Any) -> None: try: await self.aclose() except Exception: @@ -1033,5 +1227,5 @@ async def __aexit__(self, exc_type, exc, traceback): if exc is None: raise - def __repr__(self): + def __repr__(self) -> str: return f"LongCache(backend={self.config.backend!r}, format=2)" diff --git a/src/longcache/core/decision.py b/src/longcache/core/decision.py index 9d71183..a2fecc4 100644 --- a/src/longcache/core/decision.py +++ b/src/longcache/core/decision.py @@ -1,16 +1,22 @@ """Public, redacted cache decisions and policy hook results.""" +from __future__ import annotations + from dataclasses import dataclass, field -from typing import Any, Literal import re +from typing import Any + +from .types import InspectionReport, LookupOutcome, PolicyAction, StoreOutcome + +__all__ = ["PolicyDecision", "LookupDecision", "StoreDecision", "InspectionReport"] @dataclass(frozen=True) class PolicyDecision: - action: Literal["accept", "reject", "abstain"] + action: PolicyAction reason: str = "application_policy" - def __post_init__(self): + def __post_init__(self) -> None: if self.action not in ("accept", "reject", "abstain"): raise ValueError("Invalid policy action") if not re.fullmatch(r"[a-z][a-z0-9_]{0,63}", self.reason): @@ -19,7 +25,7 @@ def __post_init__(self): @dataclass(frozen=True) class LookupDecision: - outcome: Literal["HIT", "MISS", "BYPASS"] + outcome: LookupOutcome reason: str response: Any = field(default=None, repr=False) similarity: float | None = None @@ -29,7 +35,7 @@ class LookupDecision: _expires_at: float | None = field(default=None, repr=False, compare=False) _shared_epoch: int | None = field(default=None, repr=False, compare=False) - def inspection(self) -> dict: + def inspection(self) -> InspectionReport: return { "outcome": self.outcome, "reason": self.reason, @@ -42,6 +48,6 @@ def inspection(self) -> dict: @dataclass(frozen=True) class StoreDecision: - outcome: Literal["STORED", "REJECTED"] + outcome: StoreOutcome reason: str entry_id: int | None = None diff --git a/src/longcache/core/distributed_singleflight.py b/src/longcache/core/distributed_singleflight.py index 09687f9..31a8be2 100644 --- a/src/longcache/core/distributed_singleflight.py +++ b/src/longcache/core/distributed_singleflight.py @@ -6,7 +6,9 @@ import inspect import threading import time +from typing import TYPE_CHECKING, Any, Awaitable, Callable, cast +from .decision import LookupDecision from .errors import ( CacheOperationError, SingleflightInvalidatedError, @@ -15,16 +17,19 @@ ) from .identity import digest +if TYPE_CHECKING: + from longcache.core.cache import LongCache + class _LeaseRenewal: - def __init__(self, backend, key: str, token: str, ttl_ms: int, interval: float): + def __init__(self, backend: Any, key: str, token: str, ttl_ms: int, interval: float) -> None: self.backend, self.key, self.token = backend, key, token self.ttl_ms, self.interval = ttl_ms, interval self.stop_event = threading.Event() self.lost = threading.Event() self.thread = threading.Thread(target=self._run, daemon=True, name="longcache-lease") - def _run(self): + def _run(self) -> None: while not self.stop_event.wait(self.interval): try: if not self.backend.renew_flight(self.key, self.token, self.ttl_ms): @@ -34,11 +39,11 @@ def _run(self): self.lost.set() return - def __enter__(self): + def __enter__(self) -> _LeaseRenewal: self.thread.start() return self - def __exit__(self, *_): + def __exit__(self, *_: Any) -> None: self.stop_event.set() self.thread.join(timeout=2) @@ -46,19 +51,26 @@ def __exit__(self, *_): class DistributedExecution: """Shares only stored, policy-verified results; leaders own their origin calls.""" - def __init__(self, cache): + def __init__(self, cache: LongCache) -> None: self.cache = cache @staticmethod - def _invoke(function, *args, _cancel=None): + def _invoke(function: Callable[..., Any], *args: Any, _cancel: Any = None) -> Any: return function(*args) - async def _call(self, function, *args): + async def _call(self, function: Callable[..., Any], *args: Any) -> Any: return await self.cache._lifecycle.call(self._invoke, function, *args) - def _prepare(self, query, context, namespace, ttl, tags): + def _prepare( + self, + query: str, + context: Any, + namespace: str | None, + ttl: int | None, + tags: Any, + ) -> tuple[Any, str | None, int | None, str | None]: cache = self.cache - prepared, reason = cache._prepare(query, context, namespace, {}) + prepared, reason = cache._prepare(query, context, namespace) if prepared is None: return None, reason, None, None ctx = prepared[0] @@ -73,12 +85,22 @@ def _prepare(self, query, context, namespace, ttl, tags): policy = digest({"ttl": cache.config.ttl_seconds if ttl is None else ttl, "tags": tags}) return prepared, None, shared_epoch, digest({"exact": exact, "policy": policy}) - def _check_epoch(self, expected): + def _check_epoch(self, expected: int) -> None: actual = self.cache._dependency("backend", self.cache.backend.epoch) if actual != expected: raise SingleflightInvalidatedError("Shared execution was invalidated") - def _finish(self, prepared, decision, response, source, mode, role, entry_point, reason=None): + def _finish( + self, + prepared: Any, + decision: Any, + response: Any, + source: str, + mode: str, + role: str, + entry_point: str, + reason: str | None = None, + ) -> Any: return self.cache._execution.finish( prepared, decision, @@ -91,15 +113,28 @@ def _finish(self, prepared, decision, response, source, mode, role, entry_point, reason=reason, ) - def _lookup(self, query, ctx, namespace): - return self.cache.lookup(query, namespace, context=ctx, _record=False, _internal=True) + def _lookup(self, query: str, ctx: Any, namespace: str | None) -> LookupDecision: + return cast( + LookupDecision, + self.cache._lookup(query, namespace, context=ctx, _record=False, _internal=True), + ) - async def _alookup(self, query, ctx, namespace): - return await self.cache.alookup( + async def _alookup(self, query: str, ctx: Any, namespace: str | None) -> LookupDecision: + return await self.cache._alookup( query, namespace, context=ctx, _record=False, _internal=True ) - def run(self, query, compute, *, context, namespace, ttl, tags, entry_point): + def run( + self, + query: str, + compute: Callable[[], Any], + *, + context: Any, + namespace: str | None, + ttl: int | None, + tags: Any, + entry_point: str, + ) -> Any: try: return self._run( query, @@ -125,7 +160,17 @@ def run(self, query, compute, *, context, namespace, ttl, tags, entry_point): "cache_unavailable", ) - def _run(self, query, compute, *, context, namespace, ttl, tags, entry_point): + def _run( + self, + query: str, + compute: Callable[[], Any], + *, + context: Any, + namespace: str | None, + ttl: int | None, + tags: Any, + entry_point: str, + ) -> Any: try: asyncio.get_running_loop() except RuntimeError: @@ -171,13 +216,15 @@ def _run(self, query, compute, *, context, namespace, ttl, tags, entry_point): ) deadline = time.monotonic() + cache.config.singleflight_wait_timeout_seconds while True: - self._check_epoch(shared_epoch) + if shared_epoch is not None: + self._check_epoch(shared_epoch) key, token = cache._dependency( "backend", cache.backend.acquire_flight, flight_key, lease_ms ) if token is not None: try: - self._check_epoch(shared_epoch) + if shared_epoch is not None: + self._check_epoch(shared_epoch) decision = self._lookup(query, ctx, namespace) if decision.outcome == "HIT": return self._finish( @@ -209,7 +256,7 @@ def _run(self, query, compute, *, context, namespace, ttl, tags, entry_point): ) as renewal: result = cache._execution.origin(compute) if not renewal.lost.is_set(): - stored = cache.set( + stored = cache._set( query, result, namespace, @@ -247,7 +294,8 @@ def _run(self, query, compute, *, context, namespace, ttl, tags, entry_point): except CacheOperationError: pass # The bounded lease expires; never repeat a completed origin. while time.monotonic() < deadline: - self._check_epoch(shared_epoch) + if shared_epoch is not None: + self._check_epoch(shared_epoch) decision = self._lookup(query, ctx, namespace) if decision.outcome == "HIT": return self._finish( @@ -289,7 +337,17 @@ def _run(self, query, compute, *, context, namespace, ttl, tags, entry_point): else: raise SingleflightTimeoutError("Timed out waiting for shared execution") - async def arun(self, query, compute, *, context, namespace, ttl, tags, entry_point): + async def arun( + self, + query: str, + compute: Callable[[], Awaitable[Any]], + *, + context: Any, + namespace: str | None, + ttl: int | None, + tags: Any, + entry_point: str, + ) -> Any: try: return await self._arun( query, @@ -317,7 +375,17 @@ async def arun(self, query, compute, *, context, namespace, ttl, tags, entry_poi "cache_unavailable", ) - async def _arun(self, query, compute, *, context, namespace, ttl, tags, entry_point): + async def _arun( + self, + query: str, + compute: Callable[[], Awaitable[Any]], + *, + context: Any, + namespace: str | None, + ttl: int | None, + tags: Any, + entry_point: str, + ) -> Any: if not callable(compute): raise TypeError("aget_or_compute requires an awaitable-producing callable") cache = self.cache @@ -333,7 +401,7 @@ async def _arun(self, query, compute, *, context, namespace, ttl, tags, entry_po self._finish, None, None, result, "origin", "async", "none", entry_point, reason ) ctx = prepared[0] - decision = await self._alookup(query, ctx, namespace) + decision: LookupDecision = await self._alookup(query, ctx, namespace) if decision.outcome == "HIT": return await self._call( self._finish, @@ -360,11 +428,13 @@ async def _arun(self, query, compute, *, context, namespace, ttl, tags, entry_po ) deadline = time.monotonic() + cache.config.singleflight_wait_timeout_seconds while True: - await self._call(self._check_epoch, shared_epoch) + if shared_epoch is not None: + await self._call(self._check_epoch, shared_epoch) key, token = await self._call(cache.backend.acquire_flight, flight_key, lease_ms) if token is not None: try: - await self._call(self._check_epoch, shared_epoch) + if shared_epoch is not None: + await self._call(self._check_epoch, shared_epoch) decision = await self._alookup(query, ctx, namespace) if decision.outcome == "HIT": return await self._call( @@ -399,7 +469,7 @@ async def _arun(self, query, compute, *, context, namespace, ttl, tags, entry_po ) as renewal: result = await cache._execution.aorigin(compute) if not renewal.lost.is_set(): - stored = await cache.aset( + stored = await cache._aset( query, result, namespace, @@ -445,7 +515,8 @@ async def _arun(self, query, compute, *, context, namespace, ttl, tags, entry_po except CacheOperationError: pass # A lost release is bounded by the Redis TTL. while time.monotonic() < deadline: - await self._call(self._check_epoch, shared_epoch) + if shared_epoch is not None: + await self._call(self._check_epoch, shared_epoch) decision = await self._alookup(query, ctx, namespace) if decision.outcome == "HIT": return await self._call( diff --git a/src/longcache/core/embedder.py b/src/longcache/core/embedder.py index 77e9c2f..f37d088 100755 --- a/src/longcache/core/embedder.py +++ b/src/longcache/core/embedder.py @@ -9,11 +9,16 @@ import logging import threading +from typing import TYPE_CHECKING, Any import numpy as np from longcache.core.base_embedder import BaseEmbedder +if TYPE_CHECKING: + from .embedding_contract import EmbeddingDescriptor + from .types import Float32Vector + _log = logging.getLogger(__name__) @@ -34,7 +39,15 @@ class LocalEmbedder(BaseEmbedder): _instances: dict[tuple[str, str], LocalEmbedder] = {} _lock = threading.Lock() - def __new__(cls, model_name: str = "BAAI/bge-small-en-v1.5", revision: str = ""): + _model_name: str + _revision: str + _model: Any + _dimension: int | None + _model_lock: threading.Lock + + def __new__( + cls, model_name: str = "BAAI/bge-small-en-v1.5", revision: str = "" + ) -> LocalEmbedder: """ Singleton pattern: one embedder instance per model name. @@ -51,13 +64,14 @@ def __new__(cls, model_name: str = "BAAI/bge-small-en-v1.5", revision: str = "") if key not in cls._instances: instance = super().__new__(cls) instance._model_name = model_name + instance._revision = revision instance._model = None instance._dimension = None instance._model_lock = threading.Lock() cls._instances[key] = instance return cls._instances[key] - def __init__(self, model_name: str = "BAAI/bge-small-en-v1.5", revision: str = ""): + def __init__(self, model_name: str = "BAAI/bge-small-en-v1.5", revision: str = "") -> None: """ Initialize the embedder. Model is loaded lazily on first embed() call. @@ -65,24 +79,22 @@ def __init__(self, model_name: str = "BAAI/bge-small-en-v1.5", revision: str = " model_name: The sentence-transformers model identifier. Defaults to BAAI/bge-small-en-v1.5 (384 dimensions). """ - # __new__ handles actual initialization; __init__ just ensures - # the model name is set (idempotent for singleton) self._model_name = model_name self._revision = revision @property - def descriptor(self): + def descriptor(self) -> EmbeddingDescriptor: from .embedding_contract import EmbeddingDescriptor self._load_model() revision = self._revision if not revision: - revision = getattr(self._model[0].auto_model.config, "_commit_hash", None) + revision = getattr(self._model[0].auto_model.config, "_commit_hash", "") or "" if not revision: raise ValueError("Set embedder_revision to an immutable model revision") return EmbeddingDescriptor(self._model_name, revision, self.dimension) - def _load_model(self): + def _load_model(self) -> None: """ Lazily load the sentence-transformers model. @@ -112,8 +124,9 @@ def _load_model(self): getattr(self._model, "get_embedding_dimension", None) or self._model.get_sentence_embedding_dimension ) - self._dimension = get_dimension() - _log.info("Embedding model loaded: dim=%d", self._dimension) + dim_val = get_dimension() + self._dimension = int(dim_val) if dim_val is not None else None + _log.info("Embedding model loaded: dim=%s", self._dimension) @property def dimension(self) -> int: @@ -126,9 +139,11 @@ def dimension(self) -> int: The dimensionality of the embedding vectors. """ self._load_model() + if self._dimension is None: + raise ValueError("Failed to resolve embedding dimension") return self._dimension - def embed(self, text: str) -> np.ndarray: + def embed(self, text: str) -> Float32Vector: """ Generate an embedding vector for the input text. @@ -145,9 +160,9 @@ def embed(self, text: str) -> np.ndarray: """ self._load_model() vector = self._model.encode(text, normalize_embeddings=True) - return vector.astype(np.float32) + return np.asarray(vector, dtype=np.float32) - def embed_batch(self, texts: list[str]) -> np.ndarray: + def embed_batch(self, texts: list[str]) -> Float32Vector: """ Generate embedding vectors for multiple texts in a single batch. @@ -166,7 +181,7 @@ def embed_batch(self, texts: list[str]) -> np.ndarray: self._load_model() vectors = self._model.encode(texts, normalize_embeddings=True, batch_size=32) - return vectors.astype(np.float32) + return np.asarray(vectors, dtype=np.float32) def __repr__(self) -> str: model_status = "loaded" if self._model is not None else "not loaded" diff --git a/src/longcache/core/embedding_contract.py b/src/longcache/core/embedding_contract.py index 87c97d1..dd01f11 100644 --- a/src/longcache/core/embedding_contract.py +++ b/src/longcache/core/embedding_contract.py @@ -1,11 +1,19 @@ """Validated embedding generations; equal dimensions do not imply compatible models.""" +from __future__ import annotations + from dataclasses import asdict, dataclass import threading +from typing import TYPE_CHECKING, Any from weakref import WeakKeyDictionary import numpy as np + from .identity import digest +if TYPE_CHECKING: + from .base_embedder import BaseEmbedder + from .types import Float32Vector + @dataclass(frozen=True) class EmbeddingDescriptor: @@ -15,7 +23,7 @@ class EmbeddingDescriptor: preprocessing: str = "text-v1" normalization: str = "l2-float32-v1" - def __post_init__(self): + def __post_init__(self) -> None: if type(self.dimension) is not int or self.dimension < 1: raise ValueError("Embedding dimension must be a positive integer") if not all( @@ -26,11 +34,11 @@ def __post_init__(self): raise ValueError("Unsupported normalization contract") @property - def fingerprint(self): + def fingerprint(self) -> str: return digest(asdict(self)) -def validate_vector(vector, dimension): +def validate_vector(vector: Any, dimension: int) -> Float32Vector: if not isinstance(vector, np.ndarray) or vector.dtype != np.float32: raise ValueError("Embeddings must be float32 numpy arrays") if vector.shape != (dimension,) or not vector.flags.c_contiguous: @@ -40,19 +48,19 @@ def validate_vector(vector, dimension): return vector.copy() -_locks = WeakKeyDictionary() +_locks: WeakKeyDictionary[Any, threading.RLock] = WeakKeyDictionary() _registry_lock = threading.Lock() class EmbeddingGuard: - def __init__(self, embedder): + def __init__(self, embedder: BaseEmbedder) -> None: self.embedder = embedder with _registry_lock: self.lock = _locks.setdefault(embedder, threading.RLock()) - self._descriptor = None + self._descriptor: EmbeddingDescriptor | None = None @property - def descriptor(self): + def descriptor(self) -> EmbeddingDescriptor: with self.lock: current = self.embedder.descriptor if ( @@ -65,12 +73,12 @@ def descriptor(self): self._descriptor = current return current - def embed(self, text): + def embed(self, text: str) -> Float32Vector: with self.lock: descriptor = self.descriptor return validate_vector(self.embedder.embed(text), descriptor.dimension) - def batch(self, texts): + def batch(self, texts: list[str]) -> Float32Vector: with self.lock: dimension = self.descriptor.dimension vectors = self.embedder.embed_batch(texts) diff --git a/src/longcache/core/entry.py b/src/longcache/core/entry.py index ccbe214..179a432 100755 --- a/src/longcache/core/entry.py +++ b/src/longcache/core/entry.py @@ -7,12 +7,16 @@ from __future__ import annotations -import time +from collections.abc import Mapping from dataclasses import dataclass, field -from typing import Any, Optional +import time +from typing import TYPE_CHECKING, Any import numpy as np +if TYPE_CHECKING: + from .types import Float32Vector + @dataclass class CacheEntry: @@ -30,19 +34,16 @@ class CacheEntry: hit_count: Number of times this entry has been served from cache. """ - id: Optional[int] = None + id: int | None = None namespace: str = "default" query: str = "" - vector: Optional[np.ndarray] = None + vector: Float32Vector | None = None response: Any = None created_at: float = field(default_factory=time.time) - expires_at: Optional[float] = None + expires_at: float | None = None hit_count: int = 0 - model: str = "" - system_prompt: str = "" - temperature: float = -1.0 tags: list[str] = field(default_factory=list) - metadata: dict = field(default_factory=dict) + metadata: dict[str, Any] = field(default_factory=dict) def is_expired(self) -> bool: """ @@ -60,7 +61,7 @@ def increment_hit(self) -> None: """Increment the hit count for this cache entry.""" self.hit_count += 1 - def to_dict(self) -> dict: + def to_dict(self) -> dict[str, Any]: """ Serialize the cache entry to a dictionary. @@ -79,20 +80,18 @@ def to_dict(self) -> dict: "created_at": self.created_at, "expires_at": self.expires_at, "hit_count": self.hit_count, - "model": self.model, - "system_prompt": self.system_prompt, - "temperature": self.temperature, "tags": list(self.tags or []), "metadata": self.metadata, } @classmethod - def from_dict(cls, data: dict, vector_dim: Optional[int] = None) -> CacheEntry: + def from_dict(cls, data: Mapping[str, Any], vector_dim: int | None = None) -> CacheEntry: """ Deserialize a cache entry from a dictionary. The vector field is converted from bytes back to a numpy array - if vector_dim is provided. + if vector_dim is provided. Legacy v0.1-v0.4 fields (model, system_prompt, + temperature) are ignored gracefully if present in persisted payloads. Args: data: Dictionary with cache entry fields. @@ -103,23 +102,43 @@ def from_dict(cls, data: dict, vector_dim: Optional[int] = None) -> CacheEntry: CacheEntry instance. """ vector = data.get("vector") + parsed_vector: Float32Vector | None = None if isinstance(vector, bytes) and vector_dim is not None: - vector = np.frombuffer(vector, dtype=np.float32).copy() - if vector_dim and len(vector) != vector_dim: - vector = vector.reshape(vector_dim) if vector.size == vector_dim else vector + buf = np.frombuffer(vector, dtype=np.float32).copy() + if vector_dim and len(buf) != vector_dim: + parsed_vector = buf.reshape(vector_dim) if buf.size == vector_dim else buf + else: + parsed_vector = buf + elif isinstance(vector, np.ndarray) and vector.dtype == np.float32: + parsed_vector = vector + + raw_id = data.get("id") + entry_id = int(raw_id) if raw_id is not None else None + + raw_created = data.get("created_at") + created_at = float(raw_created) if raw_created is not None else time.time() + + raw_expires = data.get("expires_at") + expires_at = float(raw_expires) if raw_expires not in (None, "") else None + + raw_hits = data.get("hit_count") + hit_count = int(raw_hits) if raw_hits is not None else 0 + + raw_meta = data.get("metadata") + metadata = dict(raw_meta) if isinstance(raw_meta, Mapping) else {} + + raw_tags = data.get("tags") + tags = [str(t) for t in raw_tags] if isinstance(raw_tags, (list, tuple)) else [] return cls( - id=data.get("id"), - namespace=data.get("namespace", "default"), - query=data.get("query", ""), - vector=vector, + id=entry_id, + namespace=str(data.get("namespace", "default")), + query=str(data.get("query", "")), + vector=parsed_vector, response=data.get("response"), - created_at=data.get("created_at", time.time()), - expires_at=data.get("expires_at"), - hit_count=data.get("hit_count", 0), - model=data.get("model", ""), - system_prompt=data.get("system_prompt", ""), - temperature=float(data.get("temperature", -1.0)), - tags=list(data.get("tags") or []), - metadata=data.get("metadata") or {}, + created_at=created_at, + expires_at=expires_at, + hit_count=hit_count, + tags=tags, + metadata=metadata, ) diff --git a/src/longcache/core/errors.py b/src/longcache/core/errors.py index 9c2d6e7..c75e1fa 100644 --- a/src/longcache/core/errors.py +++ b/src/longcache/core/errors.py @@ -1,5 +1,7 @@ """Public reliability errors. Messages never contain request or dependency payloads.""" +from __future__ import annotations + class CacheError(RuntimeError): """Base class for cache reliability failures.""" @@ -20,8 +22,8 @@ class CacheCapacityError(CacheError): class CacheOperationError(CacheError): """A recognized runtime cache dependency failed.""" - def __init__(self, stage): - self.stage = stage + def __init__(self, stage: str) -> None: + self.stage: str = stage super().__init__(f"Cache operation failed at stage: {stage}") diff --git a/src/longcache/core/execution.py b/src/longcache/core/execution.py index 3ed8627..01794a7 100644 --- a/src/longcache/core/execution.py +++ b/src/longcache/core/execution.py @@ -1,31 +1,37 @@ """One exact-scoped execution protocol for manual, decorator and provider adapters.""" +from __future__ import annotations + import asyncio from concurrent.futures import TimeoutError as FutureTimeout from contextvars import ContextVar from dataclasses import dataclass, field import inspect import json -import time import threading +import time +from typing import TYPE_CHECKING, Any, Callable, cast from .decision import LookupDecision from .errors import ( CacheClosedError, CacheOperationError, + SingleflightInvalidatedError, SingleflightLeaderError, SingleflightReentrancyError, SingleflightTimeoutError, - SingleflightInvalidatedError, ) -from .identity import digest, canonical +from .identity import canonical, digest from .policy import admit, decode_response, verify from .singleflight import Completion, Flight, Singleflight +if TYPE_CHECKING: + from longcache.core.cache import LongCache + @dataclass class Start: - prepared: tuple | None + prepared: tuple[Any, ...] | None decision: LookupDecision | None flight: Flight | None leader: bool @@ -36,27 +42,29 @@ class Start: _detach_lock: threading.Lock = field(default_factory=threading.Lock, repr=False) @property - def role(self): + def role(self) -> str: return "none" if self.flight is None else "leader" if self.leader else "follower" class Execution: - def __init__(self, cache): + def __init__(self, cache: LongCache) -> None: self.cache = cache self.registry = Singleflight(cache.config) - self._owned = ContextVar(f"longcache_flight_owner_{id(self)}", default=()) + self._owned: ContextVar[tuple[Any, ...]] = ContextVar( + f"longcache_flight_owner_{id(self)}", default=() + ) - def shutdown(self): + def shutdown(self) -> None: self.registry.abort(shutdown=True) - def detach(self, state): + def detach(self, state: Start) -> None: with state._detach_lock: if state.flight is not None and not state.leader and not state.detached: state.detached = True self.registry.detach(state.flight) @staticmethod - def options(ttl, tags): + def options(ttl: int | None, tags: Any) -> tuple[str, ...]: if ttl is not None and (type(ttl) is not int or ttl < 0): raise ValueError("TTL must be a nonnegative integer") if tags is not None and ( @@ -65,12 +73,24 @@ def options(ttl, tags): raise ValueError("Tags must be strings") return tuple(sorted(set(tags or ()))) - def lookup(self, query, context, namespace, *, _cancel=None): - return self.cache.lookup( - query, namespace, context=context, _record=False, _internal=True, _cancel=_cancel + def lookup( + self, query: str, context: Any, namespace: str | None, *, _cancel: Any = None + ) -> LookupDecision: + return cast( + LookupDecision, + self.cache._lookup( + query, namespace, context=context, _record=False, _internal=True, _cancel=_cancel + ), ) - async def afast_start(self, query, context, namespace, ttl, tags): + async def afast_start( + self, + query: str, + context: Any, + namespace: str | None, + ttl: int | None, + tags: tuple[str, ...], + ) -> Start: """Join before lookup when identity is initialized; only the leader searches. Pure identity work and the short registry lock are safe on the loop. The @@ -80,7 +100,7 @@ async def afast_start(self, query, context, namespace, ttl, tags): cache = self.cache if not cache._ready or not cache._coherent or not cache.config.singleflight_enabled: return await self.astart(query, context, namespace, ttl, tags) - prepared, reason = cache._prepare(query, context, namespace, {}, _initialize=False) + prepared, reason = cache._prepare(query, context, namespace, _initialize=False) if prepared is None: return Start(None, None, None, False, cache._epoch, reason=reason) ctx = prepared[0] @@ -117,10 +137,19 @@ async def afast_start(self, query, context, namespace, ttl, tags): self._owned.reset(owner) return state - def start(self, query, context, namespace, ttl, tags, *, _cancel=None): + def start( + self, + query: str, + context: Any, + namespace: str | None, + ttl: int | None, + tags: tuple[str, ...], + *, + _cancel: Any = None, + ) -> Start: cache = self.cache try: - prepared, reason = cache._prepare(query, context, namespace, {}) + prepared, reason = cache._prepare(query, context, namespace) except CacheOperationError: if cache.config.cache_error_policy != "bypass": raise @@ -174,7 +203,7 @@ def start(self, query, context, namespace, ttl, tags, *, _cancel=None): if owner is not None: self._owned.reset(owner) - async def astart(self, *args): + async def astart(self, *args: Any) -> Start: future, token = self.cache._lifecycle.submit(self.start, *args) bridge = asyncio.wrap_future(future) bridge.add_done_callback(lambda result: None if result.cancelled() else result.exception()) @@ -185,7 +214,7 @@ async def astart(self, *args): # Cancellation can arrive after election but before the await receives # its result. Release that ownership even when the caller never saw it. - def abandon(result): + def abandon(result: Any) -> None: if not result.cancelled() and result.exception() is None: state = result.result() if state.flight is not None: @@ -199,18 +228,18 @@ def abandon(result): def finish( self, - prepared, - decision, - response, - source, - mode, - role, - entry_point, - epoch, + prepared: Any, + decision: Any, + response: Any, + source: str, + mode: str, + role: str, + entry_point: str, + epoch: int, *, - reason=None, - _cancel=None, - ): + reason: str | None = None, + _cancel: Any = None, + ) -> Any: cache = self.cache with cache._mutation_lock: cache._ensure_open() @@ -284,10 +313,12 @@ def finish( ) return response - async def afinish(self, *args, **kwargs): + async def afinish(self, *args: Any, **kwargs: Any) -> Any: return await self.cache._lifecycle.call(self.finish, *args, **kwargs) - def envelope(self, result, context, expires_at, entry_id): + def envelope( + self, result: Any, context: Any, expires_at: float | None, entry_id: int | None + ) -> Completion: payload, admission = self.cache._dependency( "admission", admit, result, context, self.cache._admit ) @@ -298,9 +329,21 @@ def envelope(self, result, context, expires_at, entry_id): return Completion("non_shareable") return Completion("value", encoded, expires_at=expires_at, entry_id=entry_id) - def replay(self, completion, query, context, prepared, *, _cancel=None): + def replay( + self, + completion: Completion, + query: str, + context: Any, + prepared: Any, + *, + _cancel: Any = None, + ) -> tuple[Any, bool, LookupDecision | None]: if completion.kind == "invalidated": - raise type(completion.error)(str(completion.error)) + raise ( + type(completion.error)(str(completion.error)) + if completion.error + else SingleflightInvalidatedError() + ) if completion.kind == "error": raise SingleflightLeaderError( "Shared execution failed or its leader was cancelled" @@ -312,6 +355,8 @@ def replay(self, completion, query, context, prepared, *, _cancel=None): and time.time() >= completion.expires_at ): return None, False, None + if completion.payload is None: + return None, False, None response = decode_response(json.loads(completion.payload), context.codec) payload, admission = cache._dependency("admission", admit, response, context, cache._admit) if admission.action != "accept": @@ -385,7 +430,15 @@ def replay(self, completion, query, context, prepared, *, _cancel=None): return None, False, None return response, True, decision - def publish(self, flight, result, context, epoch, expires_at=None, entry_id=None): + def publish( + self, + flight: Flight, + result: Any, + context: Any, + epoch: int, + expires_at: float | None = None, + entry_id: int | None = None, + ) -> None: completion = self.envelope(result, context, expires_at, entry_id) with self.cache._mutation_lock: if epoch != self.cache._epoch: @@ -395,9 +448,14 @@ def publish(self, flight, result, context, epoch, expires_at=None, entry_id=None ) self.registry.complete(flight, completion) - def cached(self, state, mode, entry_point, *, _cancel=None): + def cached(self, state: Start, mode: str, entry_point: str, *, _cancel: Any = None) -> Any: try: - if state.flight is not None and state.leader: + if ( + state.flight is not None + and state.leader + and state.decision is not None + and state.prepared is not None + ): self.publish( state.flight, state.decision.response, @@ -409,7 +467,7 @@ def cached(self, state, mode, entry_point, *, _cancel=None): return self.finish( state.prepared, state.decision, - state.decision.response, + state.decision.response if state.decision else None, "cache", mode, state.role, @@ -425,15 +483,25 @@ def cached(self, state, mode, entry_point, *, _cancel=None): if state.flight is not None and not state.leader: self.detach(state) - def store(self, state, query, result, namespace, ttl, tags, *, _cancel=None): + def store( + self, + state: Start, + query: str, + result: Any, + namespace: str | None, + ttl: int | None, + tags: Any, + *, + _cancel: Any = None, + ) -> Any: cache = self.cache effective_ttl = cache.config.ttl_seconds if ttl is None else ttl expiry = time.time() + effective_ttl if effective_ttl else None - stored = cache.set( + stored = cache._set( query, result, namespace, - context=state.prepared[0], + context=state.prepared[0] if state.prepared else None, ttl=ttl, tags=tags, _expected_epoch=state.epoch, @@ -441,7 +509,7 @@ def store(self, state, query, result, namespace, ttl, tags, *, _cancel=None): _internal=True, _cancel=_cancel, ) - if state.flight is not None and state.leader: + if state.flight is not None and state.leader and state.prepared is not None: if stored.outcome == "REJECTED" and ( cache.config.deployment_mode == "shared" or stored.reason not in ("cache_unavailable", "invalidated") @@ -453,7 +521,7 @@ def store(self, state, query, result, namespace, ttl, tags, *, _cancel=None): ) return result - def origin(self, compute): + def origin(self, compute: Callable[[], Any]) -> Any: self.cache._observability.count("origin_executions") started = time.monotonic() try: @@ -469,7 +537,7 @@ def origin(self, compute): finally: self.cache._observability.duration("origin", time.monotonic() - started) - async def aorigin(self, compute): + async def aorigin(self, compute: Callable[[], Any]) -> Any: self.cache._observability.count("origin_executions") started = time.monotonic() try: @@ -483,11 +551,22 @@ async def aorigin(self, compute): finally: self.cache._observability.duration("origin", time.monotonic() - started, "async") - def follower_result(self, state, completion, query, mode, entry_point, *, _cancel=None): + def follower_result( + self, + state: Start, + completion: Completion, + query: str, + mode: str, + entry_point: str, + *, + _cancel: Any = None, + ) -> tuple[Any, bool]: + if state.prepared is None: + return None, False response, accepted, decision = self.replay( completion, query, state.prepared[0], state.prepared ) - if not accepted: + if not accepted or decision is None: return None, False source = "cache" if decision.outcome == "HIT" else "coalesced" return self.finish( @@ -504,15 +583,15 @@ def follower_result(self, state, completion, query, mode, entry_point, *, _cance def run( self, - query, - compute, + query: str, + compute: Callable[[], Any], *, - context=None, - namespace=None, - ttl=None, - tags=None, - _entry_point="manual", - ): + context: Any = None, + namespace: str | None = None, + ttl: int | None = None, + tags: Any = None, + _entry_point: str = "manual", + ) -> Any: try: asyncio.get_running_loop() except RuntimeError: @@ -525,10 +604,10 @@ def run( or inspect.iscoroutinefunction(getattr(compute, "__call__", None)) ): raise TypeError("get_or_compute requires a synchronous callable") - tags = self.options(ttl, tags) + tags_tuple = self.options(ttl, tags) cache = self.cache with cache._lifecycle.operation(): - state = self.start(query, context, namespace, ttl, tags) + state = self.start(query, context, namespace, ttl, tags_tuple) if state.prepared is None: return self.finish( None, @@ -541,7 +620,7 @@ def run( state.epoch, reason=state.reason, ) - if state.decision.outcome == "HIT": + if state.decision is not None and state.decision.outcome == "HIT": return self.cached(state, "sync", _entry_point) if state.flight is not None and not state.leader: started = time.monotonic() @@ -569,7 +648,7 @@ def run( ) try: result = self.origin(compute) - self.store(state, query, result, namespace, ttl, tags) + self.store(state, query, result, namespace, ttl, tags_tuple) return self.finish( state.prepared, None, @@ -590,21 +669,21 @@ def run( async def arun( self, - query, - compute, + query: str, + compute: Callable[[], Any], *, - context=None, - namespace=None, - ttl=None, - tags=None, - _entry_point="manual", - ): + context: Any = None, + namespace: str | None = None, + ttl: int | None = None, + tags: Any = None, + _entry_point: str = "manual", + ) -> Any: if not callable(compute): raise TypeError("aget_or_compute requires an awaitable-producing callable") - tags = self.options(ttl, tags) + tags_tuple = self.options(ttl, tags) cache = self.cache with cache._lifecycle.operation(): - state = await self.afast_start(query, context, namespace, ttl, tags) + state = await self.afast_start(query, context, namespace, ttl, tags_tuple) if state.prepared is None: result = await self.aorigin(compute) return await self.afinish( @@ -618,7 +697,7 @@ async def arun( state.epoch, reason=state.reason, ) - if state.decision.outcome == "HIT": + if state.decision is not None and state.decision.outcome == "HIT": # If cancellation loses this scheduled result, cached() still owns # and releases any elected flight in its worker finally block. try: @@ -654,7 +733,9 @@ async def arun( ) try: result = await self.aorigin(compute) - await cache._lifecycle.call(self.store, state, query, result, namespace, ttl, tags) + await cache._lifecycle.call( + self.store, state, query, result, namespace, ttl, tags_tuple + ) return await self.afinish( state.prepared, None, diff --git a/src/longcache/core/identity.py b/src/longcache/core/identity.py index 565a43c..0862300 100644 --- a/src/longcache/core/identity.py +++ b/src/longcache/core/identity.py @@ -2,13 +2,16 @@ from __future__ import annotations +from collections.abc import Mapping from dataclasses import dataclass, field from hashlib import sha256 import json import math -from collections.abc import Mapping from types import MappingProxyType -from typing import Any +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from .types import InstructionRole def freeze(value: Any) -> Any: @@ -50,22 +53,22 @@ class RequestScope: tenant_id: str user_id: str | None = None - def __post_init__(self): + def __post_init__(self) -> None: if type(self.tenant_id) is not str or not self.tenant_id.strip(): raise ValueError("A trusted nonempty tenant identifier is required") if self.user_id is not None and (type(self.user_id) is not str or not self.user_id.strip()): raise ValueError("User identifier must be nonempty when supplied") - def __repr__(self): + def __repr__(self) -> str: return "RequestScope(