From 56764d97b017fbc4cacd4a529da03f26a510cee5 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Tue, 16 Jun 2026 18:46:49 +0000 Subject: [PATCH 01/21] =?UTF-8?q?=E2=9A=A1=20Bolt:=20Throttle=20synchronou?= =?UTF-8?q?s=20disk=20I/O=20in=20stats=20tracking?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: sheepdestroyer <1377479+sheepdestroyer@users.noreply.github.com> --- .jules/bolt.md | 3 +++ router/main.py | 30 +++++++++++++++++++++++------- 2 files changed, 26 insertions(+), 7 deletions(-) create mode 100644 .jules/bolt.md diff --git a/.jules/bolt.md b/.jules/bolt.md new file mode 100644 index 00000000..f2935c10 --- /dev/null +++ b/.jules/bolt.md @@ -0,0 +1,3 @@ +## 2024-06-16 - Synchronous I/O in Async API Handlers +**Learning:** `save_persisted_stats()` was being called synchronously on every API request, cache hit, and tool usage log, triggering blocking disk I/O in the main event loop. +**Action:** Always throttle or batch background telemetry writes in async Python applications to prevent blocking the event loop under load. diff --git a/router/main.py b/router/main.py index 64efce64..c874e629 100644 --- a/router/main.py +++ b/router/main.py @@ -151,8 +151,20 @@ def load_persisted_stats(): except Exception as e: logger.error(f"Failed to load persisted stats: {e}") -def save_persisted_stats(): +_last_stats_save = 0.0 + +def save_persisted_stats(force=False): """Persists current statistics in-memory structure to disk securely.""" + global _last_stats_save + import time + now = time.time() + + # Throttle disk writes to max once per 2 seconds, unless forced + if not force and (now - _last_stats_save < 2.0): + return + + _last_stats_save = now + try: os.makedirs(os.path.dirname(STATS_JSON_PATH), exist_ok=True) with open(STATS_JSON_PATH, "w") as f: @@ -626,12 +638,16 @@ def record_tool_usage(tool_name: str, prompt_tokens: int, completion_tokens: int if len(stats["timeline"]) > 15: stats["timeline"].pop(0) save_persisted_stats() - try: - timeline_path = os.path.join(os.path.dirname(CONFIG_PATH), "router_timeline.json") - with open(timeline_path, "w") as f: - json.dump(stats["timeline"], f) - except Exception: - pass # disk persistence failure is non-fatal + # Throttled timeline save (handled together with stats save internally, + # but we also explicitly throttle this specific file) + if time.time() - getattr(record_tool_usage, "_last_save", 0.0) >= 2.0: + try: + timeline_path = os.path.join(os.path.dirname(CONFIG_PATH), "router_timeline.json") + with open(timeline_path, "w") as f: + json.dump(stats["timeline"], f) + record_tool_usage._last_save = time.time() + except Exception: + pass # disk persistence failure is non-fatal def get_goose_sessions() -> list: """Queries the live mounted SQLite goose database to fetch the latest agentic sessions.""" From 29dbea71eaa727f19b146d35294ee41330eea003 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Tue, 16 Jun 2026 19:03:11 +0000 Subject: [PATCH 02/21] =?UTF-8?q?=E2=9A=A1=20Bolt:=20Throttle=20synchronou?= =?UTF-8?q?s=20disk=20I/O=20in=20stats=20tracking?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: sheepdestroyer <1377479+sheepdestroyer@users.noreply.github.com> From d2f3b06eac2325b92c1eb6580cb4f305a0e8a784 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Tue, 16 Jun 2026 19:14:56 +0000 Subject: [PATCH 03/21] =?UTF-8?q?=E2=9A=A1=20Bolt:=20Throttle=20synchronou?= =?UTF-8?q?s=20disk=20I/O=20in=20stats=20tracking?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: sheepdestroyer <1377479+sheepdestroyer@users.noreply.github.com> From bc052b6efba2c15f2feaf76f6b6b86651b766c51 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Wed, 17 Jun 2026 20:12:44 +0000 Subject: [PATCH 04/21] =?UTF-8?q?=E2=9A=A1=20Bolt:=20Throttle=20synchronou?= =?UTF-8?q?s=20disk=20I/O=20in=20stats=20tracking?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: sheepdestroyer <1377479+sheepdestroyer@users.noreply.github.com> From 808a28d04f39e5bfd28b530db0ec4ccbf0c2f374 Mon Sep 17 00:00:00 2001 From: sheepdestroyer Date: Thu, 18 Jun 2026 00:14:54 +0200 Subject: [PATCH 05/21] fix: address CodeRabbit and Gemini review feedback - Move _last_stats_save inside try block (only advance on successful write) - Add save_persisted_stats(force=True) + timeline flush in lifespan shutdown - Remove redundant import time inside save_persisted_stats - Fix misleading comment about independent throttle timers - Use time.monotonic() instead of time.time() for throttle timestamps - Add logger.warning to bare except in timeline write --- router/main.py | 25 +++++++++++++++---------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/router/main.py b/router/main.py index c874e629..20b0dbbf 100644 --- a/router/main.py +++ b/router/main.py @@ -156,19 +156,17 @@ def load_persisted_stats(): def save_persisted_stats(force=False): """Persists current statistics in-memory structure to disk securely.""" global _last_stats_save - import time - now = time.time() + now = time.monotonic() # Throttle disk writes to max once per 2 seconds, unless forced if not force and (now - _last_stats_save < 2.0): return - _last_stats_save = now - try: os.makedirs(os.path.dirname(STATS_JSON_PATH), exist_ok=True) with open(STATS_JSON_PATH, "w") as f: json.dump(stats, f, indent=2) + _last_stats_save = now # only advance on successful write except Exception as e: logger.error(f"Failed to persist stats to disk: {e}") @@ -345,6 +343,14 @@ async def lifespan(app: FastAPI): except Exception as e: logger.error(f"Roster sync failed: {e}") yield + # Flush any buffered stats/timeline on clean shutdown + save_persisted_stats(force=True) + try: + timeline_path = os.path.join(os.path.dirname(CONFIG_PATH), "router_timeline.json") + with open(timeline_path, "w") as f: + json.dump(stats["timeline"], f) + except Exception: + pass # Start aggregate score-push background task (runs for server lifetime) asyncio.create_task(push_aggregate_scores()) @@ -638,16 +644,15 @@ def record_tool_usage(tool_name: str, prompt_tokens: int, completion_tokens: int if len(stats["timeline"]) > 15: stats["timeline"].pop(0) save_persisted_stats() - # Throttled timeline save (handled together with stats save internally, - # but we also explicitly throttle this specific file) - if time.time() - getattr(record_tool_usage, "_last_save", 0.0) >= 2.0: + # Throttle timeline file writes independently of the stats file (max once per 2 s) + if time.monotonic() - getattr(record_tool_usage, "_last_save", 0.0) >= 2.0: try: timeline_path = os.path.join(os.path.dirname(CONFIG_PATH), "router_timeline.json") with open(timeline_path, "w") as f: json.dump(stats["timeline"], f) - record_tool_usage._last_save = time.time() - except Exception: - pass # disk persistence failure is non-fatal + record_tool_usage._last_save = time.monotonic() + except Exception as e: + logger.warning(f"Failed to persist timeline: {e}") def get_goose_sessions() -> list: """Queries the live mounted SQLite goose database to fetch the latest agentic sessions.""" From 60da6e1a0612b2c42b663dcc1f49304aac3c401a Mon Sep 17 00:00:00 2001 From: sheepdestroyer Date: Thu, 18 Jun 2026 00:23:33 +0200 Subject: [PATCH 06/21] fix: address CodeRabbit and Gemini round-2 review feedback - Add logging to bare except Exception: pass in shutdown timeline flush - Move asyncio.create_task(push_aggregate_scores()) before yield so it runs during app lifetime - Use atomic file writes (temp file + os.replace) in save_persisted_stats - Use atomic file writes in shutdown timeline flush and record_tool_usage timeline write - Restructure lifespan to single yield point (remove else: yield; return pattern) - Capture time.monotonic() once in record_tool_usage and reuse for guard + assignment --- router/main.py | 73 +++++++++++++++++++++++++++++++++++++------------- 1 file changed, 54 insertions(+), 19 deletions(-) diff --git a/router/main.py b/router/main.py index 20b0dbbf..3c734759 100644 --- a/router/main.py +++ b/router/main.py @@ -5,6 +5,7 @@ import socket import asyncio import logging +import tempfile import yaml import httpx from contextlib import asynccontextmanager @@ -164,8 +165,18 @@ def save_persisted_stats(force=False): try: os.makedirs(os.path.dirname(STATS_JSON_PATH), exist_ok=True) - with open(STATS_JSON_PATH, "w") as f: - json.dump(stats, f, indent=2) + # Atomic write via temp file + os.replace to prevent file corruption + fd, tmp_path = tempfile.mkstemp(dir=os.path.dirname(STATS_JSON_PATH), suffix=".tmp") + try: + with os.fdopen(fd, "w") as f: + json.dump(stats, f, indent=2) + os.replace(tmp_path, STATS_JSON_PATH) + except Exception: + try: + os.unlink(tmp_path) + except Exception: + pass + raise _last_stats_save = now # only advance on successful write except Exception as e: logger.error(f"Failed to persist stats to disk: {e}") @@ -335,24 +346,37 @@ async def lifespan(app: FastAPI): await asyncio.sleep(1) else: logger.warning("⚠️ LiteLLM not ready within timeout — proceeding without roster sync") - yield - return - # Sync free-model roster into LiteLLM (separate try so sync failure doesn't loop) - try: - await sync_adaptive_router_roster(litellm_master_key) - except Exception as e: - logger.error(f"Roster sync failed: {e}") + + # Sync free-model roster into LiteLLM (non-fatal if it fails) + if litellm_master_key: + try: + await sync_adaptive_router_roster(litellm_master_key) + except Exception as e: + logger.error(f"Roster sync failed: {e}") + + # Start background task before yield so it runs during app lifetime + asyncio.create_task(push_aggregate_scores()) + yield + # Flush any buffered stats/timeline on clean shutdown save_persisted_stats(force=True) try: timeline_path = os.path.join(os.path.dirname(CONFIG_PATH), "router_timeline.json") - with open(timeline_path, "w") as f: - json.dump(stats["timeline"], f) - except Exception: - pass - # Start aggregate score-push background task (runs for server lifetime) - asyncio.create_task(push_aggregate_scores()) + os.makedirs(os.path.dirname(timeline_path), exist_ok=True) + fd, tmp_path = tempfile.mkstemp(dir=os.path.dirname(timeline_path), suffix=".tmp") + try: + with os.fdopen(fd, "w") as f: + json.dump(stats["timeline"], f) + os.replace(tmp_path, timeline_path) + except Exception: + try: + os.unlink(tmp_path) + except Exception: + pass + raise + except Exception as e: + logger.warning(f"Failed to persist timeline on shutdown: {e}") app = FastAPI(title="LLM Triage Router", lifespan=lifespan) @@ -645,12 +669,23 @@ def record_tool_usage(tool_name: str, prompt_tokens: int, completion_tokens: int stats["timeline"].pop(0) save_persisted_stats() # Throttle timeline file writes independently of the stats file (max once per 2 s) - if time.monotonic() - getattr(record_tool_usage, "_last_save", 0.0) >= 2.0: + now = time.monotonic() + if now - getattr(record_tool_usage, "_last_save", 0.0) >= 2.0: try: timeline_path = os.path.join(os.path.dirname(CONFIG_PATH), "router_timeline.json") - with open(timeline_path, "w") as f: - json.dump(stats["timeline"], f) - record_tool_usage._last_save = time.monotonic() + os.makedirs(os.path.dirname(timeline_path), exist_ok=True) + fd, tmp_path = tempfile.mkstemp(dir=os.path.dirname(timeline_path), suffix=".tmp") + try: + with os.fdopen(fd, "w") as f: + json.dump(stats["timeline"], f) + os.replace(tmp_path, timeline_path) + except Exception: + try: + os.unlink(tmp_path) + except Exception: + pass + raise + record_tool_usage._last_save = now except Exception as e: logger.warning(f"Failed to persist timeline: {e}") From 418f90c32d7bc5c074803edcc1a2285ad2e277dc Mon Sep 17 00:00:00 2001 From: sheepdestroyer Date: Thu, 18 Jun 2026 00:55:51 +0200 Subject: [PATCH 07/21] =?UTF-8?q?fix:=20address=20Gemini=20round-3=20revie?= =?UTF-8?q?w=20=E2=80=94=20offload=20sync=20I/O=20to=20thread=20pool?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Extract shared _atomic_write_json_sync / _atomic_write_json_async helpers - Make save_persisted_stats async, using _atomic_write_json_async with run_in_executor + deep-copy to prevent concurrent modification - Update all 4 async call sites (lifespan, classify_request x2, chat_completions) to await save_persisted_stats() - In record_tool_usage (sync), fire-and-forget both stats and timeline writes via loop.run_in_executor with deep-copied data; fall back to sync write when no event loop is running (early startup) - Replace duplicated atomic write logic in lifespan shutdown with shared helper --- router/main.py | 123 ++++++++++++++++++++++++++++++------------------- 1 file changed, 76 insertions(+), 47 deletions(-) diff --git a/router/main.py b/router/main.py index 3c734759..cce9cfca 100644 --- a/router/main.py +++ b/router/main.py @@ -5,6 +5,7 @@ import socket import asyncio import logging +import copy import tempfile import yaml import httpx @@ -152,10 +153,42 @@ def load_persisted_stats(): except Exception as e: logger.error(f"Failed to load persisted stats: {e}") +def _atomic_write_json_sync(path: str, data) -> None: + """Synchronously write JSON data to path using atomic temp-file + os.replace.""" + os.makedirs(os.path.dirname(path), exist_ok=True) + fd, tmp_path = tempfile.mkstemp(dir=os.path.dirname(path), suffix=".tmp") + try: + with os.fdopen(fd, "w") as f: + json.dump(data, f, indent=2) + os.replace(tmp_path, path) + except Exception: + try: + os.unlink(tmp_path) + except Exception: + pass + raise + + +async def _atomic_write_json_async(path: str, data) -> None: + """Asynchronously write JSON data to path via thread pool executor. + + Deep-copies the data to prevent concurrent modification from the main thread + while the executor thread is serializing it. + """ + loop = asyncio.get_running_loop() + data_copy = copy.deepcopy(data) + await loop.run_in_executor(None, _atomic_write_json_sync, path, data_copy) + + _last_stats_save = 0.0 -def save_persisted_stats(force=False): - """Persists current statistics in-memory structure to disk securely.""" +async def save_persisted_stats(force=False): + """Persists current statistics in-memory structure to disk securely (non-blocking). + + Offloads the synchronous file write to a thread pool executor so the + event loop is not blocked. The 2-second throttle is checked before + dispatching. + """ global _last_stats_save now = time.monotonic() @@ -164,19 +197,7 @@ def save_persisted_stats(force=False): return try: - os.makedirs(os.path.dirname(STATS_JSON_PATH), exist_ok=True) - # Atomic write via temp file + os.replace to prevent file corruption - fd, tmp_path = tempfile.mkstemp(dir=os.path.dirname(STATS_JSON_PATH), suffix=".tmp") - try: - with os.fdopen(fd, "w") as f: - json.dump(stats, f, indent=2) - os.replace(tmp_path, STATS_JSON_PATH) - except Exception: - try: - os.unlink(tmp_path) - except Exception: - pass - raise + await _atomic_write_json_async(STATS_JSON_PATH, stats) _last_stats_save = now # only advance on successful write except Exception as e: logger.error(f"Failed to persist stats to disk: {e}") @@ -360,21 +381,10 @@ async def lifespan(app: FastAPI): yield # Flush any buffered stats/timeline on clean shutdown - save_persisted_stats(force=True) + await save_persisted_stats(force=True) try: timeline_path = os.path.join(os.path.dirname(CONFIG_PATH), "router_timeline.json") - os.makedirs(os.path.dirname(timeline_path), exist_ok=True) - fd, tmp_path = tempfile.mkstemp(dir=os.path.dirname(timeline_path), suffix=".tmp") - try: - with os.fdopen(fd, "w") as f: - json.dump(stats["timeline"], f) - os.replace(tmp_path, timeline_path) - except Exception: - try: - os.unlink(tmp_path) - except Exception: - pass - raise + await _atomic_write_json_async(timeline_path, stats["timeline"]) except Exception as e: logger.warning(f"Failed to persist timeline on shutdown: {e}") @@ -417,7 +427,7 @@ async def classify_request(prompt: str, bypass_cache: bool = False, langfuse_tra if time.time() - cached_time < CACHE_TTL_SECONDS: logger.info(f"⚡ Triage Cache Hit for prompt: '{normalized_prompt[:50]}...' -> routed to '{cached_decision}'") stats["cache_hits"] = stats.get("cache_hits", 0) + 1 - save_persisted_stats() + await save_persisted_stats() return cached_decision, 0.0, True, cached_decision # was_cache_hit=True start_time = time.time() @@ -430,7 +440,7 @@ async def classify_request(prompt: str, bypass_cache: bool = False, langfuse_tra if time.time() - cached_time < CACHE_TTL_SECONDS: logger.info(f"⚡ Triage Cache Hit (post-queue) for prompt: '{normalized_prompt[:50]}...' -> routed to '{cached_decision}'") stats["cache_hits"] = stats.get("cache_hits", 0) + 1 - save_persisted_stats() + await save_persisted_stats() return cached_decision, 0.0, True, cached_decision try: @@ -639,7 +649,12 @@ def detect_active_tool(body: dict) -> str: return "none" def record_tool_usage(tool_name: str, prompt_tokens: int, completion_tokens: int, model: str, latency_ms: float, route: str = "litellm_fallback"): - """Accumulates token counts in memory for active tools and tracks request timelines.""" + """Accumulates token counts in memory for active tools and tracks request timelines. + + File writes are offloaded to a thread pool executor to avoid blocking the + event loop. The 2-second throttle is checked synchronously before + dispatching. + """ if tool_name == "none": tool_name = "other" @@ -667,25 +682,39 @@ def record_tool_usage(tool_name: str, prompt_tokens: int, completion_tokens: int stats["timeline"].append(event) if len(stats["timeline"]) > 15: stats["timeline"].pop(0) - save_persisted_stats() - # Throttle timeline file writes independently of the stats file (max once per 2 s) + + # Fire-and-forget stats write via thread pool (non-blocking) + global _last_stats_save now = time.monotonic() - if now - getattr(record_tool_usage, "_last_save", 0.0) >= 2.0: + if now - _last_stats_save >= 2.0: try: - timeline_path = os.path.join(os.path.dirname(CONFIG_PATH), "router_timeline.json") - os.makedirs(os.path.dirname(timeline_path), exist_ok=True) - fd, tmp_path = tempfile.mkstemp(dir=os.path.dirname(timeline_path), suffix=".tmp") + loop = asyncio.get_running_loop() + loop.run_in_executor(None, _atomic_write_json_sync, STATS_JSON_PATH, copy.deepcopy(dict(stats))) + _last_stats_save = now + except RuntimeError: + # No running event loop (e.g. during early startup) — fall back to sync write try: - with os.fdopen(fd, "w") as f: - json.dump(stats["timeline"], f) - os.replace(tmp_path, timeline_path) - except Exception: - try: - os.unlink(tmp_path) - except Exception: - pass - raise + _atomic_write_json_sync(STATS_JSON_PATH, stats) + _last_stats_save = now + except Exception as e: + logger.error(f"Failed to persist stats to disk: {e}") + except Exception as e: + logger.error(f"Failed to persist stats to disk: {e}") + + # Throttle timeline file writes independently of the stats file (max once per 2 s) + timeline_path = os.path.join(os.path.dirname(CONFIG_PATH), "router_timeline.json") + if now - getattr(record_tool_usage, "_last_save", 0.0) >= 2.0: + try: + loop = asyncio.get_running_loop() + loop.run_in_executor(None, _atomic_write_json_sync, timeline_path, copy.deepcopy(list(stats["timeline"]))) record_tool_usage._last_save = now + except RuntimeError: + # No running event loop — fall back to sync write + try: + _atomic_write_json_sync(timeline_path, stats["timeline"]) + record_tool_usage._last_save = now + except Exception as e: + logger.warning(f"Failed to persist timeline: {e}") except Exception as e: logger.warning(f"Failed to persist timeline: {e}") @@ -1093,7 +1122,7 @@ async def chat_completions(request: Request): stats["reasoning_requests"] = stats.get("reasoning_requests", 0) + 1 elif target_model == "agent-advanced-core": stats["advanced_requests"] = stats.get("advanced_requests", 0) + 1 - save_persisted_stats() + await save_persisted_stats() # Update the parent Langfuse observation with classification results if parent_obs: From 1fd6d01fcf7a121c1469fd8e03532bf62bad588d Mon Sep 17 00:00:00 2001 From: sheepdestroyer Date: Thu, 18 Jun 2026 01:34:32 +0200 Subject: [PATCH 08/21] feat: dataset visualizer + benchmark script + gap-fill extraction --- pod.yaml | 5 + router/Containerfile | 1 + router/main.py | 48 ++++- router/static/visualizer.html | 349 ++++++++++++++++++++++++++++++++ scripts/benchmark_classifier.py | 127 ++++++++++++ scripts/extract_complex.py | 121 +++++++++++ scripts/extract_gapfill.py | 134 ++++++++++++ 7 files changed, 782 insertions(+), 3 deletions(-) create mode 100644 router/static/visualizer.html create mode 100644 scripts/benchmark_classifier.py create mode 100644 scripts/extract_complex.py create mode 100644 scripts/extract_gapfill.py diff --git a/pod.yaml b/pod.yaml index 9f47601b..caaf4468 100644 --- a/pod.yaml +++ b/pod.yaml @@ -127,6 +127,8 @@ spec: volumeMounts: - mountPath: /config/router_dir name: router-config + - mountPath: /app/data + name: dataset-data - mountPath: /config/gemini_auth name: gemini-secrets - mountPath: /root/.gemini @@ -466,3 +468,6 @@ spec: - hostPath: path: /home/gpav/Vrac/LAB/AI/LLM-Routing name: env-file + - hostPath: + path: /home/gpav/Vrac/LAB/AI/LLM-Routing/data + name: dataset-data diff --git a/router/Containerfile b/router/Containerfile index b21bf253..d685b94d 100644 --- a/router/Containerfile +++ b/router/Containerfile @@ -7,6 +7,7 @@ RUN pip install --no-cache-dir fastapi uvicorn httpx pyyaml python-multipart asy # Copy all source in one layer — removes dead config COPY (volume-mounted at runtime) COPY main.py agy_proxy.py circuit_breaker.py aa_scores.json free_models_roster.json best_free_model.json /app/ +COPY static/ /app/static/ EXPOSE 5000 diff --git a/router/main.py b/router/main.py index cce9cfca..3fb9e158 100644 --- a/router/main.py +++ b/router/main.py @@ -12,6 +12,8 @@ from contextlib import asynccontextmanager from fastapi import FastAPI, Request, HTTPException, Response from fastapi.responses import HTMLResponse, StreamingResponse, JSONResponse +from fastapi.staticfiles import StaticFiles +from pathlib import Path from circuit_breaker import get_breaker # Configure logging — respect LOG_LEVEL env var (default: WARNING) @@ -157,11 +159,24 @@ def _atomic_write_json_sync(path: str, data) -> None: """Synchronously write JSON data to path using atomic temp-file + os.replace.""" os.makedirs(os.path.dirname(path), exist_ok=True) fd, tmp_path = tempfile.mkstemp(dir=os.path.dirname(path), suffix=".tmp") + f = None try: - with os.fdopen(fd, "w") as f: - json.dump(data, f, indent=2) + f = os.fdopen(fd, "w") + json.dump(data, f, indent=2) + f.close() + f = None os.replace(tmp_path, path) except Exception: + if f is not None: + try: + f.close() + except Exception: + pass + else: + try: + os.close(fd) + except Exception: + pass try: os.unlink(tmp_path) except Exception: @@ -196,9 +211,9 @@ async def save_persisted_stats(force=False): if not force and (now - _last_stats_save < 2.0): return + _last_stats_save = now # Set immediately to prevent concurrent writes during await try: await _atomic_write_json_async(STATS_JSON_PATH, stats) - _last_stats_save = now # only advance on successful write except Exception as e: logger.error(f"Failed to persist stats to disk: {e}") @@ -2500,6 +2515,33 @@ def src_badge(label, color): """ return html_content +# --- Static files (visualizer, data files) --- +STATIC_DIR = Path(__file__).resolve().parent / "static" +DATA_DIR = Path(__file__).resolve().parent / "data" +DATA_DIR.mkdir(exist_ok=True) +app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") +app.mount("/data", StaticFiles(directory=str(DATA_DIR)), name="data") + +@app.get("/visualizer", response_class=HTMLResponse) +async def get_visualizer(): + """Serve the dataset visualizer for human review.""" + vis_path = STATIC_DIR / "visualizer.html" + if vis_path.exists(): + return HTMLResponse(vis_path.read_text()) + return HTMLResponse("

