feat(vuln): OSV cache staleness + refresh flags — closes #30

CHANGELOG.md

@@ -7,6 +7,61 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [8.2.0] — Unreleased
 
+### OSV Cache Staleness Flag + Refresh Flags (issue #30)
+
+`vuln-scan` now reports OSV cache freshness and lets agents force fresh
+data. Previously the 24h OSV cache had no staleness indicator and no way
+to force a re-fetch short of deleting the cache — agents relying on
+`vuln-scan` for security posture had no way to know whether the CVE
+results were minutes or 23 hours old.
+
+### Added (vuln-scan staleness)
+
+- **`cache_info` block in `vuln-scan` output** — additive top-level field:
+  ```json
+  "cache_info": {
+    "last_refresh": "2026-06-28T10:00:00Z",
+    "age_hours": 23.5,
+    "ttl_hours": 24,
+    "is_stale": false,
+    "stale_packages": ["requests@2.20.0"]
+  }
+  ```
+  `is_stale` is true when any cached entry is past the effective TTL, the
+  newest entry's age ≥ the effective TTL, or there are OSV-queriable
+  packages but none are cached (no fresh coverage). `stale_packages` lists
+  the `"name@version"` of cached entries past the effective TTL.
+- **`vuln-scan --refresh`** — bypasses the OSV cache and forces fresh
+  OSV.dev API calls for every package, then updates the cache with the new
+  results. No-op in `--offline` mode.
+- **`vuln-scan --max-age Nh`** — treats cache entries older than `N` hours
+  as stale for this run only (overrides the default 24h TTL without
+  changing the stored TTL). Accepts `6h`/`30m`/`2d`/`90s`/bare-integer
+  (hours). `cache_info.ttl_hours` and the staleness threshold honour this
+  override so they stay consistent. Ignored with `--refresh` and in
+  `--offline` mode.
+
+### Changed (vuln-scan staleness)
+
+- `scripts/osv_client.py`: new `OSVCache.peek(key)` — pure inspection
+  returning `(timestamp, ttl)` without mutating the cache (unlike `get()`,
+  which deletes expired entries). New `OSVClient.get_cache_info(packages,
+  max_age=None)`. `OSVClient.query_packages()` gained `force_refresh` and
+  `max_age` parameters (default behaviour unchanged).
+- `scripts/vulnscan_engine.py`: `scan_vulnerabilities()` gained `refresh`
+  and `max_age` parameters and now attaches `cache_info` to its result.
+- `scripts/commands/vuln_scan.py`: new `--refresh` and `--max-age` flags
+  plus a `_parse_max_age()` helper. Invalid `--max-age` values return an
+  `error` result instead of crashing.
+
+### Non-Breaking (vuln-scan staleness)
+
+- `cache_info` is additive — no existing `vuln-scan` output field is
+  changed or removed.
+- Default network behaviour is unchanged: the API is only contacted for
+  uncached packages, or when `--refresh` is set, or when `--max-age`
+  marks cached entries stale.
+
 ### Confidence Fields on Non-Deep Output (test fix)
 
 Previously, the `confidence` / `confidence_distribution` fields were only

README.md

@@ -121,7 +121,7 @@ python3 scripts/codelens.py query "myFunction" --lite
 | Command | Description |
 |---------|-------------|
 | `secrets [workspace] [--severity ...]` | Detect hardcoded API keys, passwords, tokens |
-| `vuln-scan [workspace]` | Scan dependencies for known CVEs (OSV.dev + native audit) |
+| `vuln-scan [workspace] [--refresh] [--max-age Nh]` | Scan dependencies for known CVEs (OSV.dev + native audit). `--refresh` bypasses the OSV cache; `--max-age Nh` treats cache entries older than N hours as stale for this run. Output includes a `cache_info` block (`last_refresh`, `age_hours`, `ttl_hours`, `is_stale`, `stale_packages`) — issue #30 |
 | `taint [workspace]` | Run AST-based taint analysis for vulnerability detection |
 | `dataflow [workspace] [--source] [--sink]` | Data flow taint analysis with cross-file call graph |
 | `env-check [workspace] [--var NAME]` | Audit environment variables |

