服务器缓存按期限清理，审计快照默认不再落地

排盘缓存、官方快照、负缓存、网关任务和分盘段过期后会删掉。同一张盘的数字写法合成一把缓存键。失败快照只用短期限。自测留存、异步任务、增长合同、安全缓存和隐私标记；一万个文件的清理慢于 2 秒，命中率提升不到 2.5 倍，都按实测写入记录。

Co-Authored-By: Grok (agent loop)

scripts/vedastro_service_adapter.py

@@ -425,14 +425,23 @@ def _repo_relative(path: Path) -> str:
         return str(path)
 
 
-def _write_artifact(result: dict[str, Any]) -> str:
+def _audit_artifacts_enabled() -> bool:
+    raw = os.environ.get("VEDASTRO_WRITE_AUDIT_ARTIFACTS", "0").strip().lower()
+    return raw in {"1", "true", "yes", "on"}
+
+
+def _write_artifact(result: dict[str, Any]) -> str | None:
+    if not _audit_artifacts_enabled():
+        return None
     metadata = result.get("source_metadata") or {}
     artifact = _artifact_path(
         str(metadata.get("operation") or result.get("operation") or "calculation"),
         str(metadata.get("request_hash") or "no-request-hash"),
         str(metadata.get("response_hash") or "no-response-hash"),
     )
-    artifact.write_text(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
+    from api_scratch_retention import write_private_json
+
+    write_private_json(artifact, result)
     return _repo_relative(artifact)
 
 
@@ -486,6 +495,22 @@ def _official_full_snapshot_cache_ttl_seconds() -> float:
         return 0.0
 
 
+def _official_full_snapshot_negative_ttl_seconds() -> float:
+    normal = _official_full_snapshot_cache_ttl_seconds()
+    if normal <= 0:
+        return 0.0
+    raw = os.environ.get("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_NEGATIVE_TTL_SECONDS", "300").strip()
+    try:
+        negative = max(0.0, float(raw)) if raw else 300.0
+    except ValueError:
+        negative = 300.0
+    return min(negative, normal)
+
+
+def _official_full_snapshot_result_is_success(payload: dict[str, Any]) -> bool:
+    return payload.get("available") is True and payload.get("status") == "ok"
+
+
 def _free_tier_max_requests() -> int:
     raw = os.environ.get(FREE_TIER_MAX_REQUESTS_ENV, "").strip()
     if not raw:
@@ -1719,9 +1744,11 @@ def _attach_official_full_snapshot_semantic_cache_metadata(
     cache_key: str,
     created_at_unix: float,
     cache_hit: bool,
+    ttl_seconds: float | None = None,
 ) -> dict[str, Any]:
     payload_copy = json.loads(json.dumps(payload))
-    ttl_seconds = _official_full_snapshot_cache_ttl_seconds()
+    if ttl_seconds is None:
+        ttl_seconds = _official_full_snapshot_cache_ttl_seconds()
     created_at = datetime.utcfromtimestamp(created_at_unix).strftime("%Y-%m-%dT%H:%M:%SZ")
     expires_at = datetime.utcfromtimestamp(created_at_unix + ttl_seconds).strftime("%Y-%m-%dT%H:%M:%SZ")
     metadata = dict(payload_copy.get("source_metadata") or {})
@@ -1745,9 +1772,15 @@ def _rebind_official_full_snapshot_case_identity(payload: dict[str, Any], case_i
     return payload_copy
 
 
+def _semantic_cache_ttl_seconds(record: dict[str, Any]) -> float:
+    # Records written before the negative flag are successes and keep the long TTL.
+    if record.get("negative") is True:
+        return _official_full_snapshot_negative_ttl_seconds()
+    return _official_full_snapshot_cache_ttl_seconds()
+
+
 def _load_official_full_snapshot_semantic_cache(case: dict[str, Any], case_id: str) -> dict[str, Any] | None:
-    ttl_seconds = _official_full_snapshot_cache_ttl_seconds()
-    if ttl_seconds <= 0:
+    if _official_full_snapshot_cache_ttl_seconds() <= 0:
         return None
     cache_key = _official_full_snapshot_cache_key(case, case_id)
     cache_path = _official_full_snapshot_cache_path(cache_key)
@@ -1763,7 +1796,8 @@ def _load_official_full_snapshot_semantic_cache(case: dict[str, Any], case_id: s
     payload = record.get("payload")
     if not isinstance(created_at_raw, (int, float)) or not isinstance(payload, dict):
         return None
-    if time.time() - float(created_at_raw) > ttl_seconds:
+    ttl_seconds = _semantic_cache_ttl_seconds(record)
+    if ttl_seconds <= 0 or time.time() - float(created_at_raw) > ttl_seconds:
         return None
     rebound = _rebind_official_full_snapshot_case_identity(payload, case_id)
     return _attach_official_full_snapshot_semantic_cache_metadata(
@@ -1771,28 +1805,37 @@ def _load_official_full_snapshot_semantic_cache(case: dict[str, Any], case_id: s
         cache_key=cache_key,
         created_at_unix=float(created_at_raw),
         cache_hit=True,
+        ttl_seconds=ttl_seconds,
     )
 
 
 def _store_official_full_snapshot_semantic_cache(case: dict[str, Any], case_id: str, payload: dict[str, Any]) -> dict[str, Any]:
     cache_key = _official_full_snapshot_cache_key(case, case_id)
     created_at_unix = time.time()
+    negative = not _official_full_snapshot_result_is_success(payload)
+    ttl_seconds = (
+        _official_full_snapshot_negative_ttl_seconds()
+        if negative
+        else _official_full_snapshot_cache_ttl_seconds()
+    )
     payload_copy = _attach_official_full_snapshot_semantic_cache_metadata(
         payload,
         cache_key=cache_key,
         created_at_unix=created_at_unix,
         cache_hit=False,
+        ttl_seconds=ttl_seconds,
     )
-    ttl_seconds = _official_full_snapshot_cache_ttl_seconds()
     if ttl_seconds > 0:
-        record = {
-            "cache_key": cache_key,
-            "created_at": created_at_unix,
-            "payload": payload_copy,
-        }
-        _official_full_snapshot_cache_path(cache_key).write_text(
-            json.dumps(record, ensure_ascii=False, sort_keys=True),
-            encoding="utf-8",
+        from api_scratch_retention import write_private_json
+
+        write_private_json(
+            _official_full_snapshot_cache_path(cache_key),
+            {
+                "cache_key": cache_key,
+                "created_at": created_at_unix,
+                "negative": negative,
+                "payload": payload_copy,
+            },
         )
     return payload_copy
 
@@ -1882,7 +1925,9 @@ def _store_cached_payload(cache_key: str, payload: dict[str, Any]) -> dict[str,
             "created_at": created_at_unix,
             "payload": payload_copy,
         }
-        cache_path.write_text(json.dumps(record, ensure_ascii=False, sort_keys=True), encoding="utf-8")
+        from api_scratch_retention import write_private_json
+
+        write_private_json(cache_path, record)
     return payload_copy