Merge pull request #36 from FluffyAIcode/AgentMemory/v030-pr7-6-bench-long-session-v2-8e7f

PR 7-6 (ADR 0007): bench_long_session scrapes path-selection metrics

Author: FluffyAIcode

scripts/bench_agentic/bench_long_session.py

@@ -36,27 +36,48 @@
 session can run for hours without OOM. A single agent driven for
 4 hours is therefore the cleanest evidence.
 
-A note on what this bench measures and what it doesn't (per the
-analysis of the 2026-05-30 Mac M4 run, ``bench_long_session_mac_
-1780130542.aborted.json``):
-
-  * **KV memory** stays bounded across hours (the §2.3 claim). The
-    ``scheduler_kv_live_bytes`` gauge is what proves it.
-  * **Per-turn latency** does NOT stay bounded. The OpenAI
-    chat-completions protocol is stateless: every turn the client
-    sends the full history, the server tokenizes it from scratch
-    and the verifier prefills the entire prompt, so prefill cost
-    grows linearly with history length. Sink+window only bounds
-    *generation-phase* memory, not prefill cost. A 30-min run on
-    Mac M4 showed p50 turn latency growing from ~15 s to ~55 s as
-    history grew from ~50 to ~3700 tokens. This is a **protocol-
-    level limitation**, not a memory-stability failure. Cross-
-    request KV reuse (a v0.4 feature) is the eventual fix; until
-    then, agent applications should manage prompt length via
-    summarization or sliding windows.
-
-The bench reports both metrics independently — KV bounded check is
-a hard claim, latency drift is a measurement, not a gate.
+What changed in PR 7-6 (ADR 0007 cross-request KV reuse)
+--------------------------------------------------------
+
+Before ADR 0007 (the 2026-05-30 4h Mac M4 run produced 58 useful
+turns then 3.5h of timeout/recovery): the bench was driving the
+v0.3.0-rc1 server, which reset the verifier cache on every
+chat-completions request. Per-turn prefill cost grew O(history),
+exceeded the 120s client timeout around turn 58, and the run
+degenerated.
+
+After ADR 0007 (PRs 7-1 to 7-5): the server now takes the
+continuation path on prompts that extend the cached state. Per-
+turn prefill cost is O(new_user_message) instead of O(history).
+The 4h re-run with the same workload should now produce
+hundreds of useful turns and never enter the timeout/recovery
+regime.
+
+This bench scrapes the new ADR §2.10 metrics on every turn:
+
+  * ``path_selection_total{path=continuation}``    Counter
+  * ``path_selection_total{path=new_session}``     Counter
+  * ``continuation_tokens_skipped_total``           Counter
+  * ``cache_invariant_violations_total{kind=...}``  Counter (must be 0)
+
+The aggregate report now includes an ``adr_0007`` block with:
+
+  continuation_decisions       int    (count this run)
+  new_session_decisions        int
+  total_decisions              int
+  continuation_rate            float in [0, 1]
+  tokens_skipped               int    (total prefill tokens reused)
+  cache_invariant_inv1_total   int    (must be 0)
+  cache_invariant_inv2_total   int    (must be 0)
+
+GA gate criteria (per ADR 0007 §6, applied to the 4h Mac M4 re-run):
+
+  1. ``agg.kv_bounded`` is True (memory bound holds across 4h)
+  2. ``agg.n_errors`` < 5 (no sustained timeout/recovery loop)
+  3. ``agg.n_turns`` >= 200 (vs 58 in v0.3.0-rc1)
+  4. ``agg.latency_drift_p50_s`` <= 5 seconds (vs +39.74s in v0.3.0-rc1)
+  5. ``adr_0007.continuation_rate`` >= 0.95
+  6. ``adr_0007.cache_invariant_inv1_total + inv2_total`` == 0
 
 Usage
 -----
@@ -166,6 +187,20 @@ def _client_rss_bytes() -> Optional[int]:
     "scheduler_kv_live_bytes",
 )
 