SKILL-QUICK.md

@@ -126,7 +126,7 @@ $CLI list --limit 5 --offset 10 --format compact            # → paginated + co
 `entrypoints` · `api-map` · `state-map` · `detect` · `handbook` · `diff [--git-aware]` · `dashboard` · `history` · `graph-schema` · `resolve-types`
 
 ### Security (5)
-`secrets [--severity ...]` · `taint` (AST-based) · `dataflow [--source ...] [--sink ...]` (cross-file) · `vuln-scan` (OSV.dev + native audit) · `env-check [--var NAME]`
+`secrets [--severity ...]` · `taint` (AST-based) · `dataflow [--source ...] [--sink ...]` (cross-file) · `vuln-scan [--refresh] [--max-age Nh]` (OSV.dev + native audit; output includes `cache_info` staleness block — issue #30) · `env-check [--var NAME]`
 
 ### Quality (9)
 `smell [--categories ...] [--severity ...]` · `complexity [--name FN] [--threshold N] [--sort ...]` · `dead-code [--categories ...]` · `debug-leak [--category ...]` · `circular [--domain ...]` · `missing-refs` · `side-effect [--name FN]` · `perf-hint [--severity ...] [--category ...]` · `fix [--apply]`

scripts/commands/vuln_scan.py

@@ -1,9 +1,55 @@
 """Vuln-scan command — Scan dependencies for known CVEs using OSV.dev + native audit tools."""
 
+import re
+
 from vulnscan_engine import scan_vulnerabilities
 from commands import register_command
 
 
+# --max-age accepts a duration string like "6h", "30m", "2d", "90s", or a
+# bare integer (interpreted as hours, matching --osv-ttl semantics).
+_MAX_AGE_RE = re.compile(r"^\s*(\d+(?:\.\d+)?)\s*([hmsd]?)\s*$", re.IGNORECASE)
+_MAX_AGE_UNITS = {
+    "": 3600,    # bare number → hours
+    "h": 3600,
+    "m": 60,
+    "s": 1,
+    "d": 86400,
+}
+
+
+def _parse_max_age(raw):
+    """Parse a ``--max-age`` duration string into seconds.
+
+    Accepts forms like ``6h`` (6 hours), ``30m`` (30 minutes), ``2d``
+    (2 days), ``90s`` (90 seconds), or a bare integer (interpreted as
+    hours, matching ``--osv-ttl`` semantics).
+
+    Args:
+        raw: The raw string from argparse. May be ``None``.
+
+    Returns:
+        int number of seconds, or ``None`` if ``raw`` is ``None``.
+
+    Raises:
+        ValueError: If the string cannot be parsed or is not positive.
+    """
+    if raw is None:
+        return None
+    match = _MAX_AGE_RE.match(str(raw))
+    if match is None:
+        raise ValueError(
+            f"invalid --max-age value {raw!r} — expected forms like "
+            f"'6h', '30m', '2d', '90s', or a bare integer (hours)"
+        )
+    value = float(match.group(1))
+    unit = match.group(2).lower()
+    seconds = int(value * _MAX_AGE_UNITS[unit])
+    if seconds <= 0:
+        raise ValueError(f"--max-age must be positive, got {raw!r}")
+    return seconds
+
+
 def add_args(parser):
     parser.add_argument("workspace", nargs="?", default=None,
                         help="Path to workspace root (auto-detected if omitted)")
@@ -13,14 +59,35 @@ def add_args(parser):
                         help="Skip OSV.dev API queries (use cached data only)")
     parser.add_argument("--osv-ttl", type=int, default=86400,
                         help="OSV cache TTL in seconds (default: 86400 = 24h)")
+    parser.add_argument("--refresh", action="store_true", default=False,
+                        help="Bypass the OSV cache and force fresh API calls for "
+                             "every package (issue #30). Updates the cache with new "
+                             "results. Ignored in --offline mode.")
+    parser.add_argument("--max-age", dest="max_age", default=None,
+                        help="Treat OSV cache entries older than this as stale for "
+                             "this run only (issue #30). Examples: '6h' (6 hours), "
+                             "'30m' (30 minutes), '2d' (2 days), '90s' (90 seconds), "
+                             "or a bare integer (interpreted as hours). Overrides the "
+                             "default 24h TTL for this run; does not change stored TTL. "
+                             "Ignored in --offline mode and with --refresh.")
 
 
 def execute(args, workspace):
