Fix safe-prefix cache boundary bypass

sdk/src/unplug/api/cache.py

@@ -7,23 +7,29 @@
 from __future__ import annotations
 
 from unplug.core.runtime.cache import (
+ DEFAULT_PREFIX_OVERLAP_CHARS,
  CacheKeyParts,
  SafePrefixState,
  ScanCache,
  cache_key_parts,
+ chunk_storage_key,
+ effective_prefix_skip,
  merge_suffix_result,
  offset_findings,
  prefix_storage_key,
 )
 from unplug.core.runtime.versions import MODEL_VERSION_LOCAL, NORMALIZER_VERSION
 
 __all__ = [
+ "DEFAULT_PREFIX_OVERLAP_CHARS",
  "MODEL_VERSION_LOCAL",
  "NORMALIZER_VERSION",
  "CacheKeyParts",
  "SafePrefixState",
  "ScanCache",
  "cache_key_parts",
+ "chunk_storage_key",
+ "effective_prefix_skip",
  "merge_suffix_result",
  "offset_findings",
  "prefix_storage_key",

sdk/src/unplug/config/cache.py

@@ -4,6 +4,9 @@
 
 from pydantic import BaseModel, Field
 
+# Keep in sync with unplug.core.runtime.cache.DEFAULT_PREFIX_OVERLAP_CHARS.
+_DEFAULT_PREFIX_OVERLAP_CHARS = 256
+
 
 class CacheConfig(BaseModel):
  """Safe-prefix and chunk LRU settings."""
@@ -13,3 +16,5 @@ class CacheConfig(BaseModel):
  enabled: bool = True
  max_chunk_entries: int = Field(default=256, ge=1)
  advance_prefix_on_redact: bool = True
+ # Re-scan this many chars at the safe-prefix boundary (StreamScanner-aligned).
+ prefix_overlap_chars: int = Field(default=_DEFAULT_PREFIX_OVERLAP_CHARS, ge=0)

sdk/src/unplug/core/runtime/cache.py

@@ -11,6 +11,9 @@
 from unplug.api.types import Finding, ScanResult
 from unplug.core.runtime.versions import MODEL_VERSION_LOCAL, NORMALIZER_VERSION
 
+# Match StreamScanner default so append-only scans re-check the boundary.
+DEFAULT_PREFIX_OVERLAP_CHARS = 256
+
 
 def _sha256(text: str) -> str:
  return hashlib.sha256(text.encode("utf-8")).hexdigest()
@@ -39,6 +42,8 @@ class CacheKeyParts(BaseModel):
  full_hash: str
  normalizer_version: str
  model_version: str
+ source: str = ""
+ policy_fingerprint: str = ""
 
 
 def cache_key_parts(
@@ -47,18 +52,40 @@ def cache_key_parts(
  document_id: str | None,
  normalizer_version: str = NORMALIZER_VERSION,
  model_version: str = MODEL_VERSION_LOCAL,
+ source: str = "",
+ policy_fingerprint: str = "",
 ) -> CacheKeyParts:
  doc_key = f"doc:{document_id}" if document_id else f"hash:{_sha256(text)[:16]}"
  return CacheKeyParts(
  doc_key=doc_key,
  full_hash=_sha256(text),
  normalizer_version=normalizer_version,
  model_version=model_version,
+ source=source,
+ policy_fingerprint=policy_fingerprint,
  )
 
 
 def prefix_storage_key(parts: CacheKeyParts) -> str:
- return f"{parts.doc_key}|{parts.normalizer_version}|{parts.model_version}"
+ return (
+ f"{parts.doc_key}|{parts.source}|{parts.policy_fingerprint}|"
+ f"{parts.normalizer_version}|{parts.model_version}"
+ )
+
+
+def chunk_storage_key(parts: CacheKeyParts) -> str:
+ return (
+ f"{parts.full_hash}|{parts.source}|{parts.policy_fingerprint}|"
+ f"{parts.normalizer_version}|{parts.model_version}"
+ )
+
+
+def effective_prefix_skip(prefix_len: int, overlap_chars: int) -> int:
+ """Return the scan start offset, re-scanning ``overlap_chars`` at the boundary."""
+ if prefix_len <= 0:
+ return 0
+ overlap = max(0, overlap_chars)
+ return max(0, prefix_len - overlap)
 
 
 class ScanCache:
@@ -76,17 +103,24 @@ def cache_key_parts(
  document_id: str | None,
  normalizer_version: str = NORMALIZER_VERSION,
  model_version: str = MODEL_VERSION_LOCAL,
+ source: str = "",
+ policy_fingerprint: str = "",
  ) -> CacheKeyParts:
  return cache_key_parts(
  text,
  document_id=document_id,
  normalizer_version=normalizer_version,
  model_version=model_version,
+ source=source,
+ policy_fingerprint=policy_fingerprint,
  )
 
  def prefix_storage_key(self, parts: CacheKeyParts) -> str:
  return prefix_storage_key(parts)
 
+ def chunk_storage_key(self, parts: CacheKeyParts) -> str:
+ return chunk_storage_key(parts)
+
  def get_safe_prefix(self, parts: CacheKeyParts) -> SafePrefixState | None:
  return self._prefixes.get(prefix_storage_key(parts))
 
@@ -102,6 +136,12 @@ def set_chunk(self, content_hash: str, result: ScanResult) -> None:
  while len(self._chunks) > self._max_chunk_entries:
  self._chunks.popitem(last=False)
 
+ def get_chunk_for_parts(self, parts: CacheKeyParts) -> ScanResult | None:
+ return self.get_chunk(chunk_storage_key(parts))
+
+ def set_chunk_for_parts(self, parts: CacheKeyParts, result: ScanResult) -> None:
+ self.set_chunk(chunk_storage_key(parts), result)
+
  @staticmethod
  def should_advance_prefix(action: Action, *, advance_on_redact: bool) -> bool:
  if action == Action.BLOCK:

sdk/src/unplug/guard.py

@@ -22,7 +22,12 @@
 from unplug.core.policy import policy_from_request
 from unplug.core.privacy import NullPrivacyFilter, PrivacyFilterService
 from unplug.core.privacy.secrets import SecretsRegistry, SecretsSanitizer
-from unplug.core.runtime.cache import SafePrefixState, ScanCache, merge_suffix_result
+from unplug.core.runtime.cache import (
+ SafePrefixState,
+ ScanCache,
+ effective_prefix_skip,
+ merge_suffix_result,
+)
 from unplug.core.runtime.logging import correlation_scope, get_logger
 from unplug.core.runtime.model_runtime import (
  load_active_model_provider,
@@ -418,6 +423,7 @@ def stream_scanner(
  self,
  source=source,
  scan_every_chars=scan_every_chars,
+ overlap_chars=self._config.cache.prefix_overlap_chars,
  document_id=document_id,
  )
 
@@ -484,6 +490,37 @@ def _request_context(self, request: ScanRequest, *, isolated: bool) -> Execution
  def _model_version_for_cache(self) -> str:
  return self._model_cache_version
 
+ def _cache_policy_fingerprint(self, request: ScanRequest) -> str:
+ """Fingerprint policy + scanner set so cache entries are not reused across configs."""
+ policy = self._config.policy
+ scanners = request.scanners if request.scanners is not None else list(self._config.scanners)
+ parts = [
+ ",".join(scanners),
+ str(
+ request.block_threshold
+ if request.block_threshold is not None
+ else policy.block_threshold
+ ),
+ str(
+ request.redact_threshold
+ if request.redact_threshold is not None
+ else policy.redact_threshold
+ ),
+ str(
+ request.review_threshold
+ if request.review_threshold is not None
+ else policy.review_threshold
+ ),
+ str(
+ request.block_coverage_ratio
+ if request.block_coverage_ratio is not None
+ else policy.block_coverage_ratio
+ ),
+ str(request.redact),
+ str(request.redaction_mode.value if request.redaction_mode is not None else ""),
+ ]
+ return "|".join(parts)
+
  def _run_input_with_cache(self, request: ScanRequest, ctx: ExecutionContext) -> ScanResult:
  cache = ctx.scan_cache
  if cache is None or not self._config.cache.enabled:
@@ -498,8 +535,10 @@ def _run_input_with_cache(self, request: ScanRequest, ctx: ExecutionContext) ->
  document_id=request.document_id,
  normalizer_version=NORMALIZER_VERSION,
  model_version=self._model_version_for_cache(),
+ source=str(request.source),
+ policy_fingerprint=self._cache_policy_fingerprint(request),
  )
- hit = cache.get_chunk(parts.full_hash)
+ hit = cache.get_chunk_for_parts(parts)
  if hit is not None:
  return hit
 
@@ -508,13 +547,20 @@ def _run_input_with_cache(self, request: ScanRequest, ctx: ExecutionContext) ->
  if prefix_state is not None and prefix_state.verify(request.text):
  prefix_len = prefix_state.prefix_len
 
+ scan_start = 0
  if 0 < prefix_len < len(request.text):
+ scan_start = effective_prefix_skip(
+ prefix_len,
+ self._config.cache.prefix_overlap_chars,
+ )
+
+ if 0 < scan_start < len(request.text):
  suffix_result = self._input_pipeline.run(
- request.text[prefix_len:],
+ request.text[scan_start:],
  source=request.source,
  context=ctx,
  )
- result = merge_suffix_result(suffix_result, prefix_len)
+ result = merge_suffix_result(suffix_result, scan_start)
  else:
  result = self._input_pipeline.run(
  request.text,
@@ -530,7 +576,7 @@ def _run_input_with_cache(self, request: ScanRequest, ctx: ExecutionContext) ->
  parts,
  SafePrefixState.from_text(request.text, len(request.text)),
  )
- cache.set_chunk(parts.full_hash, result)
+ cache.set_chunk_for_parts(parts, result)
  return result
 
  def scan_output(self, text: str | TaintedText) -> ScanResult:

sdk/src/unplug/streaming.py

@@ -7,13 +7,17 @@
 
 from unplug.api.enums import Action, Source
 from unplug.api.types import ScanResult
-from unplug.core.runtime.cache import merge_suffix_result
+from unplug.core.runtime.cache import (
+ DEFAULT_PREFIX_OVERLAP_CHARS,
+ effective_prefix_skip,
+ merge_suffix_result,
+)
 
 if TYPE_CHECKING:
  from unplug.guard import Guard
 
 _DEFAULT_SCAN_EVERY = 1024
-_DEFAULT_OVERLAP = 256
+_DEFAULT_OVERLAP = DEFAULT_PREFIX_OVERLAP_CHARS
 
 
 class StreamScanner:
@@ -73,7 +77,7 @@ def _scan_accumulated(self) -> ScanResult:
  and self._last_result.action in (Action.ALLOW, Action.REVIEW)
  and self._last_result.safe
  ):
- prefix_len = max(0, self._safe_prefix_len - self._overlap)
+ prefix_len = effective_prefix_skip(self._safe_prefix_len, self._overlap)
 
  scan_body = body[prefix_len:] if prefix_len else body
  request = self._guard._build_scan_request(scan_body, self._source)