Visualizer not found

", status_code=404) + +@app.post("/dashboard/save-annotations") +async def save_annotations(request: Request): + """Save human review annotations to disk.""" + try: + body = await request.json() + ann_path = DATA_DIR / "annotations.json" + ann_path.write_text(json.dumps(body, indent=2, ensure_ascii=False)) + return JSONResponse({"status": "ok", "saved": len(body)}) + except Exception as e: + logger.error(f"Failed to save annotations: {e}") + raise HTTPException(status_code=500, detail=str(e)) + if __name__ == "__main__": import uvicorn logger.info(f"Starting LLM Triage Router on {host}:{port}...") diff --git a/router/static/visualizer.html b/router/static/visualizer.html new file mode 100644 index 00000000..789b4298 --- /dev/null +++ b/router/static/visualizer.html @@ -0,0 +1,349 @@ + + + + + +Classifier Dataset Visualizer + + + +
+

📊 Dataset Visualizer

+
+
Total: 0
+
Agree: 0
+
Conflict: 0
+
Reviewed: 0
+
+
+
+ + + + + Showing 0 + +
+
+
+
+
+

Select a prompt from the list

+

LLM evaluation vs classifier prediction side by side

+
+
+
+ + + + diff --git a/scripts/benchmark_classifier.py b/scripts/benchmark_classifier.py new file mode 100644 index 00000000..7770e8e7 --- /dev/null +++ b/scripts/benchmark_classifier.py @@ -0,0 +1,127 @@ +"""Benchmark gemma4-26a4b-routing classifier against labeled dataset.""" +import json, urllib.request, time, sys +from collections import defaultdict, Counter +from pathlib import Path + +# Load dataset +dataset_path = Path(__file__).resolve().parent.parent / "data" / "classified_dataset.json" +with open(dataset_path) as f: + dataset = json.load(f) + +# Classifier prompt (same as router/config.yaml) +PROMPT_TEMPLATE = """Classify the coding task complexity. Output ONLY the tier name. + +agent-simple-core: trivial one-liners, syntax fixes, single-line edits +agent-medium-core: single-function changes, light refactoring, simple tests +agent-complex-core: multi-file changes, algorithmic work, data pipelines +agent-reasoning-core: deep analysis, architecture decisions, debugging complex systems +agent-advanced-core: system-level architecture, cross-cutting concerns, novel design + +Task: """ + +TIERS = [ + "agent-simple-core", "agent-medium-core", "agent-complex-core", + "agent-reasoning-core", "agent-advanced-core" +] + +def classify(prompt): + """Call gemma4-26a4b-routing via llama-server.""" + payload = { + "model": "gemma4-26a4b-routing", + "messages": [{"role": "user", "content": PROMPT_TEMPLATE + prompt}], + "temperature": 0.0, + "max_tokens": 15, + } + req = urllib.request.Request( + "http://127.0.0.1:8080/v1/chat/completions", + data=json.dumps(payload).encode(), + headers={"Content-Type": "application/json", "Authorization": "Bearer local-token"} + ) + with urllib.request.urlopen(req, timeout=30) as resp: + data = json.loads(resp.read()) + return data["choices"][0]["message"].get("content", "").strip() + +print(f"Benchmark: gemma4-26a4b-routing vs {dataset['total']} labeled prompts\n") + +# Run classification +results = [] +correct = 0 +per_tier = {t: {"correct": 0, "total": 0} for t in TIERS} +confusion = defaultdict(Counter) # confusion[expected][predicted] + +for i, item in enumerate(dataset["prompts"]): + prompt = item["prompt"] + expected = item["tier"] + + try: + predicted = classify(prompt) + except Exception as e: + predicted = f"ERROR: {str(e)[:50]}" + + results.append({ + "prompt": prompt[:100], + "expected": expected, + "predicted": predicted, + }) + + per_tier[expected]["total"] += 1 + if predicted == expected: + correct += 1 + per_tier[expected]["correct"] += 1 + + confusion[expected][predicted] += 1 + + # Progress + if (i + 1) % 20 == 0: + acc = correct / (i + 1) * 100 + print(f" {i+1}/{dataset['total']} — accuracy {acc:.1f}%") + + time.sleep(0.05) # minimal rate-limit (model handles concurrency via llama-server slots) + +# Report +total = dataset["total"] +overall = correct / total * 100 + +print(f"\n{'='*60}") +print(f"Overall accuracy: {correct}/{total} ({overall:.1f}%)") +print(f"{'='*60}") + +print(f"\nPer-tier accuracy:") +for tier in TIERS: + t = per_tier[tier] + pct = t["correct"] / t["total"] * 100 if t["total"] > 0 else 0 + bar = "█" * int(pct / 5) + "░" * (20 - int(pct / 5)) + print(f" {tier:30s} {t['correct']:3d}/{t['total']:3d} {bar} {pct:.1f}%") + +print(f"\nConfusion matrix (expected → predicted):") +header = " " * 30 + "".join(f"{t:25s}" for t in TIERS) +print(header) +for exp_tier in TIERS: + row = f"{exp_tier:30s}" + for pred_tier in TIERS: + count = confusion[exp_tier].get(pred_tier, 0) + if exp_tier == pred_tier: + row += f" \033[32m{count:3d}\033[0m" + " " * 20 + elif count > 0: + row += f" \033[31m{count:3d}\033[0m" + " " * 20 + else: + row += f" {count:3d}" + " " * 20 + print(row) + +# Save detailed results +out_path = Path(__file__).resolve().parent.parent / "data" / "benchmark_results.json" +with open(out_path, 'w') as f: + json.dump({ + "classifier": "gemma4-26a4b-routing", + "dataset_total": total, + "overall_accuracy": round(overall, 1), + "per_tier": {t: { + "correct": per_tier[t]["correct"], + "total": per_tier[t]["total"], + "accuracy": round(per_tier[t]["correct"] / per_tier[t]["total"] * 100, 1) if per_tier[t]["total"] > 0 else 0 + } for t in TIERS}, + "confusion": {t: dict(confusion[t]) for t in TIERS}, + "details": results, + }, f, indent=2, ensure_ascii=False) + +print(f"\nDetailed results saved to {out_path}") diff --git a/scripts/extract_complex.py b/scripts/extract_complex.py new file mode 100644 index 00000000..2a98c4bc --- /dev/null +++ b/scripts/extract_complex.py @@ -0,0 +1,121 @@ +"""Final gap-fill: deep extraction targeting complex + advanced tiers only.""" +import os, base64, json, urllib.request, time +from pathlib import Path + +env = {} +env_path = Path(__file__).resolve().parent.parent / ".env" +with open(env_path) as f: + for line in f: + line = line.strip() + if line and not line.startswith('#') and '=' in line: + k, v = line.split('=', 1) + env[k.strip()] = v.strip().strip('"').strip("'") + +pk = env['LANGFUSE_PUBLIC_KEY'] +sk = env['LANGFUSE_SECRET_KEY'] +auth = base64.b64encode(f"{pk}:{sk}".encode()).decode() +base_url = "http://localhost:3001" + +# Load already-classified +existing = set() +dataset_path = Path(__file__).resolve().parent.parent / "data" / "classified_dataset.json" +with open(dataset_path) as f: + existing_data = json.load(f) +for p in existing_data.get('prompts', []): + existing.add(p['prompt'].strip().lower()) + +print(f"Already classified: {len(existing)} prompts") + +def fetch_observations(page=1, limit=50): + url = f"{base_url}/api/public/observations?limit={limit}&page={page}&orderBy=timestamp.desc&level=DEFAULT&name=litellm-acompletion" + req = urllib.request.Request(url) + req.add_header("Authorization", f"Basic {auth}") + with urllib.request.urlopen(req, timeout=15) as resp: + return json.loads(resp.read()) + +def extract_user_prompt(obs): + inp = obs.get('input') + if not inp: return None + if isinstance(inp, str): + try: inp = json.loads(inp) + except: return None + if not isinstance(inp, dict): return None + messages = inp.get('messages', []) + if not messages: return None + for msg in reversed(messages): + if isinstance(msg, dict) and msg.get('role') == 'user': + content = msg.get('content', '') + if isinstance(content, str) and len(content.strip()) > 3: + return content.strip() + return None + +# Keywords suggesting complex/advanced work +COMPLEX_KEYWORDS = [ + 'refactor', 'architect', 'design', 'implement', 'migrate', 'debug', + 'diagnose', 'review', 'analyze', 'optimize', 'restructure', 'pipeline', + 'system', 'module', 'framework', 'pattern', 'strategy', 'algorithm', + 'multi-tenant', 'distributed', 'sharding', 'consensus', 'isolation', + 'security', 'vulnerability', 'deadlock', 'concurrent', 'scale', + 'infrastructure', 'deploy', 'orchestrate', 'integrate', 'protocol', +] + +def looks_complex(prompt): + """Heuristic: longer prompts with complex keywords.""" + lower = prompt.lower() + if len(prompt) < 200: + return False + score = sum(1 for kw in COMPLEX_KEYWORDS if kw in lower) + return score >= 2 + +print("Deep extraction: targeting complex/advanced prompts...") +prompts = [] +seen = set() +page = 1 +target = 50 +max_pages = 200 # go deep into history + +while len(prompts) < target and page <= max_pages: + try: + data = fetch_observations(page=page, limit=50) + except Exception as e: + print(f" Page {page} failed: {e}") + break + + obs_list = data.get('data', []) + if not obs_list: + print(f" Page {page}: empty, stopping") + break + + added = 0 + for obs in obs_list: + if len(prompts) >= target: + break + prompt = extract_user_prompt(obs) + if not prompt: continue + norm = prompt.strip().lower() + if norm in seen: continue + if norm in existing: continue + if not looks_complex(prompt): continue + + seen.add(norm) + prompts.append({ + "prompt": prompt, + "observation_id": obs.get('id', ''), + "trace_id": obs.get('traceId', ''), + "timestamp": obs.get('startTime', ''), + }) + added += 1 + + print(f" Page {page}: +{added} new → {len(prompts)} total") + page += 1 + time.sleep(0.1) + +out_path = Path(__file__).resolve().parent.parent / "data" / "raw_prompts_complex.json" +with open(out_path, 'w') as f: + json.dump(prompts, f, indent=2, ensure_ascii=False) + +print(f"\nSaved {len(prompts)} complex prompts to {out_path}") +lengths = [len(p['prompt']) for p in prompts] +print(f"Length range: {min(lengths)}-{max(lengths)} chars, avg: {sum(lengths)/len(lengths):.0f}") +for p in prompts[:5]: + print(f" [{p['timestamp'][:19]}] ({len(p['prompt'])} chars) {p['prompt'][:120]}...") diff --git a/scripts/extract_gapfill.py b/scripts/extract_gapfill.py new file mode 100644 index 00000000..0eb61615 --- /dev/null +++ b/scripts/extract_gapfill.py @@ -0,0 +1,134 @@ +"""Gap-fill extraction: pull longer/older prompts targeting complex+ tiers.""" +import os, base64, json, urllib.request, time +from pathlib import Path + +env = {} +env_path = Path(__file__).resolve().parent.parent / ".env" +with open(env_path) as f: + for line in f: + line = line.strip() + if line and not line.startswith('#') and '=' in line: + k, v = line.split('=', 1) + env[k.strip()] = v.strip().strip('"').strip("'") + +pk = env['LANGFUSE_PUBLIC_KEY'] +sk = env['LANGFUSE_SECRET_KEY'] +auth = base64.b64encode(f"{pk}:{sk}".encode()).decode() +base_url = "http://localhost:3001" + +# Load already-classified prompts to skip +existing = set() +dataset_path = Path(__file__).resolve().parent.parent / "data" / "classified_dataset.json" +if dataset_path.exists(): + with open(dataset_path) as f: + existing_data = json.load(f) + for p in existing_data.get('prompts', []): + existing.add(p['prompt'].strip().lower()) + +print(f"Already classified: {len(existing)} prompts") + +def fetch_observations(page=1, limit=50): + """Fetch DEFAULT-level litellm-acompletion observations.""" + url = f"{base_url}/api/public/observations?limit={limit}&page={page}&orderBy=timestamp.desc&level=DEFAULT&name=litellm-acompletion" + req = urllib.request.Request(url) + req.add_header("Authorization", f"Basic {auth}") + with urllib.request.urlopen(req, timeout=15) as resp: + return json.loads(resp.read()) + +def extract_user_prompt(obs): + inp = obs.get('input') + if not inp: + return None + if isinstance(inp, str): + try: inp = json.loads(inp) + except: return None + if not isinstance(inp, dict): + return None + messages = inp.get('messages', []) + if not messages: + return None + for msg in reversed(messages): + if isinstance(msg, dict) and msg.get('role') == 'user': + content = msg.get('content', '') + if isinstance(content, str) and len(content.strip()) > 3: + return content.strip() + return None + +def is_trivial(prompt): + lower = prompt.strip().lower() + if len(lower) < 20: + return True + trivial = ["say hello", "hi", "test", "ping", "hello", "hey", "what's up", + "how are you", "good morning", "what model are you", "who are you", + "tell me a joke", "what is 2+2", "what is the capital"] + for pat in trivial: + if pat in lower and len(lower) < 50: + return True + return False + +print("Extracting gap-fill prompts (longer, older, targeting complex+)...") +prompts = [] +seen = set() +page = 1 +target = 80 # generous — workers will filter further +max_pages = 100 + +while len(prompts) < target and page <= max_pages: + try: + data = fetch_observations(page=page, limit=50) + except Exception as e: + print(f" Page {page} failed: {e}") + break + + obs_list = data.get('data', []) + if not obs_list: + print(f" Page {page}: empty, stopping") + break + + added = 0 + for obs in obs_list: + if len(prompts) >= target: + break + + prompt = extract_user_prompt(obs) + if not prompt: + continue + if is_trivial(prompt): + continue + + norm = prompt.strip().lower() + if norm in seen: + continue + if norm in existing: + continue + + # Bias toward complex: prefer longer prompts (>200 chars) + # but don't exclude shorter ones entirely + if len(prompt) < 100 and len(prompts) > 40: + continue # after 40 collected, only take substantial prompts + + seen.add(norm) + prompts.append({ + "prompt": prompt, + "observation_id": obs.get('id', ''), + "trace_id": obs.get('traceId', ''), + "timestamp": obs.get('startTime', ''), + "model": obs.get('model', ''), + }) + added += 1 + + print(f" Page {page}: +{added} new → {len(prompts)} total") + page += 1 + time.sleep(0.1) + +out_dir = Path(__file__).resolve().parent.parent / "data" +out_path = out_dir / "raw_prompts_gapfill.json" +with open(out_path, 'w') as f: + json.dump(prompts, f, indent=2, ensure_ascii=False) + +print(f"\nSaved {len(prompts)} gap-fill prompts to {out_path}") +lengths = [len(p['prompt']) for p in prompts] +print(f"Length range: {min(lengths)}-{max(lengths)} chars, avg: {sum(lengths)/len(lengths):.0f}") +print(f"Sample:") +for p in prompts[:5]: + print(f" [{p['timestamp'][:19]}] ({len(p['prompt'])} chars) {p['prompt'][:100]}...") From 7fc31cb2ba37c05304f049b2267721d2644922c7 Mon Sep 17 00:00:00 2001 From: sheepdestroyer Date: Thu, 18 Jun 2026 02:01:46 +0200 Subject: [PATCH 09/21] fix: restore tier table widget + extract first user msg from Hermes sessions --- scripts/extract_prompts.py | 81 ++++++++++++++++++++++---------------- 1 file changed, 47 insertions(+), 34 deletions(-) diff --git a/scripts/extract_prompts.py b/scripts/extract_prompts.py index 8c3e454d..d5d19b50 100644 --- a/scripts/extract_prompts.py +++ b/scripts/extract_prompts.py @@ -1,5 +1,12 @@ -"""Extract 300 meaningful coding prompts from Langfuse observations.""" -import os, base64, json, urllib.request, urllib.error, sys, time +"""Re-extract prompts using FIRST user message (not last) — fixes truncation.""" +import os +import base64 +import json +import urllib.request +import urllib.error +import sys +import time +import re from pathlib import Path env = {} @@ -25,13 +32,15 @@ ] def is_trivial(prompt): - """Filter out test pings and trivial prompts.""" + """Filter out test pings and trivial prompts using word boundaries to avoid partial matches.""" lower = prompt.strip().lower() if len(lower) < 20: return True for pat in TRIVIAL_PATTERNS: - if pat in lower and len(lower) < 50: - return True + if len(lower) < 50: + escaped = re.escape(pat) + if re.search(r'\b' + escaped + r'\b', lower): + return True return False def fetch_observations(page=1, limit=50): @@ -42,8 +51,8 @@ def fetch_observations(page=1, limit=50): with urllib.request.urlopen(req, timeout=15) as resp: return json.loads(resp.read()) -def extract_user_prompt(obs): - """Extract the last user message from an observation's input.""" +def extract_first_user_prompt(obs): + """Extract the FIRST real user message (skip system notes).""" inp = obs.get('input') if not inp: return None @@ -57,19 +66,31 @@ def extract_user_prompt(obs): messages = inp.get('messages', []) if not messages: return None - for msg in reversed(messages): - if isinstance(msg, dict) and msg.get('role') == 'user': - content = msg.get('content', '') - if isinstance(content, str) and len(content.strip()) > 3: - return content.strip() + + for msg in messages: + if not isinstance(msg, dict): + continue + if msg.get('role') != 'user': + continue + content = msg.get('content', '') + if not isinstance(content, str) or len(content.strip()) <= 3: + continue + # Skip Hermes system notes injected as user messages + stripped = content.strip() + if stripped.startswith('[System:') or stripped.startswith('[Note:'): + continue + if stripped.startswith('[IMPORTANT:'): + # Skill invocations — keep these, they're real prompts + pass + return stripped return None -print("Extracting meaningful coding prompts from Langfuse observations...") +print("Re-extracting prompts using FIRST user message...") prompts = [] seen = set() page = 1 target = 300 -max_pages = 50 # 50 pages × 50 = 2500 observations +max_pages = 100 while len(prompts) < target and page <= max_pages: try: @@ -79,26 +100,21 @@ def extract_user_prompt(obs): break obs_list = data.get('data', []) - total_available = data.get('meta', {}).get('totalItems', 0) - if not obs_list: print(f" Page {page}: empty, stopping") break - added_this_page = 0 + added = 0 for obs in obs_list: if len(prompts) >= target: break - prompt = extract_user_prompt(obs) + prompt = extract_first_user_prompt(obs) if not prompt: continue - - # Skip trivial test pings if is_trivial(prompt): continue - # Deduplicate norm = prompt.strip().lower() if norm in seen: continue @@ -111,26 +127,23 @@ def extract_user_prompt(obs): "timestamp": obs.get('startTime', ''), "model": obs.get('model', ''), }) - added_this_page += 1 + added += 1 - print(f" Page {page}: {len(obs_list)} obs, +{added_this_page} new → {len(prompts)} total (of {total_available} available)") + print(f" Page {page}: +{added} new → {len(prompts)} total") page += 1 - time.sleep(0.1) # gentle rate limit + time.sleep(0.1) -# Save out_dir = Path(__file__).resolve().parent.parent / "data" -out_dir.mkdir(exist_ok=True) -out_path = out_dir / "raw_prompts.json" - +out_path = out_dir / "raw_prompts_v2.json" with open(out_path, 'w') as f: json.dump(prompts, f, indent=2, ensure_ascii=False) print(f"\nSaved {len(prompts)} prompts to {out_path}") -# Stats lengths = [len(p['prompt']) for p in prompts] -print(f"Length range: {min(lengths)}-{max(lengths)} chars") -print(f"Avg length: {sum(lengths)/len(lengths):.0f} chars") -print(f"\nSample (first 10):") -for p in prompts[:10]: - print(f" [{p['timestamp'][:19]}] {p['prompt'][:120]}...") +if lengths: + print(f"Length: min={min(lengths)}, max={max(lengths)}, median={sorted(lengths)[len(lengths)//2]}, avg={sum(lengths)/len(lengths):.0f}") + print(f"Short (<100 chars): {sum(1 for l in lengths if l < 100)}") + print(f"\nSample (first 10):") + for p in prompts[:10]: + print(f" [{p['timestamp'][:19]}] ({len(p['prompt'])}c) {p['prompt'][:120]}...") From a4fcb01fb9c81d3b1c7fbf645d23f537f692b228 Mon Sep 17 00:00:00 2001 From: sheepdestroyer Date: Thu, 18 Jun 2026 05:13:50 +0200 Subject: [PATCH 10/21] feat: gemma4 classifier with grammar, combined evals visualizer, dashboard link --- router/main.py | 3 + router/static/visualizer.html | 39 ++++++++----- scripts/reclassify_all.py | 101 ++++++++++++++++++++++++++++++++++ 3 files changed, 130 insertions(+), 13 deletions(-) create mode 100644 scripts/reclassify_all.py diff --git a/router/main.py b/router/main.py index 3fb9e158..ca8398d9 100644 --- a/router/main.py +++ b/router/main.py @@ -2260,6 +2260,9 @@ def src_badge(label, color):
Antigravity Gateway
System Control Center
+ {oauth_banner_html} diff --git a/router/static/visualizer.html b/router/static/visualizer.html index 789b4298..7003c340 100644 --- a/router/static/visualizer.html +++ b/router/static/visualizer.html @@ -71,6 +71,7 @@

📊 Dataset Visualizer

Agree: 0
Conflict: 0
Reviewed: 0
+
@@ -112,9 +113,15 @@

Select a prompt from the list

async function init() { try { const resp = await fetch('data/classified_dataset.json'); - dataset = (await resp.json()).prompts; + const raw = await resp.json(); + dataset = raw.prompts || []; + // Extract global stats from combined dataset + if (raw.agreement !== undefined) { + document.getElementById('stat-agreement').textContent = raw.agreement + '%'; + document.getElementById('stat-agreement').style.display = ''; + } } catch(e) { - console.warn('Dataset not found, trying local path...'); + console.warn('Dataset not found'); dataset = []; } @@ -139,8 +146,9 @@

Select a prompt from the list

} function getBenchmarkPrediction(idx) { - if (benchmark.details && benchmark.details[idx]) { - return benchmark.details[idx].predicted; + // Read clf_tier directly from combined dataset (reclassify_all.py output) + if (dataset[idx] && dataset[idx].clf_tier) { + return dataset[idx].clf_tier; } return null; } @@ -150,9 +158,12 @@

Select a prompt from the list

} function isConflict(idx) { - const pred = getBenchmarkPrediction(idx); - if (!pred || pred.startsWith('ERROR')) return null; // unknown - return pred !== dataset[idx].tier; + const p = dataset[idx]; + if (!p) return null; + const llm = p.llm_tier || p.tier; // support both old and new format + const pred = p.clf_tier || getBenchmarkPrediction(idx); + if (!pred || pred.startsWith('ERROR')) return null; + return pred !== llm; } function setFilter(filter) { @@ -199,7 +210,8 @@

Select a prompt from the list

const listEl = document.getElementById('list'); listEl.innerHTML = filtered.map(i => { const p = dataset[i]; - const pred = getBenchmarkPrediction(i); + const llm = p.llm_tier || p.tier; + const pred = p.clf_tier; const ann = getAnnotation(i); const conflict = isConflict(i); @@ -210,7 +222,7 @@

Select a prompt from the list

if (ann) tags += 'REVIEWED'; return `
-
#${i} · LLM: ${p.tier.replace('agent-','')}${pred ? ' · CLS: ' + pred.replace('agent-','') : ''}
+
#${i} · LLM: ${llm.replace('agent-','')}${pred ? ' · CLS: ' + pred.replace('agent-','') : ''}
${escapeHtml(p.prompt.substring(0, 120))}
${tags}
`; @@ -228,9 +240,10 @@

Select a prompt from the list

function renderDetail(idx) { const p = dataset[idx]; - const pred = getBenchmarkPrediction(idx); + const llm = p.llm_tier || p.tier; + const pred = p.clf_tier; const ann = getAnnotation(idx); - const conflict = isConflict(idx); + const conflict = pred && !pred.startsWith('ERROR') ? (pred !== llm) : null; let predCardClass = 'eval-card'; let llmCardClass = 'eval-card'; @@ -247,7 +260,7 @@

Select a prompt from the list

let tierBtns = TIERS.map(tier => { let cls = 'tier-btn'; if (ann && ann.tier === tier) cls += ' selected-human'; - if (p.tier === tier && !ann) cls += ' llm-label'; + if (llm === tier && !ann) cls += ' llm-label'; if (pred === tier) cls += ' classifier-label'; return ``; }).join(''); @@ -261,7 +274,7 @@

Prompt #${idx} · ${p.prompt.length} chars

🤖 LLM Evaluation (kanban worker)

-
${p.tier}
+
${llm}
original dataset label
diff --git a/scripts/reclassify_all.py b/scripts/reclassify_all.py new file mode 100644 index 00000000..5170d552 --- /dev/null +++ b/scripts/reclassify_all.py @@ -0,0 +1,101 @@ +"""Re-run gemma4 classifier (with grammar) on all dataset prompts via router.""" +import json, urllib.request, time, sys +from pathlib import Path +from collections import Counter + +TIERS = ['agent-simple-core','agent-medium-core','agent-complex-core','agent-reasoning-core','agent-advanced-core'] + +ROUTER_URL = "http://127.0.0.1:8080/v1/chat/completions" +print(f"Using router on {ROUTER_URL}") + +PROMPT_TEMPLATE = """Analyze the request complexity. Respond with exactly one of: +- simple boilerplate: agent-simple-core +- moderate complexity: agent-medium-core +- deep algorithms: agent-complex-core +- heavy multi-step reasoning: agent-reasoning-core +- system-level / novel design: agent-advanced-core + +Request: """ + +def classify(prompt): + if len(prompt) > 600: + prompt = prompt[:600] + payload = { + 'model': 'gemma4-26a4b-routing', + 'messages': [{'role': 'user', 'content': PROMPT_TEMPLATE + prompt}], + 'max_tokens': 15, 'temperature': 0, + 'grammar': 'root ::= "agent-simple-core" | "agent-medium-core" | "agent-complex-core" | "agent-reasoning-core" | "agent-advanced-core"' + } + req = urllib.request.Request(ROUTER_URL, data=json.dumps(payload).encode(), headers={'Content-Type':'application/json','Authorization':'Bearer local-token'}) + with urllib.request.urlopen(req, timeout=30) as resp: + data = json.loads(resp.read()) + return data['choices'][0]['message'].get('content','').strip() + +# Load existing dataset (kanban/llm evals) +data_dir = Path(__file__).resolve().parent.parent / 'data' +with open(data_dir / 'classified_dataset.json') as f: + dataset = json.load(f) + +# Load raw prompts for full text +with open(data_dir / 'raw_prompts_hermes.json') as f: + all_prompts = json.load(f) + +# Build prompt lookup +prompt_map = {} +for p in all_prompts: + prompt_map[p['prompt']] = p + +print(f"Classifying {len(dataset['prompts'])} prompts with gemma4-26a4b (grammar-enforced)...") + +results = [] +for i, item in enumerate(dataset['prompts']): + prompt = item['prompt'] + + # Original LLM/kanban eval + llm_tier = item.get('tier', '?') + + # Classifier eval + try: + clf_tier = classify(prompt) + except Exception as e: + clf_tier = f"ERROR: {str(e)[:50]}" + + results.append({ + 'prompt': prompt, + 'llm_tier': llm_tier, + 'clf_tier': clf_tier, + 'session_id': item.get('session_id', ''), + }) + + if (i + 1) % 30 == 0: + agree = sum(1 for r in results if r['llm_tier'] == r['clf_tier']) + print(f" {i+1}/{len(dataset['prompts'])} — {agree}/{i+1} agree ({agree/(i+1)*100:.0f}%)") + sys.stdout.flush() + +# Stats +clf_counts = Counter(r['clf_tier'] for r in results) +llm_counts = Counter(r['llm_tier'] for r in results) +agree = sum(1 for r in results if r['llm_tier'] == r['clf_tier']) + +print(f"\n{'='*60}") +print(f"Agreement: {agree}/{len(results)} ({agree/len(results)*100:.1f}%)") +print(f"\nTier distribution:") +print(f"{'Tier':30s} {'LLM':>6s} {'CLF':>6s} {'Δ':>6s}") +for t in TIERS: + lc = llm_counts.get(t, 0) + cc = clf_counts.get(t, 0) + print(f" {t:30s} {lc:>6d} {cc:>6d} {cc-lc:>+6d}") + +# Save combined dataset +combined = { + 'total': len(results), + 'agreement': round(agree / len(results) * 100, 1), + 'llm_counts': dict(llm_counts), + 'clf_counts': dict(clf_counts), + 'prompts': results, +} + +with open(data_dir / 'classified_dataset.json', 'w') as f: + json.dump(combined, f, indent=2, ensure_ascii=False) + +print(f"\nSaved to classified_dataset.json (now with llm_tier + clf_tier)") From 640d2b682f26e03e867b216685e94df2b25eca24 Mon Sep 17 00:00:00 2001 From: sheepdestroyer Date: Thu, 18 Jun 2026 09:26:49 +0200 Subject: [PATCH 11/21] feat: direct classifier + error retry scripts for dataset building --- scripts/classify_direct.py | 95 ++++++++++++++++++++++++++++++++++++ scripts/retry_errors.py | 99 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 194 insertions(+) create mode 100644 scripts/classify_direct.py create mode 100644 scripts/retry_errors.py diff --git a/scripts/classify_direct.py b/scripts/classify_direct.py new file mode 100644 index 00000000..be6fa954 --- /dev/null +++ b/scripts/classify_direct.py @@ -0,0 +1,95 @@ +"""Direct classification of Hermes prompts using gemma4-26a4b-routing.""" +import json, urllib.request, time +from pathlib import Path + +PROMPT_TEMPLATE = """Classify the coding task complexity. Output ONLY the tier name. + +agent-simple-core: trivial one-liners, syntax fixes, single-line edits +agent-medium-core: single-function changes, light refactoring, simple tests +agent-complex-core: multi-file changes, algorithmic work, data pipelines +agent-reasoning-core: deep analysis, architecture decisions, debugging complex systems +agent-advanced-core: system-level architecture, cross-cutting concerns, novel design + +Task: """ + +def classify(prompt): + payload = { + "model": "gemma4-26a4b-routing", + "messages": [{"role": "user", "content": PROMPT_TEMPLATE + prompt}], + "temperature": 0.0, + "max_tokens": 15, + } + req = urllib.request.Request( + "http://127.0.0.1:8080/v1/chat/completions", + data=json.dumps(payload).encode(), + headers={"Content-Type": "application/json", "Authorization": "Bearer local-token"} + ) + with urllib.request.urlopen(req, timeout=30) as resp: + data = json.loads(resp.read()) + return data["choices"][0]["message"].get("content", "").strip() + +# Load prompts +data_dir = Path(__file__).resolve().parent.parent / "data" +with open(data_dir / "raw_prompts_hermes.json") as f: + prompts = json.load(f) + +print(f"Classifying {len(prompts)} prompts with gemma4-26a4b-routing...") + +results = [] +errors = 0 +for i, p in enumerate(prompts): + prompt = p['prompt'] + # Truncate very long prompts to classifier context window (~3500 chars safe margin) + if len(prompt) > 3500: + prompt = prompt[:3500] + + try: + tier = classify(prompt) + results.append({"id": i, "tier": tier, "prompt_snippet": prompt[:60]}) + except Exception as e: + tier = f"ERROR" + errors += 1 + results.append({"id": i, "tier": tier, "error": str(e)[:100]}) + print(f" [{i}] ERROR: {str(e)[:80]}") + + if (i + 1) % 30 == 0: + print(f" {i+1}/{len(prompts)} — {errors} errors") + time.sleep(0.05) + +# Count tiers +from collections import Counter +counts = Counter(r["tier"] for r in results) + +print(f"\nDone. {len(results)} classified, {errors} errors") +print(f"Counts:") +for tier in ['agent-simple-core', 'agent-medium-core', 'agent-complex-core', 'agent-reasoning-core', 'agent-advanced-core']: + c = counts.get(tier, 0) + print(f" {tier}: {c}") +error_count = sum(1 for r in results if r["tier"] == "ERROR") +if error_count: + print(f" ERROR: {error_count}") + +# Build dataset +dataset_prompts = [] +for p, r in zip(prompts, results): + dataset_prompts.append({ + "prompt": p['prompt'], + "tier": r['tier'], + "classifier": "uuid", + "session_id": p.get('session_id', ''), + }) + +dataset = { + "prompts": dataset_prompts, + "counts": dict(counts), + "total": len(dataset_prompts), + "gaps": [t for t in ['agent-simple-core','agent-medium-core','agent-complex-core','agent-reasoning-core','agent-advanced-core'] + if counts.get(t, 0) < 20] +} + +out_path = data_dir / "classified_dataset.json" +with open(out_path, 'w') as f: + json.dump(dataset, f, indent=2, ensure_ascii=False) + +print(f"\nSaved to {out_path}") +print(f"Gaps: {dataset['gaps']}") \ No newline at end of file diff --git a/scripts/retry_errors.py b/scripts/retry_errors.py new file mode 100644 index 00000000..78b655aa --- /dev/null +++ b/scripts/retry_errors.py @@ -0,0 +1,99 @@ +"""Retry the 94 failed prompts with 800-char truncation (safe for 4096-ctx model).""" +import json, urllib.request, time, subprocess +from pathlib import Path +from collections import Counter + +PROMPT_TEMPLATE = """Classify the coding task complexity. Output ONLY the tier name. + +agent-simple-core: trivial one-liners, syntax fixes, single-line edits +agent-medium-core: single-function changes, light refactoring, simple tests +agent-complex-core: multi-file changes, algorithmic work, data pipelines +agent-reasoning-core: deep analysis, architecture decisions, debugging complex systems +agent-advanced-core: system-level architecture, cross-cutting concerns, novel design + +Task: """ + +MAX_CHARS = 600 # proven safe across all prompt types in this dataset + +def get_model_port(): + """Discover the gemma4-26a4b-routing model's direct port (bypass router prompt-cache bug).""" + req = urllib.request.Request('http://127.0.0.1:8080/v1/models') + with urllib.request.urlopen(req, timeout=5) as resp: + data = json.loads(resp.read()) + for m in data.get('data', []): + if 'gemma4-26a4b' in m.get('id', ''): + args = m['status']['args'] + for i, v in enumerate(args): + if v == '--port': + return args[i + 1] + raise RuntimeError("gemma4-26a4b-routing model port not found") + +MODEL_PORT = get_model_port() +MODEL_URL = f"http://127.0.0.1:{MODEL_PORT}/v1/chat/completions" +print(f"Using model directly on port {MODEL_PORT}") + +def classify(prompt): + if len(prompt) > MAX_CHARS: + prompt = prompt[:MAX_CHARS] + payload = { + "model": "gemma4-26a4b-routing", + "messages": [{"role": "user", "content": PROMPT_TEMPLATE + prompt}], + "temperature": 0.0, + "max_tokens": 15, + } + req = urllib.request.Request( + MODEL_URL, + data=json.dumps(payload).encode(), + headers={"Content-Type": "application/json"} + ) + with urllib.request.urlopen(req, timeout=30) as resp: + data = json.loads(resp.read()) + content = data["choices"][0]["message"].get("content", "").strip() + # Normalize: strip "tier:" prefix, extract just the tier name + for tier in TIERS: + if tier in content: + return tier + return content + +TIERS = ['agent-simple-core','agent-medium-core','agent-complex-core','agent-reasoning-core','agent-advanced-core'] + +data_dir = Path(__file__).resolve().parent.parent / "data" +with open(data_dir / "classified_dataset.json") as f: + dataset = json.load(f) +with open(data_dir / "raw_prompts_hermes.json") as f: + all_prompts = json.load(f) + +error_indices = [i for i, p in enumerate(dataset['prompts']) if p['tier'] == 'ERROR'] +print(f"Retrying {len(error_indices)} failed prompts (max {MAX_CHARS} chars)...") + +fixed = 0 +errors = 0 + +for batch_start in range(0, len(error_indices), 5): + batch = error_indices[batch_start:batch_start + 5] + for idx in batch: + prompt = all_prompts[idx]['prompt'] + try: + tier = classify(prompt) + dataset['prompts'][idx]['tier'] = tier + fixed += 1 + except Exception as e: + errors += 1 + print(f" [{idx}] still failing: {str(e)[:80]}") + time.sleep(3) # single-slot server needs headroom + if batch_start + 5 < len(error_indices): + print(f" batch {batch_start//5 + 1}/{(len(error_indices)+4)//5}: {fixed} fixed, {errors} errors") + time.sleep(5) + +from collections import Counter +new_counts = Counter(p['tier'] for p in dataset['prompts']) +dataset['counts'] = {k: v for k, v in new_counts.items()} +dataset['gaps'] = [t for t in ['agent-simple-core','agent-medium-core','agent-complex-core','agent-reasoning-core','agent-advanced-core'] + if new_counts.get(t, 0) < 20] + +with open(data_dir / "classified_dataset.json", 'w') as f: + json.dump(dataset, f, indent=2, ensure_ascii=False) + +print(f"\nDone. Fixed: {fixed}, Errors: {errors}") +for tier in sorted(new_counts.keys()): + print(f" {tier:30s} {new_counts[tier]:3d}") \ No newline at end of file From 958e6b88c625692f902ee8f4c9c6e76c881122aa Mon Sep 17 00:00:00 2001 From: sheepdestroyer Date: Thu, 18 Jun 2026 10:35:36 +0200 Subject: [PATCH 12/21] fix: address PR reviews and align test suite for DualCircuitBreaker --- pod.yaml | 1 + router/main.py | 14 ++- router/static/visualizer.html | 18 ++-- scripts/benchmark_classifier.py | 10 +- scripts/classify_direct.py | 18 +++- scripts/extract_complex.py | 22 +++-- scripts/extract_gapfill.py | 19 ++-- scripts/reclassify_all.py | 34 ++++--- scripts/retry_errors.py | 4 +- test_a2_verify.py | 7 +- test_circuit_breaker.py | 167 ++++++++++++++++++++------------ verify_breaker.py | 23 +++-- 12 files changed, 213 insertions(+), 124 deletions(-) diff --git a/pod.yaml b/pod.yaml index caaf4468..bad4946c 100644 --- a/pod.yaml +++ b/pod.yaml @@ -470,4 +470,5 @@ spec: name: env-file - hostPath: path: /home/gpav/Vrac/LAB/AI/LLM-Routing/data + type: DirectoryOrCreate name: dataset-data diff --git a/router/main.py b/router/main.py index ca8398d9..05d906fa 100644 --- a/router/main.py +++ b/router/main.py @@ -215,6 +215,7 @@ async def save_persisted_stats(force=False): try: await _atomic_write_json_async(STATS_JSON_PATH, stats) except Exception as e: + _last_stats_save = 0.0 # Reset on failure to allow immediate retry logger.error(f"Failed to persist stats to disk: {e}") # Load initial stats from persistent storage @@ -391,10 +392,17 @@ async def lifespan(app: FastAPI): logger.error(f"Roster sync failed: {e}") # Start background task before yield so it runs during app lifetime - asyncio.create_task(push_aggregate_scores()) + task = asyncio.create_task(push_aggregate_scores()) yield + # Cancel background score task + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + # Flush any buffered stats/timeline on clean shutdown await save_persisted_stats(force=True) try: @@ -2539,11 +2547,11 @@ async def save_annotations(request: Request): try: body = await request.json() ann_path = DATA_DIR / "annotations.json" - ann_path.write_text(json.dumps(body, indent=2, ensure_ascii=False)) + await _atomic_write_json_async(str(ann_path), body) return JSONResponse({"status": "ok", "saved": len(body)}) except Exception as e: logger.error(f"Failed to save annotations: {e}") - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail="Failed to save annotations") if __name__ == "__main__": import uvicorn diff --git a/router/static/visualizer.html b/router/static/visualizer.html index 7003c340..aed63687 100644 --- a/router/static/visualizer.html +++ b/router/static/visualizer.html @@ -137,9 +137,10 @@

Select a prompt from the list

// Load annotations try { const resp = await fetch('data/annotations.json'); - annotations = await resp.json(); + const serverAnn = await resp.json(); + annotations = { ...serverAnn, ...annotations }; } catch(e) { - annotations = {}; + // Keep local annotations as is } render(); @@ -221,8 +222,11 @@

Select a prompt from the list

else tags += 'NO DATA'; if (ann) tags += 'REVIEWED'; + const displayLlm = escapeHtml(llm.replace('agent-','')); + const displayPred = pred ? escapeHtml(pred.replace('agent-','')) : ''; + return `
-
#${i} · LLM: ${llm.replace('agent-','')}${pred ? ' · CLS: ' + pred.replace('agent-','') : ''}
+
#${i} · LLM: ${displayLlm}${pred ? ' · CLS: ' + displayPred : ''}
${escapeHtml(p.prompt.substring(0, 120))}
${tags}
`; @@ -274,12 +278,12 @@

Prompt #${idx} · ${p.prompt.length} chars

🤖 LLM Evaluation (kanban worker)

-
${llm}
+
${escapeHtml(llm)}
original dataset label

🧠 Classifier Prediction (gemma4-26a4b)

-
${pred || 'pending...'}
+
${pred ? escapeHtml(pred) : 'pending...'}
${conflict === true ? '⚠ Differs from LLM label' : conflict === false ? '✓ Matches LLM label' : 'no data yet'}
@@ -293,10 +297,10 @@

✎ Human Review

Human selection
- +
- ${ann ? `
✓ Reviewed as ${ann.tier} · clear
` : ''} + ${ann ? `
✓ Reviewed as ${escapeHtml(ann.tier)} · clear
` : ''}
`; } diff --git a/scripts/benchmark_classifier.py b/scripts/benchmark_classifier.py index 7770e8e7..7fab3069 100644 --- a/scripts/benchmark_classifier.py +++ b/scripts/benchmark_classifier.py @@ -41,7 +41,8 @@ def classify(prompt): data = json.loads(resp.read()) return data["choices"][0]["message"].get("content", "").strip() -print(f"Benchmark: gemma4-26a4b-routing vs {dataset['total']} labeled prompts\n") +total = len(dataset.get("prompts", [])) +print(f"Benchmark: gemma4-26a4b-routing vs {total} labeled prompts\n") # Run classification results = [] @@ -49,7 +50,7 @@ def classify(prompt): per_tier = {t: {"correct": 0, "total": 0} for t in TIERS} confusion = defaultdict(Counter) # confusion[expected][predicted] -for i, item in enumerate(dataset["prompts"]): +for i, item in enumerate(dataset.get("prompts", [])): prompt = item["prompt"] expected = item["tier"] @@ -74,13 +75,12 @@ def classify(prompt): # Progress if (i + 1) % 20 == 0: acc = correct / (i + 1) * 100 - print(f" {i+1}/{dataset['total']} — accuracy {acc:.1f}%") + print(f" {i+1}/{total} — accuracy {acc:.1f}%") time.sleep(0.05) # minimal rate-limit (model handles concurrency via llama-server slots) # Report -total = dataset["total"] -overall = correct / total * 100 +overall = (correct / total * 100) if total > 0 else 0.0 print(f"\n{'='*60}") print(f"Overall accuracy: {correct}/{total} ({overall:.1f}%)") diff --git a/scripts/classify_direct.py b/scripts/classify_direct.py index be6fa954..4bd85ff1 100644 --- a/scripts/classify_direct.py +++ b/scripts/classify_direct.py @@ -11,6 +11,11 @@ agent-advanced-core: system-level architecture, cross-cutting concerns, novel design Task: """ +LLAMA_SERVER_URL = "http://127.0.0.1:8080/v1/chat/completions" +TIERS = { + "agent-simple-core", "agent-medium-core", "agent-complex-core", + "agent-reasoning-core", "agent-advanced-core" +} def classify(prompt): payload = { @@ -20,7 +25,7 @@ def classify(prompt): "max_tokens": 15, } req = urllib.request.Request( - "http://127.0.0.1:8080/v1/chat/completions", + LLAMA_SERVER_URL, data=json.dumps(payload).encode(), headers={"Content-Type": "application/json", "Authorization": "Bearer local-token"} ) @@ -44,8 +49,15 @@ def classify(prompt): prompt = prompt[:3500] try: - tier = classify(prompt) - results.append({"id": i, "tier": tier, "prompt_snippet": prompt[:60]}) + raw_tier = classify(prompt) + if raw_tier in TIERS: + tier = raw_tier + extra = {} + else: + tier = "ERROR" + extra = {"raw_output": raw_tier} + errors += 1 + results.append({"id": i, "tier": tier, "prompt_snippet": prompt[:60], **extra}) except Exception as e: tier = f"ERROR" errors += 1 diff --git a/scripts/extract_complex.py b/scripts/extract_complex.py index 2a98c4bc..0460aba1 100644 --- a/scripts/extract_complex.py +++ b/scripts/extract_complex.py @@ -16,13 +16,16 @@ auth = base64.b64encode(f"{pk}:{sk}".encode()).decode() base_url = "http://localhost:3001" -# Load already-classified existing = set() dataset_path = Path(__file__).resolve().parent.parent / "data" / "classified_dataset.json" -with open(dataset_path) as f: - existing_data = json.load(f) -for p in existing_data.get('prompts', []): - existing.add(p['prompt'].strip().lower()) +if dataset_path.exists(): + try: + with open(dataset_path) as f: + existing_data = json.load(f) + for p in existing_data.get('prompts', []): + existing.add(p['prompt'].strip().lower()) + except Exception as e: + print(f"Warning: Failed to load existing dataset: {e}") print(f"Already classified: {len(existing)} prompts") @@ -116,6 +119,9 @@ def looks_complex(prompt): print(f"\nSaved {len(prompts)} complex prompts to {out_path}") lengths = [len(p['prompt']) for p in prompts] -print(f"Length range: {min(lengths)}-{max(lengths)} chars, avg: {sum(lengths)/len(lengths):.0f}") -for p in prompts[:5]: - print(f" [{p['timestamp'][:19]}] ({len(p['prompt'])} chars) {p['prompt'][:120]}...") +if lengths: + print(f"Length range: {min(lengths)}-{max(lengths)} chars, avg: {sum(lengths)/len(lengths):.0f}") + for p in prompts[:5]: + print(f" [{p['timestamp'][:19]}] ({len(p['prompt'])} chars) {p['prompt'][:120]}...") +else: + print("No prompts collected.") diff --git a/scripts/extract_gapfill.py b/scripts/extract_gapfill.py index 0eb61615..a8a30781 100644 --- a/scripts/extract_gapfill.py +++ b/scripts/extract_gapfill.py @@ -1,5 +1,5 @@ """Gap-fill extraction: pull longer/older prompts targeting complex+ tiers.""" -import os, base64, json, urllib.request, time +import os, base64, json, urllib.request, time, re from pathlib import Path env = {} @@ -62,8 +62,10 @@ def is_trivial(prompt): "how are you", "good morning", "what model are you", "who are you", "tell me a joke", "what is 2+2", "what is the capital"] for pat in trivial: - if pat in lower and len(lower) < 50: - return True + if len(lower) < 50: + escaped = re.escape(pat) + if re.search(r'\b' + escaped + r'\b', lower): + return True return False print("Extracting gap-fill prompts (longer, older, targeting complex+)...") @@ -128,7 +130,10 @@ def is_trivial(prompt): print(f"\nSaved {len(prompts)} gap-fill prompts to {out_path}") lengths = [len(p['prompt']) for p in prompts] -print(f"Length range: {min(lengths)}-{max(lengths)} chars, avg: {sum(lengths)/len(lengths):.0f}") -print(f"Sample:") -for p in prompts[:5]: - print(f" [{p['timestamp'][:19]}] ({len(p['prompt'])} chars) {p['prompt'][:100]}...") +if lengths: + print(f"Length range: {min(lengths)}-{max(lengths)} chars, avg: {sum(lengths)/len(lengths):.0f}") + print(f"Sample:") + for p in prompts[:5]: + print(f" [{p['timestamp'][:19]}] ({len(p['prompt'])} chars) {p['prompt'][:100]}...") +else: + print("No prompts collected.") diff --git a/scripts/reclassify_all.py b/scripts/reclassify_all.py index 5170d552..46bcd1fb 100644 --- a/scripts/reclassify_all.py +++ b/scripts/reclassify_all.py @@ -1,12 +1,12 @@ """Re-run gemma4 classifier (with grammar) on all dataset prompts via router.""" -import json, urllib.request, time, sys +import json, urllib.request, time, sys, os, tempfile from pathlib import Path from collections import Counter TIERS = ['agent-simple-core','agent-medium-core','agent-complex-core','agent-reasoning-core','agent-advanced-core'] -ROUTER_URL = "http://127.0.0.1:8080/v1/chat/completions" -print(f"Using router on {ROUTER_URL}") +LLAMA_SERVER_URL = "http://127.0.0.1:8080/v1/chat/completions" +print(f"Using llama-server on {LLAMA_SERVER_URL}") PROMPT_TEMPLATE = """Analyze the request complexity. Respond with exactly one of: - simple boilerplate: agent-simple-core @@ -18,15 +18,13 @@ Request: """ def classify(prompt): - if len(prompt) > 600: - prompt = prompt[:600] payload = { 'model': 'gemma4-26a4b-routing', 'messages': [{'role': 'user', 'content': PROMPT_TEMPLATE + prompt}], 'max_tokens': 15, 'temperature': 0, 'grammar': 'root ::= "agent-simple-core" | "agent-medium-core" | "agent-complex-core" | "agent-reasoning-core" | "agent-advanced-core"' } - req = urllib.request.Request(ROUTER_URL, data=json.dumps(payload).encode(), headers={'Content-Type':'application/json','Authorization':'Bearer local-token'}) + req = urllib.request.Request(LLAMA_SERVER_URL, data=json.dumps(payload).encode(), headers={'Content-Type':'application/json','Authorization':'Bearer local-token'}) with urllib.request.urlopen(req, timeout=30) as resp: data = json.loads(resp.read()) return data['choices'][0]['message'].get('content','').strip() @@ -77,25 +75,33 @@ def classify(prompt): llm_counts = Counter(r['llm_tier'] for r in results) agree = sum(1 for r in results if r['llm_tier'] == r['clf_tier']) +total_results = len(results) print(f"\n{'='*60}") -print(f"Agreement: {agree}/{len(results)} ({agree/len(results)*100:.1f}%)") -print(f"\nTier distribution:") +if total_results > 0: + print(f"Agreement: {agree}/{total_results} ({agree/total_results*100:.1f}%)") +else: + print("Agreement: 0/0 (0.0%)") +print("\nTier distribution:") print(f"{'Tier':30s} {'LLM':>6s} {'CLF':>6s} {'Δ':>6s}") for t in TIERS: lc = llm_counts.get(t, 0) cc = clf_counts.get(t, 0) print(f" {t:30s} {lc:>6d} {cc:>6d} {cc-lc:>+6d}") -# Save combined dataset +# Save combined dataset atomically combined = { - 'total': len(results), - 'agreement': round(agree / len(results) * 100, 1), + 'total': total_results, + 'agreement': round(agree / total_results * 100, 1) if total_results > 0 else 0.0, 'llm_counts': dict(llm_counts), 'clf_counts': dict(clf_counts), 'prompts': results, } -with open(data_dir / 'classified_dataset.json', 'w') as f: - json.dump(combined, f, indent=2, ensure_ascii=False) +dest_path = data_dir / 'classified_dataset.json' +with tempfile.NamedTemporaryFile('w', dir=str(data_dir), delete=False, encoding='utf-8') as tmp_f: + json.dump(combined, tmp_f, indent=2, ensure_ascii=False) + tmp_name = tmp_f.name -print(f"\nSaved to classified_dataset.json (now with llm_tier + clf_tier)") +os.replace(tmp_name, str(dest_path)) + +print("\nSaved to classified_dataset.json (now with llm_tier + clf_tier)") diff --git a/scripts/retry_errors.py b/scripts/retry_errors.py index 78b655aa..39802a5b 100644 --- a/scripts/retry_errors.py +++ b/scripts/retry_errors.py @@ -53,7 +53,7 @@ def classify(prompt): for tier in TIERS: if tier in content: return tier - return content + return "ERROR" TIERS = ['agent-simple-core','agent-medium-core','agent-complex-core','agent-reasoning-core','agent-advanced-core'] @@ -72,7 +72,7 @@ def classify(prompt): for batch_start in range(0, len(error_indices), 5): batch = error_indices[batch_start:batch_start + 5] for idx in batch: - prompt = all_prompts[idx]['prompt'] + prompt = dataset['prompts'][idx]['prompt'] try: tier = classify(prompt) dataset['prompts'][idx]['tier'] = tier diff --git a/test_a2_verify.py b/test_a2_verify.py index 13c872c0..b88a03a4 100644 --- a/test_a2_verify.py +++ b/test_a2_verify.py @@ -8,9 +8,10 @@ import asyncio, time b = get_breaker() -b.tier = 3 -b.cooldown_until = time.time() + 18000 -b.probe_granted = False +for sub in (b.google, b.vendor): + sub.tier = 3 + sub.cooldown_until = time.time() + 18000 + sub.probe_granted = False result = asyncio.run(try_agy_proxy('test prompt')) assert result is None, f'Breaker should return None when blocked, got: {result}' diff --git a/test_circuit_breaker.py b/test_circuit_breaker.py index 932c2960..9e43f5ae 100644 --- a/test_circuit_breaker.py +++ b/test_circuit_breaker.py @@ -1,14 +1,16 @@ #!/usr/bin/env python3 """ -Integration test for the agy circuit breaker. +Integration test for the agy dual circuit breaker. -Simulates 4 consecutive quota failures and verifies: +Simulates consecutive quota failures and verifies: + - Independent google and vendor breakers - Tier 1 cooldown (5 min) after 1st failure - Tier 2 cooldown (30 min) after 2nd failure - Tier 3 cooldown (5 hours) after 3rd failure - Probe behavior: one allowed attempt after cooldown - Reset to Tier 0 on success - Stay at Tier 3 on repeated failure + - Backward compatibility of master breaker methods """ import sys @@ -18,118 +20,156 @@ from circuit_breaker import get_breaker, TIER_COOLDOWNS, MAX_TIER +def reset_breakers(): + b = get_breaker() + for sub in (b.google, b.vendor): + sub.tier = 0 + sub.cooldown_until = 0.0 + sub.probe_granted = False + sub.total_trips = 0 + sub.last_trip_time = 0.0 + + def test_initial_state(): """Breaker starts at Tier 0 (open).""" + reset_breakers() b = get_breaker() - b.tier = 0 - b.cooldown_until = 0 - b.probe_granted = False - b.total_trips = 0 assert b.is_allowed() == True assert b.tier == 0 - print("✓ Initial state: Tier 0, agy allowed") + assert b.google.is_allowed() == True + assert b.vendor.is_allowed() == True + print("✓ Initial state: Tier 0, allowed") def test_first_failure_trips_to_tier1(): """1st failure → Tier 1, 5 min cooldown.""" + reset_breakers() b = get_breaker() - b.tier = 0 - b.cooldown_until = 0 - b.probe_granted = False - b.record_failure() - assert b.tier == 1, f"Expected tier 1, got {b.tier}" - assert b.cooldown_until > time.time(), "Cooldown should be set" - assert b.is_allowed() == False, "Should block during cooldown" - print("✓ 1st failure → Tier 1 (5 min cooldown)") + + b.google.record_failure() + assert b.google.tier == 1 + assert b.google.cooldown_until > time.time() + assert b.google.is_allowed() == False + + # Master breaker is still allowed because vendor is allowed (backward compatible fallback) + assert b.is_allowed() == True + print("✓ 1st failure → Tier 1 (5 min cooldown) on google breaker") def test_probe_granted_after_cooldown(): """After cooldown expires, exactly one probe is allowed.""" + reset_breakers() b = get_breaker() - b.tier = 1 - b.cooldown_until = time.time() - 10 # expired 10s ago - b.probe_granted = False - assert b.is_allowed() == True, "Probe should be granted" - assert b.probe_granted == True, "Probe flag should be set" - assert b.is_allowed() == False, "Second call should be denied" + + b.google.tier = 1 + b.google.cooldown_until = time.time() - 10 # expired 10s ago + b.google.probe_granted = False + + assert b.google.is_allowed() == True, "Probe should be granted" + assert b.google.probe_granted == True, "Probe flag should be set" + assert b.google.is_allowed() == False, "Second call should be denied" print("✓ Probe granted after cooldown expiry, consumed on next check") def test_probe_failure_advances_tier(): """Probe failure → advance to next tier.""" + reset_breakers() b = get_breaker() - b.tier = 1 - b.cooldown_until = time.time() - 10 - b.probe_granted = True # probe was granted - b.record_failure() # probe fails - assert b.tier == 2, f"Expected tier 2, got {b.tier}" - assert b.probe_granted == False + + b.google.tier = 1 + b.google.cooldown_until = time.time() - 10 + b.google.probe_granted = True # probe was granted + b.google.record_failure() # probe fails + + assert b.google.tier == 2, f"Expected tier 2, got {b.google.tier}" + assert b.google.probe_granted == False print("✓ Failed probe at Tier 1 → advanced to Tier 2 (30 min)") def test_tier3_stays_at_tier3(): """At Tier 3, failure → stays at Tier 3 (renews cooldown).""" + reset_breakers() b = get_breaker() - b.tier = MAX_TIER - b.cooldown_until = time.time() - 10 - b.probe_granted = True - old_until = b.cooldown_until - b.record_failure() - assert b.tier == MAX_TIER, "Should stay at Tier 3" - assert b.cooldown_until > old_until, "Cooldown should be renewed" - assert b.probe_granted == False + + b.google.tier = MAX_TIER + b.google.cooldown_until = time.time() - 10 + b.google.probe_granted = True + old_until = b.google.cooldown_until + b.google.record_failure() + + assert b.google.tier == MAX_TIER, "Should stay at Tier 3" + assert b.google.cooldown_until > old_until, "Cooldown should be renewed" + assert b.google.probe_granted == False print("✓ Tier 3 failure → stays at Tier 3 (renews 5-hour cooldown)") def test_success_resets(): """Success at any tier → reset to Tier 0.""" + reset_breakers() b = get_breaker() - b.tier = 2 - b.cooldown_until = time.time() + 1000 - b.probe_granted = False + + b.google.tier = 2 + b.google.cooldown_until = time.time() + 1000 + b.google.probe_granted = False + b.google.record_success() + + assert b.google.tier == 0 + assert b.google.is_allowed() == True + print("✓ Success resets breaker to Tier 0 from any tier") + + +def test_backward_compatibility(): + """Master breaker record_failure and record_success affect both breakers.""" + reset_breakers() + b = get_breaker() + + b.record_failure() + assert b.google.tier == 1 + assert b.vendor.tier == 1 + assert b.is_allowed() == False # both blocked + b.record_success() - assert b.tier == 0 + assert b.google.tier == 0 + assert b.vendor.tier == 0 assert b.is_allowed() == True - print("✓ Success resets breaker to Tier 0 from any tier") + print("✓ Master record_failure and record_success maintain compatibility") def test_full_cycle(): """Complete cycle: success → 3 failures → probe success → reset.""" + reset_breakers() b = get_breaker() - b.tier = 0 - b.cooldown_until = 0 - b.probe_granted = False - b.total_trips = 0 + sub = b.google # Operate normally - assert b.is_allowed() - b.record_success() - assert b.tier == 0 + assert sub.is_allowed() + sub.record_success() + assert sub.tier == 0 # 1st failure - b.record_failure() - assert b.tier == 1 - assert not b.is_allowed() + sub.record_failure() + assert sub.tier == 1 + assert not sub.is_allowed() # Simulate cooldown expiry - b.cooldown_until = time.time() - 10 - assert b.is_allowed() # probe granted - b.record_failure() # probe fails - assert b.tier == 2 + sub.cooldown_until = time.time() - 10 + assert sub.is_allowed() # probe granted + sub.record_failure() # probe fails + assert sub.tier == 2 # Simulate cooldown expiry - b.cooldown_until = time.time() - 10 - assert b.is_allowed() # probe granted - b.record_failure() # probe fails again - assert b.tier == 3 + sub.cooldown_until = time.time() - 10 + assert sub.is_allowed() # probe granted + sub.record_failure() # probe fails again + assert sub.tier == 3 assert TIER_COOLDOWNS[3] == 18000, "Tier 3 must be 5 hours" # Simulate cooldown expiry + probe success - b.cooldown_until = time.time() - 10 - assert b.is_allowed() # probe granted - b.record_success() # probe succeeds - assert b.tier == 0 - assert b.total_trips == 3 + sub.cooldown_until = time.time() - 10 + assert sub.is_allowed() # probe granted + sub.record_success() # probe succeeds + assert sub.tier == 0 + assert sub.total_trips == 3 print("✓ Full cycle: 3 failures → Tier 3 → probe success → reset") @@ -141,6 +181,7 @@ def test_full_cycle(): test_probe_failure_advances_tier() test_tier3_stays_at_tier3() test_success_resets() + test_backward_compatibility() test_full_cycle() print("\n" + "=" * 60) diff --git a/verify_breaker.py b/verify_breaker.py index 4dbbd194..8c95bcb7 100644 --- a/verify_breaker.py +++ b/verify_breaker.py @@ -5,13 +5,18 @@ b = get_breaker() assert b.is_allowed() == True, 'Tier 0 should be open' -b.record_failure() -assert b.tier == 1, 'Should be at Tier 1' -assert b.is_allowed() == False, 'Tier 1 should block (cooldown active)' -# Force cooldown expiry -b.cooldown_until = 0 -assert b.is_allowed() == True, 'Probe should be granted' -assert b.probe_granted == True -b.record_failure() # probe fails -assert b.tier == 2, 'Should advance to Tier 2' + +for sub in (b.google, b.vendor): + assert sub.is_allowed() == True + sub.record_failure() + assert sub.tier == 1, 'Should be at Tier 1' + assert sub.is_allowed() == False, 'Tier 1 should block (cooldown active)' + # Force cooldown expiry + sub.cooldown_until = 0 + assert sub.is_allowed() == True, 'Probe should be granted' + assert sub.probe_granted == True + sub.record_failure() # probe fails + assert sub.tier == 2, 'Should advance to Tier 2' + +assert b.tier == 2 print('All assertions passed') From 43f0eefd5b555c675e83b04ed72b0dbc4a95dbe1 Mon Sep 17 00:00:00 2001 From: sheepdestroyer Date: Thu, 18 Jun 2026 10:46:09 +0200 Subject: [PATCH 13/21] fix: prevent file descriptor leaks, deduplicate save stats, and log background writes --- router/main.py | 66 +++++++++++++++++++++++++++----------------------- 1 file changed, 36 insertions(+), 30 deletions(-) diff --git a/router/main.py b/router/main.py index 05d906fa..4801af2f 100644 --- a/router/main.py +++ b/router/main.py @@ -159,28 +159,24 @@ def _atomic_write_json_sync(path: str, data) -> None: """Synchronously write JSON data to path using atomic temp-file + os.replace.""" os.makedirs(os.path.dirname(path), exist_ok=True) fd, tmp_path = tempfile.mkstemp(dir=os.path.dirname(path), suffix=".tmp") - f = None try: - f = os.fdopen(fd, "w") - json.dump(data, f, indent=2) - f.close() - f = None - os.replace(tmp_path, path) - except Exception: - if f is not None: - try: - f.close() - except Exception: - pass - else: + try: + f = os.fdopen(fd, "w") + except Exception: + os.close(fd) + raise + + try: + with f: + json.dump(data, f, indent=2) + os.replace(tmp_path, path) + except Exception: try: - os.close(fd) + os.unlink(tmp_path) except Exception: pass - try: - os.unlink(tmp_path) - except Exception: - pass + raise + except Exception: raise @@ -706,21 +702,18 @@ def record_tool_usage(tool_name: str, prompt_tokens: int, completion_tokens: int if len(stats["timeline"]) > 15: stats["timeline"].pop(0) - # Fire-and-forget stats write via thread pool (non-blocking) - global _last_stats_save + # Fire-and-forget stats write via save_persisted_stats (non-blocking) now = time.monotonic() - if now - _last_stats_save >= 2.0: + try: + loop = asyncio.get_running_loop() + loop.create_task(save_persisted_stats()) + except RuntimeError: + # No running event loop (e.g. during early startup) — fall back to sync write try: - loop = asyncio.get_running_loop() - loop.run_in_executor(None, _atomic_write_json_sync, STATS_JSON_PATH, copy.deepcopy(dict(stats))) - _last_stats_save = now - except RuntimeError: - # No running event loop (e.g. during early startup) — fall back to sync write - try: + global _last_stats_save + if now - _last_stats_save >= 2.0: _atomic_write_json_sync(STATS_JSON_PATH, stats) _last_stats_save = now - except Exception as e: - logger.error(f"Failed to persist stats to disk: {e}") except Exception as e: logger.error(f"Failed to persist stats to disk: {e}") @@ -729,8 +722,21 @@ def record_tool_usage(tool_name: str, prompt_tokens: int, completion_tokens: int if now - getattr(record_tool_usage, "_last_save", 0.0) >= 2.0: try: loop = asyncio.get_running_loop() - loop.run_in_executor(None, _atomic_write_json_sync, timeline_path, copy.deepcopy(list(stats["timeline"]))) + fut = loop.run_in_executor( + None, + _atomic_write_json_sync, + timeline_path, + copy.deepcopy(list(stats["timeline"])) + ) record_tool_usage._last_save = now + + def done_callback(f): + try: + f.result() + except Exception as e: + logger.warning(f"Failed to persist timeline in background: {e}") + + fut.add_done_callback(done_callback) except RuntimeError: # No running event loop — fall back to sync write try: From ef47704f5255eff9ecef47632b9df12674c10505 Mon Sep 17 00:00:00 2001 From: sheepdestroyer Date: Thu, 18 Jun 2026 10:54:17 +0200 Subject: [PATCH 14/21] Address code review feedback from pullrequestreview-4523251587 --- router/circuit_breaker.py | 2 -- router/main.py | 26 +++++++++++++++----------- router/static/visualizer.html | 12 ++++++------ scripts/benchmark_classifier.py | 10 +++++++--- scripts/retry_errors.py | 17 +++++++++++------ 5 files changed, 39 insertions(+), 28 deletions(-) diff --git a/router/circuit_breaker.py b/router/circuit_breaker.py index b40a1969..eb8c5a6a 100644 --- a/router/circuit_breaker.py +++ b/router/circuit_breaker.py @@ -84,8 +84,6 @@ def record_failure(self): if self.tier == 0: new_tier = 1 - elif self.probe_granted: - new_tier = min(self.tier + 1, MAX_TIER) else: new_tier = min(self.tier + 1, MAX_TIER) diff --git a/router/main.py b/router/main.py index 4801af2f..78d8f0f8 100644 --- a/router/main.py +++ b/router/main.py @@ -15,6 +15,8 @@ from fastapi.staticfiles import StaticFiles from pathlib import Path from circuit_breaker import get_breaker +from pydantic import BaseModel +from typing import Dict, Optional, Union # Configure logging — respect LOG_LEVEL env var (default: WARNING) _log_level_str = os.getenv("LOG_LEVEL", "WARNING").upper() @@ -166,17 +168,14 @@ def _atomic_write_json_sync(path: str, data) -> None: os.close(fd) raise + with f: + json.dump(data, f, indent=2) + os.replace(tmp_path, path) + except Exception: try: - with f: - json.dump(data, f, indent=2) - os.replace(tmp_path, path) + os.unlink(tmp_path) except Exception: - try: - os.unlink(tmp_path) - except Exception: - pass - raise - except Exception: + pass raise @@ -2547,11 +2546,16 @@ async def get_visualizer(): return HTMLResponse(vis_path.read_text()) return HTMLResponse("

Visualizer not found

", status_code=404) +class AnnotationItem(BaseModel): + tier: Union[int, str, None] = None + note: str = "" + ts: Optional[str] = None + @app.post("/dashboard/save-annotations") -async def save_annotations(request: Request): +async def save_annotations(payload: Dict[str, AnnotationItem]): """Save human review annotations to disk.""" try: - body = await request.json() + body = {k: (v.model_dump() if hasattr(v, "model_dump") else v.dict()) for k, v in payload.items()} ann_path = DATA_DIR / "annotations.json" await _atomic_write_json_async(str(ann_path), body) return JSONResponse({"status": "ok", "saved": len(body)}) diff --git a/router/static/visualizer.html b/router/static/visualizer.html index aed63687..3017bd6c 100644 --- a/router/static/visualizer.html +++ b/router/static/visualizer.html @@ -161,7 +161,7 @@

Select a prompt from the list

function isConflict(idx) { const p = dataset[idx]; if (!p) return null; - const llm = p.llm_tier || p.tier; // support both old and new format + const llm = p.llm_tier || p.tier || '?'; // support both old and new format const pred = p.clf_tier || getBenchmarkPrediction(idx); if (!pred || pred.startsWith('ERROR')) return null; return pred !== llm; @@ -211,7 +211,7 @@

Select a prompt from the list

const listEl = document.getElementById('list'); listEl.innerHTML = filtered.map(i => { const p = dataset[i]; - const llm = p.llm_tier || p.tier; + const llm = p.llm_tier || p.tier || '?'; const pred = p.clf_tier; const ann = getAnnotation(i); const conflict = isConflict(i); @@ -222,7 +222,7 @@

Select a prompt from the list

else tags += 'NO DATA'; if (ann) tags += 'REVIEWED'; - const displayLlm = escapeHtml(llm.replace('agent-','')); + const displayLlm = typeof llm === 'string' ? escapeHtml(llm.replace('agent-','')) : '?'; const displayPred = pred ? escapeHtml(pred.replace('agent-','')) : ''; return `
@@ -244,7 +244,7 @@

Select a prompt from the list

function renderDetail(idx) { const p = dataset[idx]; - const llm = p.llm_tier || p.tier; + const llm = p.llm_tier || p.tier || '?'; const pred = p.clf_tier; const ann = getAnnotation(idx); const conflict = pred && !pred.startsWith('ERROR') ? (pred !== llm) : null; @@ -259,7 +259,7 @@

Select a prompt from the list

llmCardClass += ' match'; } - const effectiveTier = ann ? ann.tier : (conflict === false ? p.tier : '?'); + const effectiveTier = ann ? ann.tier : (conflict === false ? (p.tier || p.llm_tier || '?') : '?'); let tierBtns = TIERS.map(tier => { let cls = 'tier-btn'; @@ -314,7 +314,7 @@

✎ Human Review

function saveNote(idx) { const note = document.getElementById('note-input').value; if (!annotations[idx]) { - annotations[idx] = { tier: dataset[idx].tier, note: '', ts: new Date().toISOString() }; + annotations[idx] = { tier: dataset[idx].tier || dataset[idx].llm_tier || '?', note: '', ts: new Date().toISOString() }; } annotations[idx].note = note; saveAnnotations(); diff --git a/scripts/benchmark_classifier.py b/scripts/benchmark_classifier.py index 7fab3069..443c6e19 100644 --- a/scripts/benchmark_classifier.py +++ b/scripts/benchmark_classifier.py @@ -100,12 +100,16 @@ def classify(prompt): row = f"{exp_tier:30s}" for pred_tier in TIERS: count = confusion[exp_tier].get(pred_tier, 0) + count_str = f"{count:3d}" if exp_tier == pred_tier: - row += f" \033[32m{count:3d}\033[0m" + " " * 20 + cell = f" \033[32m{count_str}\033[0m" + row += f"{cell:34s}" elif count > 0: - row += f" \033[31m{count:3d}\033[0m" + " " * 20 + cell = f" \033[31m{count_str}\033[0m" + row += f"{cell:34s}" else: - row += f" {count:3d}" + " " * 20 + cell = f" {count_str}" + row += f"{cell:25s}" print(row) # Save detailed results diff --git a/scripts/retry_errors.py b/scripts/retry_errors.py index 39802a5b..229114f7 100644 --- a/scripts/retry_errors.py +++ b/scripts/retry_errors.py @@ -22,7 +22,7 @@ def get_model_port(): data = json.loads(resp.read()) for m in data.get('data', []): if 'gemma4-26a4b' in m.get('id', ''): - args = m['status']['args'] + args = m.get('status', {}).get('args', []) for i, v in enumerate(args): if v == '--port': return args[i + 1] @@ -48,7 +48,10 @@ def classify(prompt): ) with urllib.request.urlopen(req, timeout=30) as resp: data = json.loads(resp.read()) - content = data["choices"][0]["message"].get("content", "").strip() + choices = data.get("choices", []) + content = "" + if choices: + content = choices[0].get("message", {}).get("content", "").strip() # Normalize: strip "tier:" prefix, extract just the tier name for tier in TIERS: if tier in content: @@ -63,7 +66,7 @@ def classify(prompt): with open(data_dir / "raw_prompts_hermes.json") as f: all_prompts = json.load(f) -error_indices = [i for i, p in enumerate(dataset['prompts']) if p['tier'] == 'ERROR'] +error_indices = [i for i, p in enumerate(dataset.get('prompts', [])) if p.get('tier') == 'ERROR'] print(f"Retrying {len(error_indices)} failed prompts (max {MAX_CHARS} chars)...") fixed = 0 @@ -72,10 +75,12 @@ def classify(prompt): for batch_start in range(0, len(error_indices), 5): batch = error_indices[batch_start:batch_start + 5] for idx in batch: - prompt = dataset['prompts'][idx]['prompt'] + prompts_list = dataset.get('prompts', []) + prompt = prompts_list[idx].get('prompt') if idx < len(prompts_list) else "" try: tier = classify(prompt) - dataset['prompts'][idx]['tier'] = tier + if idx < len(prompts_list): + prompts_list[idx]['tier'] = tier fixed += 1 except Exception as e: errors += 1 @@ -86,7 +91,7 @@ def classify(prompt): time.sleep(5) from collections import Counter -new_counts = Counter(p['tier'] for p in dataset['prompts']) +new_counts = Counter(p.get('tier', 'ERROR') for p in dataset.get('prompts', [])) dataset['counts'] = {k: v for k, v in new_counts.items()} dataset['gaps'] = [t for t in ['agent-simple-core','agent-medium-core','agent-complex-core','agent-reasoning-core','agent-advanced-core'] if new_counts.get(t, 0) < 20] From 764fe646179bd68f99e3b0fd57707004236794b2 Mon Sep 17 00:00:00 2001 From: sheepdestroyer Date: Thu, 18 Jun 2026 11:00:34 +0200 Subject: [PATCH 15/21] Add robustness fixes for reclassify/retry datasets, utf-8, and error index bounds --- router/main.py | 2 +- scripts/extract_prompts.py | 2 +- scripts/reclassify_all.py | 2 +- scripts/retry_errors.py | 4 ++-- 4 files changed, 5 insertions(+), 5 deletions(-) diff --git a/router/main.py b/router/main.py index 78d8f0f8..fe28b56e 100644 --- a/router/main.py +++ b/router/main.py @@ -163,7 +163,7 @@ def _atomic_write_json_sync(path: str, data) -> None: fd, tmp_path = tempfile.mkstemp(dir=os.path.dirname(path), suffix=".tmp") try: try: - f = os.fdopen(fd, "w") + f = os.fdopen(fd, "w", encoding="utf-8") except Exception: os.close(fd) raise diff --git a/scripts/extract_prompts.py b/scripts/extract_prompts.py index d5d19b50..028baa8d 100644 --- a/scripts/extract_prompts.py +++ b/scripts/extract_prompts.py @@ -59,7 +59,7 @@ def extract_first_user_prompt(obs): if isinstance(inp, str): try: inp = json.loads(inp) - except: + except Exception: return None if not isinstance(inp, dict): return None diff --git a/scripts/reclassify_all.py b/scripts/reclassify_all.py index 46bcd1fb..3de3f1d4 100644 --- a/scripts/reclassify_all.py +++ b/scripts/reclassify_all.py @@ -50,7 +50,7 @@ def classify(prompt): prompt = item['prompt'] # Original LLM/kanban eval - llm_tier = item.get('tier', '?') + llm_tier = item.get('llm_tier') or item.get('tier', '?') # Classifier eval try: diff --git a/scripts/retry_errors.py b/scripts/retry_errors.py index 229114f7..dc9aaae4 100644 --- a/scripts/retry_errors.py +++ b/scripts/retry_errors.py @@ -24,7 +24,7 @@ def get_model_port(): if 'gemma4-26a4b' in m.get('id', ''): args = m.get('status', {}).get('args', []) for i, v in enumerate(args): - if v == '--port': + if v == '--port' and i + 1 < len(args): return args[i + 1] raise RuntimeError("gemma4-26a4b-routing model port not found") @@ -91,7 +91,7 @@ def classify(prompt): time.sleep(5) from collections import Counter -new_counts = Counter(p.get('tier', 'ERROR') for p in dataset.get('prompts', [])) +new_counts = Counter(p.get('llm_tier') or p.get('tier', 'ERROR') for p in dataset.get('prompts', [])) dataset['counts'] = {k: v for k, v in new_counts.items()} dataset['gaps'] = [t for t in ['agent-simple-core','agent-medium-core','agent-complex-core','agent-reasoning-core','agent-advanced-core'] if new_counts.get(t, 0) < 20] From 9bfbdba4bd8c94afc22fc3c4357569a97e0bf65c Mon Sep 17 00:00:00 2001 From: sheepdestroyer Date: Thu, 18 Jun 2026 11:06:38 +0200 Subject: [PATCH 16/21] Use absolute paths in visualizer fetch requests to prevent trailing slash 404s --- router/static/visualizer.html | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/router/static/visualizer.html b/router/static/visualizer.html index 3017bd6c..e31d4352 100644 --- a/router/static/visualizer.html +++ b/router/static/visualizer.html @@ -112,7 +112,7 @@

Select a prompt from the list

// Load data async function init() { try { - const resp = await fetch('data/classified_dataset.json'); + const resp = await fetch('/data/classified_dataset.json'); const raw = await resp.json(); dataset = raw.prompts || []; // Extract global stats from combined dataset @@ -127,7 +127,7 @@

Select a prompt from the list

// Load benchmark results if available try { - const resp = await fetch('data/benchmark_results.json'); + const resp = await fetch('/data/benchmark_results.json'); benchmark = await resp.json(); // Map benchmark results to dataset by index } catch(e) { @@ -136,7 +136,7 @@

Select a prompt from the list

// Load annotations try { - const resp = await fetch('data/annotations.json'); + const resp = await fetch('/data/annotations.json'); const serverAnn = await resp.json(); annotations = { ...serverAnn, ...annotations }; } catch(e) { From 9c8f286ea92085e485e462724ad013465a748eab Mon Sep 17 00:00:00 2001 From: sheepdestroyer Date: Thu, 18 Jun 2026 11:08:58 +0200 Subject: [PATCH 17/21] Add GitHub Actions testing workflow and correct Dependabot Docker directory --- .github/dependabot.yml | 2 +- .github/workflows/test.yml | 28 ++++++++++++++++++++++++++++ test_a2_verify.py | 3 ++- test_circuit_breaker.py | 3 ++- verify_breaker.py | 4 ++++ 5 files changed, 37 insertions(+), 3 deletions(-) create mode 100644 .github/workflows/test.yml diff --git a/.github/dependabot.yml b/.github/dependabot.yml index 4f44831c..38c056fa 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -2,7 +2,7 @@ version: 2 updates: # Docker image dependency updates - package-ecosystem: "docker" - directory: "/" + directory: "/router" schedule: interval: "daily" time: "18:22" diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 00000000..93a280fb --- /dev/null +++ b/.github/workflows/test.yml @@ -0,0 +1,28 @@ +name: Run Tests + +on: + push: + branches: [ master ] + pull_request: + branches: [ master ] + +jobs: + test: + runs-on: ubuntu-latest + steps: + - name: Checkout Repository + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.11' + + - name: Run Circuit Breaker Tests + run: python3 test_circuit_breaker.py + + - name: Run Breaker Verification + run: python3 verify_breaker.py + + - name: Run Integration Verification + run: python3 test_a2_verify.py diff --git a/test_a2_verify.py b/test_a2_verify.py index b88a03a4..42cde1e1 100644 --- a/test_a2_verify.py +++ b/test_a2_verify.py @@ -1,7 +1,8 @@ #!/usr/bin/env python3 """Verify circuit breaker integration into agy_proxy.py""" import sys -sys.path.insert(0, '/home/gpav/Vrac/LAB/AI/LLM-Routing/router') +from pathlib import Path +sys.path.insert(0, str(Path(__file__).resolve().parent / 'router')) from circuit_breaker import get_breaker from agy_proxy import try_agy_proxy diff --git a/test_circuit_breaker.py b/test_circuit_breaker.py index 9e43f5ae..4db46c0a 100644 --- a/test_circuit_breaker.py +++ b/test_circuit_breaker.py @@ -15,7 +15,8 @@ import sys import time -sys.path.insert(0, '/home/gpav/Vrac/LAB/AI/LLM-Routing/router') +from pathlib import Path +sys.path.insert(0, str(Path(__file__).resolve().parent / 'router')) from circuit_breaker import get_breaker, TIER_COOLDOWNS, MAX_TIER diff --git a/verify_breaker.py b/verify_breaker.py index 8c95bcb7..6f309fcc 100644 --- a/verify_breaker.py +++ b/verify_breaker.py @@ -1,6 +1,10 @@ #!/usr/bin/env python3 """Verification test for the agy circuit breaker.""" +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).resolve().parent)) + from router.circuit_breaker import get_breaker b = get_breaker() From 3b824247867c9afba584f421ced8cfccee7cd95d Mon Sep 17 00:00:00 2001 From: sheepdestroyer Date: Thu, 18 Jun 2026 11:10:14 +0200 Subject: [PATCH 18/21] Fix Dependabot schema validation errors by removing invalid prefix-major and empty ignore blocks --- .github/dependabot.yml | 6 ------ 1 file changed, 6 deletions(-) diff --git a/.github/dependabot.yml b/.github/dependabot.yml index 38c056fa..37aa7f1f 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -11,7 +11,6 @@ updates: separator: "/" commit-message: prefix: "chore(deps)" - prefix-major: "chore(deps)!" include: "scope" labels: - "dependencies" @@ -20,11 +19,6 @@ updates: - "sheepdestroyer" allow: - dependency-type: "direct" - ignore: - # Ignore major version updates for critical services - # Uncomment and adjust as needed - # - dependency-name: "postgres" - # update-types: ["version-update:semver-major"] # GitHub Actions dependency updates (optional - add if using GH Actions) - package-ecosystem: "github-actions" From e37b24b20dbc3aa3f5e4ca3244ddf6ad8d7de792 Mon Sep 17 00:00:00 2001 From: sheepdestroyer Date: Thu, 18 Jun 2026 11:53:30 +0200 Subject: [PATCH 19/21] fix(router): resolve visualizer bugs, dynamic breaker mapping, container build, and docstrings --- router/Containerfile | 2 +- router/agy_proxy.py | 4 ++-- router/circuit_breaker.py | 5 +++++ router/main.py | 10 ++++++++++ router/static/visualizer.html | 5 ++--- scripts/classify_direct.py | 1 + scripts/extract_complex.py | 2 ++ scripts/extract_gapfill.py | 2 ++ scripts/reclassify_all.py | 1 + scripts/retry_errors.py | 1 + 10 files changed, 27 insertions(+), 6 deletions(-) diff --git a/router/Containerfile b/router/Containerfile index d685b94d..c26ad091 100644 --- a/router/Containerfile +++ b/router/Containerfile @@ -6,7 +6,7 @@ WORKDIR /app RUN pip install --no-cache-dir fastapi uvicorn httpx pyyaml python-multipart asyncpg langfuse # Copy all source in one layer — removes dead config COPY (volume-mounted at runtime) -COPY main.py agy_proxy.py circuit_breaker.py aa_scores.json free_models_roster.json best_free_model.json /app/ +COPY main.py agy_proxy.py circuit_breaker.py aa_scores.json free_models_roster.json /app/ COPY static/ /app/static/ EXPOSE 5000 diff --git a/router/agy_proxy.py b/router/agy_proxy.py index 0a2c03c2..918bfcb6 100644 --- a/router/agy_proxy.py +++ b/router/agy_proxy.py @@ -253,7 +253,7 @@ async def try_agy_proxy(prompt: str, messages: list = None, # Determine which breaker to use for this tier # Tier 0 (idx 0): gemini-3.5-flash → google_breaker # Tier 1 (idx 1): claude-opus-4.6 → vendor_breaker - is_google_tier = (actual_tier_idx == 0) + is_google_tier = "gemini" in tier.get("model_name", "").lower() tier_breaker = google_breaker if is_google_tier else vendor_breaker if not tier_breaker.is_allowed(): @@ -323,8 +323,8 @@ async def try_agy_proxy(prompt: str, messages: list = None, # Success! Stream has started. tier_breaker.record_success() - # Define the async generator async def token_generator(stream_resp, httpx_client, initial_line, current_conv_id): + """Asynchronously yields tokens from the agy daemon stream and manages session state updates.""" # Yield the initial token if it was a token init_data = json.loads(initial_line) if init_data.get("type") == "token" and init_data.get("content"): diff --git a/router/circuit_breaker.py b/router/circuit_breaker.py index eb8c5a6a..df17fcd2 100644 --- a/router/circuit_breaker.py +++ b/router/circuit_breaker.py @@ -34,6 +34,7 @@ class PerModelBreaker: """Tracks quota exhaustion for a specific model family with exponential cooldown.""" def __init__(self, name: str): + """Initialize the per-model circuit breaker with a name and default tier/cooldown states.""" self.name = name self.tier: int = 0 # 0 = open (allowed), 1-3 = cooldown active self.cooldown_until: float = 0.0 @@ -128,6 +129,7 @@ class DualCircuitBreaker: """ def __init__(self): + """Initialize the dual circuit breaker with separate google and vendor sub-breakers.""" self.google = PerModelBreaker("google") self.vendor = PerModelBreaker("vendor") @@ -135,6 +137,7 @@ def __init__(self): # Default to checking BOTH — if either allows, return allowed. # This ensures old code without model awareness works correctly. def is_allowed(self) -> bool: + """Check if either the google or vendor breaker allows the request (backward-compat).""" return self.google.is_allowed() or self.vendor.is_allowed() def record_failure(self): @@ -149,9 +152,11 @@ def record_success(self): @property def tier(self) -> int: + """Return the maximum cooldown tier across both google and vendor sub-breakers.""" return max(self.google.tier, self.vendor.tier) def status(self) -> dict: + """Return the aggregated status dictionary of both sub-breakers for the dashboard.""" return { "google": self.google.status(), "vendor": self.vendor.status(), diff --git a/router/main.py b/router/main.py index fe28b56e..c054c6af 100644 --- a/router/main.py +++ b/router/main.py @@ -287,6 +287,7 @@ async def sync_adaptive_router_roster(master_key: str): if max_score < 1.0: max_score = 55.0 # safety floor def norm(s: float) -> float: + """Helper to scale raw model index score against max score in roster to 0-100 range.""" return (s / max_score) * 100.0 for score, mid in free_models: # include all models — top 2 are also assigned to their correct tier n = norm(score) @@ -548,6 +549,7 @@ async def classify_request(prompt: str, bypass_cache: bool = False, langfuse_tra return "agent-advanced-core", latency, False, "advanced (exception)" def get_live_gemini_oauth_token() -> str | None: + """Retrieve the current valid Gemini OAuth access token from local storage if not expired.""" try: creds_path = "/config/gemini_auth/oauth_creds.json" if os.path.exists(creds_path): @@ -730,6 +732,7 @@ def record_tool_usage(tool_name: str, prompt_tokens: int, completion_tokens: int record_tool_usage._last_save = now def done_callback(f): + """Log any uncaught exceptions returned from the background timeline executor thread.""" try: f.result() except Exception as e: @@ -830,6 +833,7 @@ async def get_llamacpp_metrics() -> dict: _AA_SCORES_LOADED = False def _load_aa_scores(): + """Load the Artificial Analysis agentic scores cache from local config.""" global _AA_SCORES_CACHE, _AA_SCORES_LOADED if _AA_SCORES_LOADED: return @@ -1058,6 +1062,7 @@ async def proxy_models(): @app.post("/v1/chat/completions") async def chat_completions(request: Request): + """Handle incoming OpenAI-compatible chat completions requests and route them dynamically based on triage logic.""" global stats start_time = time.time() @@ -1264,6 +1269,7 @@ async def chat_completions(request: Request): if body.get("stream", False): content = agy_response.get("choices", [{}])[0].get("message", {}).get("content", "") async def agy_stream_generator(): + """Asynchronous generator yielding simulated OpenAI-compatible streaming chunks from a static agy response.""" import uuid created_time = int(time.time()) chunk_id = f"chatcmpl-{uuid.uuid4().hex[:12]}" @@ -1421,6 +1427,7 @@ async def agy_stream_generator(): r = await client.send(req, stream=True) if r.status_code == 200: async def stream_generator(): + """Asynchronous generator that yields streaming chunks from LiteLLM completions response and logs usage stats on completion.""" completion_chars = 0 request_tokens = len(json.dumps(body_to_send)) // 4 try: @@ -1558,6 +1565,7 @@ async def metrics(): @app.get("/dashboard", response_class=HTMLResponse) async def get_dashboard(): + """Render the router main dashboard HTML showing system metrics, health checks, and recent token usage.""" # 1. Run live health checks valkey_status = await check_tcp_port("127.0.0.1", 6379) litellm_status = await check_http_endpoint("http://127.0.0.1:4000/") @@ -1813,6 +1821,7 @@ async def get_dashboard(): # Source badge helper: generates a colored inline source tag def src_badge(label, color): + """Generate inline HTML span styled as a colored status/category badge.""" return f"{label}" # 10. Pre-compute llama.cpp HTML cards @@ -2547,6 +2556,7 @@ async def get_visualizer(): return HTMLResponse("

Visualizer not found

", status_code=404) class AnnotationItem(BaseModel): + """Pydantic model representing a single human dataset review annotation.""" tier: Union[int, str, None] = None note: str = "" ts: Optional[str] = None diff --git a/router/static/visualizer.html b/router/static/visualizer.html index e31d4352..54fa6d77 100644 --- a/router/static/visualizer.html +++ b/router/static/visualizer.html @@ -117,7 +117,7 @@

Select a prompt from the list

dataset = raw.prompts || []; // Extract global stats from combined dataset if (raw.agreement !== undefined) { - document.getElementById('stat-agreement').textContent = raw.agreement + '%'; + document.getElementById('stat-agreement').querySelector('.val').textContent = raw.agreement + '%'; document.getElementById('stat-agreement').style.display = ''; } } catch(e) { @@ -259,8 +259,6 @@

Select a prompt from the list

llmCardClass += ' match'; } - const effectiveTier = ann ? ann.tier : (conflict === false ? (p.tier || p.llm_tier || '?') : '?'); - let tierBtns = TIERS.map(tier => { let cls = 'tier-btn'; if (ann && ann.tier === tier) cls += ' selected-human'; @@ -318,6 +316,7 @@

✎ Human Review

} annotations[idx].note = note; saveAnnotations(); + render(); } function clearAnnotation(idx) { diff --git a/scripts/classify_direct.py b/scripts/classify_direct.py index 4bd85ff1..ddf8a99a 100644 --- a/scripts/classify_direct.py +++ b/scripts/classify_direct.py @@ -18,6 +18,7 @@ } def classify(prompt): + """Query the llama-server to classify the prompt task complexity.""" payload = { "model": "gemma4-26a4b-routing", "messages": [{"role": "user", "content": PROMPT_TEMPLATE + prompt}], diff --git a/scripts/extract_complex.py b/scripts/extract_complex.py index 0460aba1..96f3acd0 100644 --- a/scripts/extract_complex.py +++ b/scripts/extract_complex.py @@ -30,6 +30,7 @@ print(f"Already classified: {len(existing)} prompts") def fetch_observations(page=1, limit=50): + """Fetch a paginated list of observations from the Langfuse public API.""" url = f"{base_url}/api/public/observations?limit={limit}&page={page}&orderBy=timestamp.desc&level=DEFAULT&name=litellm-acompletion" req = urllib.request.Request(url) req.add_header("Authorization", f"Basic {auth}") @@ -37,6 +38,7 @@ def fetch_observations(page=1, limit=50): return json.loads(resp.read()) def extract_user_prompt(obs): + """Extract and parse the raw user prompt from the observation input payload.""" inp = obs.get('input') if not inp: return None if isinstance(inp, str): diff --git a/scripts/extract_gapfill.py b/scripts/extract_gapfill.py index a8a30781..86a1d643 100644 --- a/scripts/extract_gapfill.py +++ b/scripts/extract_gapfill.py @@ -36,6 +36,7 @@ def fetch_observations(page=1, limit=50): return json.loads(resp.read()) def extract_user_prompt(obs): + """Extract and parse the raw user prompt from the observation input payload.""" inp = obs.get('input') if not inp: return None @@ -55,6 +56,7 @@ def extract_user_prompt(obs): return None def is_trivial(prompt): + """Check if the prompt matches a list of trivial test patterns to filter out.""" lower = prompt.strip().lower() if len(lower) < 20: return True diff --git a/scripts/reclassify_all.py b/scripts/reclassify_all.py index 3de3f1d4..02042260 100644 --- a/scripts/reclassify_all.py +++ b/scripts/reclassify_all.py @@ -18,6 +18,7 @@ Request: """ def classify(prompt): + """Query the llama-server to classify the prompt complexity with grammar enforcement.""" payload = { 'model': 'gemma4-26a4b-routing', 'messages': [{'role': 'user', 'content': PROMPT_TEMPLATE + prompt}], diff --git a/scripts/retry_errors.py b/scripts/retry_errors.py index dc9aaae4..54ca4967 100644 --- a/scripts/retry_errors.py +++ b/scripts/retry_errors.py @@ -33,6 +33,7 @@ def get_model_port(): print(f"Using model directly on port {MODEL_PORT}") def classify(prompt): + """Query the direct model port to classify the prompt complexity, handling truncations.""" if len(prompt) > MAX_CHARS: prompt = prompt[:MAX_CHARS] payload = { From 219dd8df5998121749118ef5aea11ba168c75115 Mon Sep 17 00:00:00 2001 From: sheepdestroyer Date: Thu, 18 Jun 2026 12:03:58 +0200 Subject: [PATCH 20/21] fix(router): address PR#3 review feedback (endpoint validation, auto-sync, native streaming) --- router/main.py | 214 ++++++++++++++++++++++++++-------- router/static/visualizer.html | 16 +-- 2 files changed, 175 insertions(+), 55 deletions(-) diff --git a/router/main.py b/router/main.py index c054c6af..80aa0245 100644 --- a/router/main.py +++ b/router/main.py @@ -1237,75 +1237,152 @@ async def chat_completions(request: Request): except Exception: pass + is_stream_requested = body.get("stream", False) agy_response = await try_agy_proxy( prompt=last_prompt, messages=messages, session_id=session_id, total_timeout=300.0, + stream=is_stream_requested, target_tier=target_model ) if agy_response: - latency_ms = (time.time() - start_time) * 1000.0 model_name = agy_response.get("model", "gemini-3.5-flash (via agy)") - usage = agy_response.get("usage", {}) - prompt_tokens = usage.get("prompt_tokens", 0) - completion_tokens = usage.get("completion_tokens", 0) - record_tool_usage( - active_tool, prompt_tokens, completion_tokens, - model_name, latency_ms, route="google_oauth_direct" - ) - logger.info(f"✅ agy proxy succeeded: {model_name}, {latency_ms:.0f}ms") - - # Finalize agy span - if agy_span_obj: - try: - agy_span_obj.end( - output={"model": model_name, "tokens": completion_tokens}, - metadata={"latency_ms": latency_ms, "tier": target_model}, - ) - except Exception: - pass - - if body.get("stream", False): - content = agy_response.get("choices", [{}])[0].get("message", {}).get("content", "") - async def agy_stream_generator(): - """Asynchronous generator yielding simulated OpenAI-compatible streaming chunks from a static agy response.""" + + if "stream" in agy_response: + # Real native stream generator + async def native_agy_stream_generator(stream_gen, model_name): + """Asynchronous generator yielding native OpenAI-compatible streaming chunks from the real agy daemon.""" import uuid created_time = int(time.time()) chunk_id = f"chatcmpl-{uuid.uuid4().hex[:12]}" - chunk_size = 40 - for i in range(0, len(content), chunk_size): - chunk_text = content[i:i+chunk_size] - chunk_data = { + token_count = 0 + try: + async for token in stream_gen: + if not token: + continue + token_count += 1 + chunk_data = { + "id": chunk_id, + "object": "chat.completion.chunk", + "created": created_time, + "model": model_name, + "choices": [{ + "index": 0, + "delta": {"content": token}, + "finish_reason": None + }] + } + yield f"data: {json.dumps(chunk_data)}\n\n".encode("utf-8") + + # End of stream chunk + finish_data = { "id": chunk_id, "object": "chat.completion.chunk", "created": created_time, "model": model_name, "choices": [{ "index": 0, - "delta": {"content": chunk_text}, - "finish_reason": None + "delta": {}, + "finish_reason": "stop" }] } - yield f"data: {json.dumps(chunk_data)}\n\n".encode("utf-8") - await asyncio.sleep(0.005) - - finish_data = { - "id": chunk_id, - "object": "chat.completion.chunk", - "created": created_time, - "model": model_name, - "choices": [{ - "index": 0, - "delta": {}, - "finish_reason": "stop" - }] - } - yield f"data: {json.dumps(finish_data)}\n\n".encode("utf-8") - yield b"data: [DONE]\n\n" - return StreamingResponse(agy_stream_generator(), media_type="text/event-stream") + yield f"data: {json.dumps(finish_data)}\n\n".encode("utf-8") + yield b"data: [DONE]\n\n" + + # Success telemetry + latency_ms = (time.time() - start_time) * 1000.0 + # Approximate prompt tokens based on messages characters + prompt_chars = sum(len(m.get("content", "")) for m in messages) + approx_prompt_tokens = max(1, prompt_chars // 4) + + record_tool_usage( + active_tool, approx_prompt_tokens, token_count, + model_name, latency_ms, route="google_oauth_direct" + ) + logger.info(f"✅ native agy stream succeeded: {model_name}, {latency_ms:.0f}ms") + if agy_span_obj: + try: + agy_span_obj.end( + output={"model": model_name, "tokens": token_count}, + metadata={"latency_ms": latency_ms, "tier": target_model}, + ) + except Exception: + pass + except Exception as stream_err: + logger.error(f"Error during native agy stream generation: {stream_err}") + if agy_span_obj: + try: + agy_span_obj.end( + output={"error": str(stream_err)[:200]}, + metadata={"status": "failed"}, + ) + except Exception: + pass + raise + return StreamingResponse(native_agy_stream_generator(agy_response["stream"], model_name), media_type="text/event-stream") else: - return agy_response + latency_ms = (time.time() - start_time) * 1000.0 + usage = agy_response.get("usage", {}) + prompt_tokens = usage.get("prompt_tokens", 0) + completion_tokens = usage.get("completion_tokens", 0) + record_tool_usage( + active_tool, prompt_tokens, completion_tokens, + model_name, latency_ms, route="google_oauth_direct" + ) + logger.info(f"✅ agy proxy succeeded: {model_name}, {latency_ms:.0f}ms") + + # Finalize agy span + if agy_span_obj: + try: + agy_span_obj.end( + output={"model": model_name, "tokens": completion_tokens}, + metadata={"latency_ms": latency_ms, "tier": target_model}, + ) + except Exception: + pass + + if is_stream_requested: + # Robust fallback: simulate stream if we requested stream but got buffered response + content = agy_response.get("choices", [{}])[0].get("message", {}).get("content", "") + async def agy_stream_generator(): + """Asynchronous generator yielding simulated OpenAI-compatible streaming chunks from a static agy response.""" + import uuid + created_time = int(time.time()) + chunk_id = f"chatcmpl-{uuid.uuid4().hex[:12]}" + chunk_size = 40 + for i in range(0, len(content), chunk_size): + chunk_text = content[i:i+chunk_size] + chunk_data = { + "id": chunk_id, + "object": "chat.completion.chunk", + "created": created_time, + "model": model_name, + "choices": [{ + "index": 0, + "delta": {"content": chunk_text}, + "finish_reason": None + }] + } + yield f"data: {json.dumps(chunk_data)}\n\n".encode("utf-8") + await asyncio.sleep(0.005) + + finish_data = { + "id": chunk_id, + "object": "chat.completion.chunk", + "created": created_time, + "model": model_name, + "choices": [{ + "index": 0, + "delta": {}, + "finish_reason": "stop" + }] + } + yield f"data: {json.dumps(finish_data)}\n\n".encode("utf-8") + yield b"data: [DONE]\n\n" + return StreamingResponse(agy_stream_generator(), media_type="text/event-stream") + else: + return agy_response except ImportError: if agy_span_obj: try: @@ -2561,9 +2638,50 @@ class AnnotationItem(BaseModel): note: str = "" ts: Optional[str] = None +VALID_TIERS = {"agent-simple-core", "agent-medium-core", "agent-complex-core", "agent-reasoning-core", "agent-advanced-core"} + @app.post("/dashboard/save-annotations") async def save_annotations(payload: Dict[str, AnnotationItem]): """Save human review annotations to disk.""" + if len(payload) > 1000: + raise HTTPException( + status_code=400, + detail="Payload size limit exceeded: maximum of 1000 annotations allowed per request." + ) + + for k, item in payload.items(): + if not k.isdigit(): + raise HTTPException( + status_code=400, + detail=f"Invalid payload key '{k}': keys must be numeric strings (dataset indexes)." + ) + + t = item.tier + if t is not None: + if isinstance(t, int): + if t < 0 or t > 4: + raise HTTPException( + status_code=400, + detail=f"Invalid tier index {t} for index {k}: must be between 0 and 4." + ) + elif isinstance(t, str): + if t not in VALID_TIERS and t != "?": + raise HTTPException( + status_code=400, + detail=f"Invalid tier string '{t}' for index {k}." + ) + else: + raise HTTPException( + status_code=400, + detail=f"Invalid tier type for index {k}: must be int, str, or null." + ) + + if len(item.note) > 1000: + raise HTTPException( + status_code=400, + detail=f"Note length limit exceeded at index {k}: maximum of 1000 characters allowed." + ) + try: body = {k: (v.model_dump() if hasattr(v, "model_dump") else v.dict()) for k, v in payload.items()} ann_path = DATA_DIR / "annotations.json" diff --git a/router/static/visualizer.html b/router/static/visualizer.html index 54fa6d77..46a67228 100644 --- a/router/static/visualizer.html +++ b/router/static/visualizer.html @@ -326,8 +326,14 @@

✎ Human Review

} function saveAnnotations() { - // Save to localStorage and trigger download + // Save to localStorage localStorage.setItem('classifier-annotations', JSON.stringify(annotations)); + // Auto-sync by posting to dashboard save endpoint + fetch('/dashboard/save-annotations', { + method: 'POST', + headers: {'Content-Type': 'application/json'}, + body: JSON.stringify(annotations), + }).catch((err) => console.error('Failed to auto-sync annotations:', err)); } function exportAnnotations() { @@ -339,12 +345,8 @@

✎ Human Review

a.click(); URL.revokeObjectURL(url); - // Also try to save via POST if a save endpoint exists - fetch('/dashboard/save-annotations', { - method: 'POST', - headers: {'Content-Type': 'application/json'}, - body: JSON.stringify(annotations), - }).catch(() => {}); + // Save locally and sync + saveAnnotations(); } function escapeHtml(text) { From aa9fe101c0f7d768dca585428c12ca0f91e1194b Mon Sep 17 00:00:00 2001 From: sheepdestroyer Date: Thu, 18 Jun 2026 12:23:50 +0200 Subject: [PATCH 21/21] =?UTF-8?q?fix:=20address=20PR#3=20round-2=20review?= =?UTF-8?q?=20=E2=80=94=20robustness,=20a11y,=20and=20code=20style?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - router/main.py: wrap lifespan yield in try/finally so stats flush always runs; add _background_tasks set to prevent fire-and-forget task GC (RUF006) - router/static/visualizer.html: add keyboard accessibility (tabindex, role, onkeydown) to prompt list items; switch annotation keys from array index to stable djb2 prompt hash with backward-compatible index fallback - scripts/benchmark_classifier.py: safe tier field lookup (tier/llm_tier/clf_tier), guard per_tier keying to skip ERROR/unknown labels, safe choices[0] access - scripts/extract_complex.py: use forward iteration for first user message, add system-note filtering ([System: / [Note:]) - scripts/extract_gapfill.py: same as extract_complex.py - scripts/retry_errors.py: safe status.get() to prevent AttributeError, schema-aware ERROR detection (tier + clf_tier), only increment fixed on success - scripts/reclassify_all.py: safe choices[0] fallback to prevent IndexError - test_circuit_breaker.py: replace == True/False with idiomatic assert (14 sites) - verify_breaker.py: replace == True/False with idiomatic assert (5 sites) --- router/main.py | 40 ++++++++++++++++++------------- router/static/visualizer.html | 42 +++++++++++++++++++++++++-------- scripts/benchmark_classifier.py | 21 ++++++++++++----- scripts/extract_complex.py | 17 +++++++++---- scripts/extract_gapfill.py | 17 +++++++++---- scripts/reclassify_all.py | 5 +++- scripts/retry_errors.py | 12 +++++++--- test_circuit_breaker.py | 28 +++++++++++----------- verify_breaker.py | 8 +++---- 9 files changed, 128 insertions(+), 62 deletions(-) diff --git a/router/main.py b/router/main.py index 80aa0245..9db0142c 100644 --- a/router/main.py +++ b/router/main.py @@ -132,6 +132,10 @@ async def push_aggregate_scores(): STATS_JSON_PATH = "/config/router_dir/router_stats.json" +# Module-level set to hold references to fire-and-forget background tasks, +# preventing premature garbage collection before the task completes (Ruff RUF006). +_background_tasks: set = set() + def load_persisted_stats(): """Loads persisted statistics from disk on startup to prevent resets on pod redeployment.""" global stats @@ -390,22 +394,23 @@ async def lifespan(app: FastAPI): # Start background task before yield so it runs during app lifetime task = asyncio.create_task(push_aggregate_scores()) - yield - - # Cancel background score task - task.cancel() try: - await task - except asyncio.CancelledError: - pass + yield + finally: + # Cancel background score task + task.cancel() + try: + await task + except asyncio.CancelledError: + pass - # Flush any buffered stats/timeline on clean shutdown - await save_persisted_stats(force=True) - try: - timeline_path = os.path.join(os.path.dirname(CONFIG_PATH), "router_timeline.json") - await _atomic_write_json_async(timeline_path, stats["timeline"]) - except Exception as e: - logger.warning(f"Failed to persist timeline on shutdown: {e}") + # Flush any buffered stats/timeline on clean shutdown (always runs) + await save_persisted_stats(force=True) + try: + timeline_path = os.path.join(os.path.dirname(CONFIG_PATH), "router_timeline.json") + await _atomic_write_json_async(timeline_path, stats["timeline"]) + except Exception as e: + logger.warning(f"Failed to persist timeline on shutdown: {e}") app = FastAPI(title="LLM Triage Router", lifespan=lifespan) @@ -703,11 +708,14 @@ def record_tool_usage(tool_name: str, prompt_tokens: int, completion_tokens: int if len(stats["timeline"]) > 15: stats["timeline"].pop(0) - # Fire-and-forget stats write via save_persisted_stats (non-blocking) + # Fire-and-forget stats write via save_persisted_stats (non-blocking). + # Store the task reference in _background_tasks to prevent GC before completion (RUF006). now = time.monotonic() try: loop = asyncio.get_running_loop() - loop.create_task(save_persisted_stats()) + _task = loop.create_task(save_persisted_stats()) + _background_tasks.add(_task) + _task.add_done_callback(_background_tasks.discard) except RuntimeError: # No running event loop (e.g. during early startup) — fall back to sync write try: diff --git a/router/static/visualizer.html b/router/static/visualizer.html index 46a67228..3a837ae8 100644 --- a/router/static/visualizer.html +++ b/router/static/visualizer.html @@ -154,8 +154,22 @@

Select a prompt from the list

return null; } +function promptKey(idx) { + // Stable key derived from prompt text (djb2 hash), with index fallback for legacy annotations. + const p = dataset[idx]; + if (!p || !p.prompt) return String(idx); + let hash = 5381; + for (let i = 0; i < Math.min(p.prompt.length, 500); i++) { + hash = ((hash << 5) + hash) + p.prompt.charCodeAt(i); + hash |= 0; // convert to 32-bit int + } + return 'h' + (hash >>> 0).toString(16); +} + function getAnnotation(idx) { - return annotations[idx] || null; + const key = promptKey(idx); + // Stable key lookup first; fall back to legacy index key for backward compatibility + return annotations[key] || annotations[idx] || null; } function isConflict(idx) { @@ -215,17 +229,17 @@

Select a prompt from the list

const pred = p.clf_tier; const ann = getAnnotation(i); const conflict = isConflict(i); - + let tags = ''; if (conflict === true) tags += 'CONFLICT'; else if (conflict === false) tags += 'AGREE'; else tags += 'NO DATA'; if (ann) tags += 'REVIEWED'; - + const displayLlm = typeof llm === 'string' ? escapeHtml(llm.replace('agent-','')) : '?'; const displayPred = pred ? escapeHtml(pred.replace('agent-','')) : ''; - - return `
+ + return `
#${i} · LLM: ${displayLlm}${pred ? ' · CLS: ' + displayPred : ''}
${escapeHtml(p.prompt.substring(0, 120))}
${tags}
@@ -304,23 +318,31 @@

✎ Human Review

} function annotate(idx, tier) { - annotations[idx] = { tier, note: annotations[idx]?.note || '', ts: new Date().toISOString() }; + const key = promptKey(idx); + annotations[key] = { tier, note: getAnnotation(idx)?.note || '', ts: new Date().toISOString() }; + // Migrate legacy index-keyed annotation if it exists + if (annotations[idx]) delete annotations[idx]; saveAnnotations(); render(); } function saveNote(idx) { const note = document.getElementById('note-input').value; - if (!annotations[idx]) { - annotations[idx] = { tier: dataset[idx].tier || dataset[idx].llm_tier || '?', note: '', ts: new Date().toISOString() }; + const key = promptKey(idx); + if (!annotations[key]) { + annotations[key] = { tier: dataset[idx].tier || dataset[idx].llm_tier || '?', note: '', ts: new Date().toISOString() }; + // Migrate legacy index-keyed annotation + if (annotations[idx]) delete annotations[idx]; } - annotations[idx].note = note; + annotations[key].note = note; saveAnnotations(); render(); } function clearAnnotation(idx) { - delete annotations[idx]; + const key = promptKey(idx); + delete annotations[key]; + delete annotations[idx]; // also clear legacy index key if present saveAnnotations(); render(); } diff --git a/scripts/benchmark_classifier.py b/scripts/benchmark_classifier.py index 443c6e19..bbeace10 100644 --- a/scripts/benchmark_classifier.py +++ b/scripts/benchmark_classifier.py @@ -39,7 +39,10 @@ def classify(prompt): ) with urllib.request.urlopen(req, timeout=30) as resp: data = json.loads(resp.read()) - return data["choices"][0]["message"].get("content", "").strip() + choices = data.get("choices", []) + if not choices: + return "ERROR" + return choices[0].get("message", {}).get("content", "").strip() total = len(dataset.get("prompts", [])) print(f"Benchmark: gemma4-26a4b-routing vs {total} labeled prompts\n") @@ -52,24 +55,30 @@ def classify(prompt): for i, item in enumerate(dataset.get("prompts", [])): prompt = item["prompt"] - expected = item["tier"] - + # Support both old schema ("tier") and new schema ("llm_tier" / "clf_tier") + expected = item.get("tier") or item.get("llm_tier") or item.get("clf_tier", "") + try: predicted = classify(prompt) except Exception as e: predicted = f"ERROR: {str(e)[:50]}" - + results.append({ "prompt": prompt[:100], "expected": expected, "predicted": predicted, }) - + + # Only score against known tiers — skip ERROR/unknown labels gracefully + if expected not in per_tier: + confusion[expected][predicted] += 1 + continue + per_tier[expected]["total"] += 1 if predicted == expected: correct += 1 per_tier[expected]["correct"] += 1 - + confusion[expected][predicted] += 1 # Progress diff --git a/scripts/extract_complex.py b/scripts/extract_complex.py index 96f3acd0..75af8375 100644 --- a/scripts/extract_complex.py +++ b/scripts/extract_complex.py @@ -38,7 +38,11 @@ def fetch_observations(page=1, limit=50): return json.loads(resp.read()) def extract_user_prompt(obs): - """Extract and parse the raw user prompt from the observation input payload.""" + """Extract and parse the FIRST real user prompt from the observation input payload. + + Uses forward iteration (not reversed) to return the first user message, + matching the semantics of extract_prompts.py. Skips pseudo-user system notes. + """ inp = obs.get('input') if not inp: return None if isinstance(inp, str): @@ -47,11 +51,16 @@ def extract_user_prompt(obs): if not isinstance(inp, dict): return None messages = inp.get('messages', []) if not messages: return None - for msg in reversed(messages): + for msg in messages: # forward iteration: first user message if isinstance(msg, dict) and msg.get('role') == 'user': content = msg.get('content', '') - if isinstance(content, str) and len(content.strip()) > 3: - return content.strip() + if not isinstance(content, str) or len(content.strip()) <= 3: + continue + stripped = content.strip() + # Skip Hermes system notes injected as user messages + if stripped.startswith('[System:') or stripped.startswith('[Note:'): + continue + return stripped return None # Keywords suggesting complex/advanced work diff --git a/scripts/extract_gapfill.py b/scripts/extract_gapfill.py index 86a1d643..cdb5f525 100644 --- a/scripts/extract_gapfill.py +++ b/scripts/extract_gapfill.py @@ -36,7 +36,11 @@ def fetch_observations(page=1, limit=50): return json.loads(resp.read()) def extract_user_prompt(obs): - """Extract and parse the raw user prompt from the observation input payload.""" + """Extract and parse the FIRST real user prompt from the observation input payload. + + Uses forward iteration (not reversed) to return the first user message, + matching the semantics of extract_prompts.py. Skips pseudo-user system notes. + """ inp = obs.get('input') if not inp: return None @@ -48,11 +52,16 @@ def extract_user_prompt(obs): messages = inp.get('messages', []) if not messages: return None - for msg in reversed(messages): + for msg in messages: # forward iteration: first user message if isinstance(msg, dict) and msg.get('role') == 'user': content = msg.get('content', '') - if isinstance(content, str) and len(content.strip()) > 3: - return content.strip() + if not isinstance(content, str) or len(content.strip()) <= 3: + continue + stripped = content.strip() + # Skip Hermes system notes injected as user messages + if stripped.startswith('[System:') or stripped.startswith('[Note:'): + continue + return stripped return None def is_trivial(prompt): diff --git a/scripts/reclassify_all.py b/scripts/reclassify_all.py index 02042260..91ea6b61 100644 --- a/scripts/reclassify_all.py +++ b/scripts/reclassify_all.py @@ -28,7 +28,10 @@ def classify(prompt): req = urllib.request.Request(LLAMA_SERVER_URL, data=json.dumps(payload).encode(), headers={'Content-Type':'application/json','Authorization':'Bearer local-token'}) with urllib.request.urlopen(req, timeout=30) as resp: data = json.loads(resp.read()) - return data['choices'][0]['message'].get('content','').strip() + choices = data.get('choices', []) + if not choices: + return f"ERROR: empty response" + return choices[0].get('message', {}).get('content', '').strip() # Load existing dataset (kanban/llm evals) data_dir = Path(__file__).resolve().parent.parent / 'data' diff --git a/scripts/retry_errors.py b/scripts/retry_errors.py index 54ca4967..26724ccc 100644 --- a/scripts/retry_errors.py +++ b/scripts/retry_errors.py @@ -22,7 +22,8 @@ def get_model_port(): data = json.loads(resp.read()) for m in data.get('data', []): if 'gemma4-26a4b' in m.get('id', ''): - args = m.get('status', {}).get('args', []) + status_obj = m.get('status') or {} + args = status_obj.get('args', []) if isinstance(status_obj, dict) else [] for i, v in enumerate(args): if v == '--port' and i + 1 < len(args): return args[i + 1] @@ -67,7 +68,11 @@ def classify(prompt): with open(data_dir / "raw_prompts_hermes.json") as f: all_prompts = json.load(f) -error_indices = [i for i, p in enumerate(dataset.get('prompts', [])) if p.get('tier') == 'ERROR'] +# Schema-aware: support both old schema ("tier") and new reclassify_all.py schema ("clf_tier") +error_indices = [ + i for i, p in enumerate(dataset.get('prompts', [])) + if p.get('tier') == 'ERROR' or p.get('clf_tier') == 'ERROR' +] print(f"Retrying {len(error_indices)} failed prompts (max {MAX_CHARS} chars)...") fixed = 0 @@ -82,7 +87,8 @@ def classify(prompt): tier = classify(prompt) if idx < len(prompts_list): prompts_list[idx]['tier'] = tier - fixed += 1 + if tier != 'ERROR': # only count as fixed if classification succeeded + fixed += 1 except Exception as e: errors += 1 print(f" [{idx}] still failing: {str(e)[:80]}") diff --git a/test_circuit_breaker.py b/test_circuit_breaker.py index 4db46c0a..becb686a 100644 --- a/test_circuit_breaker.py +++ b/test_circuit_breaker.py @@ -35,10 +35,10 @@ def test_initial_state(): """Breaker starts at Tier 0 (open).""" reset_breakers() b = get_breaker() - assert b.is_allowed() == True + assert b.is_allowed() assert b.tier == 0 - assert b.google.is_allowed() == True - assert b.vendor.is_allowed() == True + assert b.google.is_allowed() + assert b.vendor.is_allowed() print("✓ Initial state: Tier 0, allowed") @@ -50,10 +50,10 @@ def test_first_failure_trips_to_tier1(): b.google.record_failure() assert b.google.tier == 1 assert b.google.cooldown_until > time.time() - assert b.google.is_allowed() == False - + assert not b.google.is_allowed() + # Master breaker is still allowed because vendor is allowed (backward compatible fallback) - assert b.is_allowed() == True + assert b.is_allowed() print("✓ 1st failure → Tier 1 (5 min cooldown) on google breaker") @@ -66,9 +66,9 @@ def test_probe_granted_after_cooldown(): b.google.cooldown_until = time.time() - 10 # expired 10s ago b.google.probe_granted = False - assert b.google.is_allowed() == True, "Probe should be granted" + assert b.google.is_allowed(), "Probe should be granted" assert b.google.probe_granted == True, "Probe flag should be set" - assert b.google.is_allowed() == False, "Second call should be denied" + assert not b.google.is_allowed(), "Second call should be denied" print("✓ Probe granted after cooldown expiry, consumed on next check") @@ -83,7 +83,7 @@ def test_probe_failure_advances_tier(): b.google.record_failure() # probe fails assert b.google.tier == 2, f"Expected tier 2, got {b.google.tier}" - assert b.google.probe_granted == False + assert not b.google.probe_granted print("✓ Failed probe at Tier 1 → advanced to Tier 2 (30 min)") @@ -100,7 +100,7 @@ def test_tier3_stays_at_tier3(): assert b.google.tier == MAX_TIER, "Should stay at Tier 3" assert b.google.cooldown_until > old_until, "Cooldown should be renewed" - assert b.google.probe_granted == False + assert not b.google.probe_granted print("✓ Tier 3 failure → stays at Tier 3 (renews 5-hour cooldown)") @@ -115,7 +115,7 @@ def test_success_resets(): b.google.record_success() assert b.google.tier == 0 - assert b.google.is_allowed() == True + assert b.google.is_allowed() print("✓ Success resets breaker to Tier 0 from any tier") @@ -127,12 +127,12 @@ def test_backward_compatibility(): b.record_failure() assert b.google.tier == 1 assert b.vendor.tier == 1 - assert b.is_allowed() == False # both blocked - + assert not b.is_allowed() # both blocked + b.record_success() assert b.google.tier == 0 assert b.vendor.tier == 0 - assert b.is_allowed() == True + assert b.is_allowed() print("✓ Master record_failure and record_success maintain compatibility") diff --git a/verify_breaker.py b/verify_breaker.py index 6f309fcc..9198b583 100644 --- a/verify_breaker.py +++ b/verify_breaker.py @@ -8,16 +8,16 @@ from router.circuit_breaker import get_breaker b = get_breaker() -assert b.is_allowed() == True, 'Tier 0 should be open' +assert b.is_allowed(), 'Tier 0 should be open' for sub in (b.google, b.vendor): - assert sub.is_allowed() == True + assert sub.is_allowed() sub.record_failure() assert sub.tier == 1, 'Should be at Tier 1' - assert sub.is_allowed() == False, 'Tier 1 should block (cooldown active)' + assert not sub.is_allowed(), 'Tier 1 should block (cooldown active)' # Force cooldown expiry sub.cooldown_until = 0 - assert sub.is_allowed() == True, 'Probe should be granted' + assert sub.is_allowed(), 'Probe should be granted' assert sub.probe_granted == True sub.record_failure() # probe fails assert sub.tier == 2, 'Should advance to Tier 2'