+    try:
+        max_age_seconds = _parse_max_age(getattr(args, "max_age", None))
+    except ValueError as exc:
+        return {
+            "status": "error",
+            "error": "invalid_argument",
+            "message": str(exc),
+        }
     return scan_vulnerabilities(
         workspace,
         severity=args.severity,
         offline=args.offline,
         osv_ttl=args.osv_ttl,
+        refresh=args.refresh,
+        max_age=max_age_seconds,
     )
 
 

scripts/osv_client.py

@@ -27,6 +27,7 @@
 from dataclasses import dataclass, field, asdict
 from typing import Dict, List, Any, Optional, Set, Tuple
 from collections import defaultdict
+from datetime import datetime, timezone
 from pathlib import Path
 
 try:
@@ -348,6 +349,33 @@ def get_stats(self) -> Dict[str, Any]:
             finally:
                 conn.close()
 
+    def peek(self, key: str) -> Optional[Tuple[float, int]]:
+        """Return ``(timestamp, ttl)`` for a cache key WITHOUT mutating the cache.
+
+        Unlike :meth:`get`, this is a pure inspection — it never deletes
+        expired or corrupt entries. Used by staleness reporting (issue #30)
+        so we can describe cache freshness without side effects.
+
+        Args:
+            key: Cache key (e.g., "npm|lodash|4.17.15")
+
+        Returns:
+            ``(timestamp, ttl)`` tuple if the key exists, else ``None``.
+        """
+        with self._lock:
+            conn = sqlite3.connect(self.db_path)
+            try:
+                cursor = conn.execute(
+                    "SELECT timestamp, ttl FROM cache WHERE package_ecosystem_version = ?",
+                    (key,)
+                )
+                row = cursor.fetchone()
+                if row is None:
+                    return None
+                return (float(row[0]), int(row[1]))
+            finally:
+                conn.close()
+
 
 # ─── Rate Limiter ──────────────────────────────────────────────
 