+# ADR 0007 §2.10 path-selection metrics — labeled, parsed separately
+# from the unlabeled scheduler gauges. continuation_rate is the
+# headline KPI for the cross-request KV reuse fix: ≥95% on a healthy
+# long-session agent run.
+_PATH_SELECTION_METRIC = "path_selection_total"
+_CONTINUATION_TOKENS_SKIPPED_METRIC = "continuation_tokens_skipped_total"
+_CACHE_INVARIANT_VIOLATIONS_METRIC = "cache_invariant_violations_total"
+
+# Match a Prometheus exposition line with labels: e.g.
+# `path_selection_total{path="continuation"} 5.0`
+_LABELED_METRIC_LINE = re.compile(
+    r'^(?P<name>[a-zA-Z_:][a-zA-Z0-9_:]*)\{(?P<labels>[^}]*)\}\s+(?P<value>[-+0-9eE.\.NaNinf]+)\s*$'
+)
+
 
 _METRIC_LINE = re.compile(
     r"^(?P<name>[a-zA-Z_:][a-zA-Z0-9_:]*)(\{[^}]*\})?\s+(?P<value>[-+0-9eE.\.NaNinf]+)\s*$"
@@ -175,28 +210,75 @@ def _client_rss_bytes() -> Optional[int]:
 def _parse_prom_text(body: str) -> dict[str, float]:
     """Tiny Prometheus text parser.
 
-    Only the four metrics we care about are extracted; everything
-    else is skipped. Values are coerced to float; ``NaN`` and ``inf``
-    are preserved as floats. We do not try to handle multi-label
-    series — these are gauges with no labels in our exporter.
+    Returns a flat dict where unlabeled metrics keep their bare
+    name and labeled ADR 0007 §2.10 metrics get a synthesized name
+    of the form ``"{base}__{label_value}"``:
+
+      * ``path_selection_total{path="continuation"}`` →
+        ``"path_selection_total__continuation"``
+      * ``path_selection_total{path="new_session"}`` →
+        ``"path_selection_total__new_session"``
+      * ``cache_invariant_violations_total{kind="inv1"}`` →
+        ``"cache_invariant_violations_total__inv1"``
+      * ``continuation_tokens_skipped_total`` (no label) →
+        ``"continuation_tokens_skipped_total"``
+
+    Values are coerced to float; ``NaN`` and ``inf`` are preserved.
     """
     out: dict[str, float] = {}
     for line in body.splitlines():
         if not line or line.startswith("#"):
             continue
+        # Try labeled match first (more specific).
+        labeled = _LABELED_METRIC_LINE.match(line)
+        if labeled is not None:
+            name = labeled.group("name")
+            label_str = labeled.group("labels")
+            try:
+                value = float(labeled.group("value"))
+            except ValueError:  # pragma: no cover - malformed exporter
+                continue
+            if name == _PATH_SELECTION_METRIC:
+                # Extract the path label (continuation|new_session)
+                path = _extract_label(label_str, "path")
+                if path:
+                    out[f"{name}__{path}"] = value
+            elif name == _CACHE_INVARIANT_VIOLATIONS_METRIC:
+                kind = _extract_label(label_str, "kind")
+                if kind:
+                    out[f"{name}__{kind}"] = value
+            continue
+        # Unlabeled.
         m = _METRIC_LINE.match(line)
         if m is None:
             continue
         name = m.group("name")
-        if name not in _METRIC_NAMES:
-            continue
-        try:
-            out[name] = float(m.group("value"))
-        except ValueError:  # pragma: no cover - malformed exporter
-            continue
+        if name in _METRIC_NAMES or name == _CONTINUATION_TOKENS_SKIPPED_METRIC:
+            try:
+                out[name] = float(m.group("value"))
+            except ValueError:  # pragma: no cover - malformed exporter
+                continue
     return out
 
 
+def _extract_label(label_str: str, key: str) -> Optional[str]:
+    """Pull a single label value out of a Prometheus label segment.
+
+    Input is the part between the curly braces, e.g.
+    ``path="continuation"`` or
+    ``path="continuation",result="ok"``. Returns the value (without
+    quotes) for the requested key, or ``None`` if absent.
+    """
+    for fragment in label_str.split(","):
+        fragment = fragment.strip()
+        if "=" not in fragment:
+            continue
+        k, v = fragment.split("=", 1)
+        if k.strip() == key:
+            return v.strip().strip('"')
+    return None
+
+
 async def _scrape_metrics(
     client: httpx.AsyncClient,
     *,
@@ -669,6 +751,58 @@ def _aggregate(
     }
 
 
+def _adr_0007_summary(turns: list[dict[str, Any]]) -> dict[str, Any]:
+    """ADR 0007 §2.10 path-selection summary from the per-turn idle
+    scrapes (`metrics_idle`).
+
+    The path_selection_total counters are CUMULATIVE across the
+    server's lifetime, so we take the LAST observed value (= final
+    counter at end of run) and the FIRST observed value (= counter
+    at start of run, may be > 0 if the server already handled
+    requests before this bench started). Difference = decisions
+    this run made.
+
+    Returns a dict with:
+      continuation_decisions       int
+      new_session_decisions        int
+      total_decisions              int
+      continuation_rate            float in [0, 1] (None if 0 decisions)
+      tokens_skipped               int (delta of the counter)
+      cache_invariant_inv1_total   int (last - first; should be 0)
+      cache_invariant_inv2_total   int (last - first; should be 0)
+    """
+    def _delta(name: str) -> int:
+        first: Optional[float] = None
+        last: Optional[float] = None
+        for t in turns:
+            idle = t.get("metrics_idle") or {}
+            v = idle.get(name)
+            if v is None:
+                continue
+            if first is None:
+                first = v
+            last = v
+        if first is None or last is None:
+            return 0
+        return int(round(last - first))
+
+    cont = _delta(f"{_PATH_SELECTION_METRIC}__continuation")
+    news = _delta(f"{_PATH_SELECTION_METRIC}__new_session")
+    skipped = _delta(_CONTINUATION_TOKENS_SKIPPED_METRIC)
+    inv1 = _delta(f"{_CACHE_INVARIANT_VIOLATIONS_METRIC}__inv1")
+    inv2 = _delta(f"{_CACHE_INVARIANT_VIOLATIONS_METRIC}__inv2")
+    total = cont + news
+    return {
+        "continuation_decisions": cont,
+        "new_session_decisions": news,
+        "total_decisions": total,
+        "continuation_rate": (cont / total) if total > 0 else None,
+        "tokens_skipped": skipped,
+        "cache_invariant_inv1_total": inv1,
+        "cache_invariant_inv2_total": inv2,
+    }
+
+
 def _build_payload(
     *,
     turns: list[dict[str, Any]],
@@ -687,6 +821,7 @@ def _build_payload(
         "turns": turns,
         "errors": errors,
         "agg": _aggregate(turns, errors, duration_s),
+        "adr_0007": _adr_0007_summary(turns),
     }
 
 
@@ -745,6 +880,42 @@ def _fmt_bytes(b: Optional[float]) -> str:
                 f"{b['p95_latency_s']:>7.3f}s  "
                 f"{(mkb / (1024 * 1024)) if mkb is not None else float('nan'):>7.1f} MiB"
             )
+    # ADR 0007 §2.10 path-selection block (only emitted when the
+    # payload actually carries the data — backward-compat with old
+    # checkpoints that pre-date PR 7-6).
+    adr_0007 = payload.get("adr_0007")
+    if adr_0007 is not None:
+        rate = adr_0007.get("continuation_rate")
+        rate_str = (
+            f"{rate * 100:.2f}%" if rate is not None else "n/a (no decisions)"
+        )
+        lines.append("")
+        lines.append("  ADR 0007 §2.10 — cross-request KV reuse")
+        lines.append(
+            f"    continuation decisions  = "
+            f"{adr_0007['continuation_decisions']}"
+        )
+        lines.append(
+            f"    new-session decisions   = "
+            f"{adr_0007['new_session_decisions']}"
+        )
+        lines.append(f"    continuation rate       = {rate_str}")
+        lines.append(
+            f"    total tokens skipped    = "
+            f"{adr_0007['tokens_skipped']:,}"
+        )
+        inv1 = adr_0007["cache_invariant_inv1_total"]
+        inv2 = adr_0007["cache_invariant_inv2_total"]
+        inv_marker = "" if (inv1 == 0 and inv2 == 0) else "  ← CRITICAL"
+        lines.append(
+            f"    INV-1 violations        = {inv1}"
+            + (inv_marker if inv1 > 0 else "")
+        )
+        lines.append(
+            f"    INV-2 violations        = {inv2}"
+            + (inv_marker if inv2 > 0 else "")
+        )
+
     lines.append("=" * 78)
     return "\n".join(lines)