@@ -449,14 +477,26 @@ def query_single(
     def query_packages(
         self,
         packages: List[OSVPackage],
+        force_refresh: bool = False,
+        max_age: Optional[int] = None,
     ) -> List[OSVVulnerability]:
         """Query multiple packages against OSV.dev.
 
-        Uses batch queries for efficiency. Checks cache first,
-        then queries API for uncached packages.
+        Uses batch queries for efficiency. Checks cache first, then queries
+        the API for uncached (or stale) packages. Cached API responses are
+        updated with the fresh results.
 
         Args:
-            packages: List of OSVPackage objects to query
+            packages: List of OSVPackage objects to query.
+            force_refresh: If True, bypass the cache and force fresh API
+                calls for every package (issue #30 ``--refresh`` flag).
+                Ignored when ``offline`` is True (cannot hit the network).
+            max_age: Optional per-run TTL override in seconds. When set,
+                cached entries older than ``max_age`` are treated as stale
+                and re-fetched from the API for this run only (issue #30
+                ``--max-age`` flag). The stored TTL is unchanged. ``None``
+                means use each entry's stored TTL. Ignored when
+                ``force_refresh`` is True (cache is bypassed entirely).
 
         Returns:
             List of OSVVulnerability objects (deduplicated)
@@ -468,14 +508,38 @@ def query_packages(
         uncached_packages: List[OSVPackage] = []
         uncached_keys: List[str] = []
 
+        now = time.time()
+        # --refresh bypasses cache reads entirely (issue #30). It is a no-op
+        # in offline mode, where we cannot reach the API and must rely on
+        # whatever the cache already holds.
+        bypass_cache = bool(force_refresh) and not self.offline
+        # --max-age (issue #30): treat still-fresh entries as stale for this
+        # run only. Only meaningful when we are actually consulting the cache.
+        max_age_ttl = max_age if (max_age is not None and max_age > 0) else None
+
         # Check cache first
         for pkg in packages:
             if pkg.ecosystem is None:
                 continue  # Not supported by OSV
 
             cache_key = pkg.cache_key()
+
+            if bypass_cache:
+                uncached_packages.append(pkg)
+                uncached_keys.append(cache_key)
+                continue
+
             cached = self.cache.get(cache_key)
             if cached is not None:
+                # Honor --max-age: an entry that is still fresh per its
+                # stored TTL may be considered stale for this run.
+                if max_age_ttl is not None:
+                    peek = self.cache.peek(cache_key)
+                    if peek is not None and (now - peek[0]) > max_age_ttl:
+                        uncached_packages.append(pkg)
+                        uncached_keys.append(cache_key)
+                        continue
+
                 self._cache_hit_count += 1
 
                 # cached can be:
@@ -520,6 +584,91 @@ def batch_query(
         """
         return self.query_packages(packages)
 
+    def get_cache_info(
+        self,
+        packages: List[OSVPackage],
+        max_age: Optional[int] = None,
+    ) -> Dict[str, Any]:
+        """Return OSV cache freshness info for ``packages`` (issue #30).
+
+        Inspects the cache WITHOUT mutating it (uses :meth:`OSVCache.peek`).
+        Intended to be attached to ``vuln-scan`` output as ``cache_info`` so
+        agents can decide whether to trust the cached CVE data or re-scan.
+
+        Args:
+            packages: The OSV packages that were (or would be) queried.
+            max_age: Optional per-run TTL override in seconds. When set, the
+                effective TTL is ``max_age`` instead of the configured cache
+                TTL — entries older than this count as stale for this run.
+                ``None`` means use the configured cache TTL.
+
+        Returns:
+            Dict with::
+
+                {
+                  "last_refresh": "2026-06-28T10:00:00Z" | None,
+                  "age_hours": 23.5 | None,            # hours since last_refresh
+                  "ttl_hours": 24.0,                    # effective TTL in hours
+                  "is_stale": bool,
+                  "stale_packages": ["requests@2.20.0", ...]
+                }
+
+            ``last_refresh`` / ``age_hours`` are ``None`` when none of the
+            packages are cached. ``is_stale`` is True when any cached entry
+            is past the effective TTL, OR the newest entry's age >= the
+            effective TTL, OR there are OSV-queriable packages but none are
+            cached (no fresh coverage).
+        """
+        now = time.time()
+        configured_ttl = self.cache.ttl
+        effective_ttl = (
+            max_age if (max_age is not None and max_age > 0) else configured_ttl
+        )
+        ttl_hours = round(effective_ttl / 3600.0, 2)
+
+        timestamps: List[float] = []
+        stale_packages: List[str] = []
+        seen: Set[str] = set()
+
+        for pkg in packages:
+            if pkg.ecosystem is None:
+                continue  # Not supported by OSV — never cached
+            key = pkg.cache_key()
+            if key in seen:
+                continue
+            seen.add(key)
+
+            peek = self.cache.peek(key)
+            if peek is None:
+                continue  # Not cached yet
+            ts, _stored_ttl = peek
+            timestamps.append(ts)
+            if (now - ts) > effective_ttl:
+                stale_packages.append(f"{pkg.name}@{pkg.version}")
+
+        if timestamps:
+            last_refresh_ts = max(timestamps)
+            last_refresh = datetime.fromtimestamp(
+                last_refresh_ts, tz=timezone.utc
+            ).strftime("%Y-%m-%dT%H:%M:%SZ")
+            age_raw = (now - last_refresh_ts) / 3600.0
+            age_hours = round(age_raw, 2)
+            is_stale = bool(stale_packages) or age_raw >= (effective_ttl / 3600.0)
+        else:
+            last_refresh = None
+            age_hours = None
+            # OSV-queriable packages exist but none are cached → no fresh
+            # coverage, so the data is effectively stale.
+            is_stale = len(seen) > 0
+
+        return {
+            "last_refresh": last_refresh,
+            "age_hours": age_hours,
+            "ttl_hours": ttl_hours,
+            "is_stale": is_stale,
+            "stale_packages": stale_packages,
+        }
+
     def _batch_query_api(
         self,
         packages: List[OSVPackage],