", 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
+
+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"
+ 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="Failed to save annotations")
+
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..3a837ae8
--- /dev/null
+++ b/router/static/visualizer.html
@@ -0,0 +1,389 @@
+
+
+
+
+
+Classifier Dataset Visualizer
+
+
+
+
+
📊 Dataset Visualizer
+
+
Total: 0
+
Agree: 0
+
Conflict: 0
+
Reviewed: 0
+
Agreement: -
+
+
+
+
+
+
+
+ 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..bbeace10
--- /dev/null
+++ b/scripts/benchmark_classifier.py
@@ -0,0 +1,140 @@
+"""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())
+ 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")
+
+# 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.get("prompts", [])):
+ prompt = item["prompt"]
+ # 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
+ if (i + 1) % 20 == 0:
+ acc = correct / (i + 1) * 100
+ print(f" {i+1}/{total} — accuracy {acc:.1f}%")
+
+ time.sleep(0.05) # minimal rate-limit (model handles concurrency via llama-server slots)
+
+# Report
+overall = (correct / total * 100) if total > 0 else 0.0
+
+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)
+ count_str = f"{count:3d}"
+ if exp_tier == pred_tier:
+ cell = f" \033[32m{count_str}\033[0m"
+ row += f"{cell:34s}"
+ elif count > 0:
+ cell = f" \033[31m{count_str}\033[0m"
+ row += f"{cell:34s}"
+ else:
+ cell = f" {count_str}"
+ row += f"{cell:25s}"
+ 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/classify_direct.py b/scripts/classify_direct.py
new file mode 100644
index 00000000..ddf8a99a
--- /dev/null
+++ b/scripts/classify_direct.py
@@ -0,0 +1,108 @@
+"""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: """
+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):
+ """Query the llama-server to classify the prompt task complexity."""
+ payload = {
+ "model": "gemma4-26a4b-routing",
+ "messages": [{"role": "user", "content": PROMPT_TEMPLATE + prompt}],
+ "temperature": 0.0,
+ "max_tokens": 15,
+ }
+ 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()
+
+# 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:
+ 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
+ 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/extract_complex.py b/scripts/extract_complex.py
new file mode 100644
index 00000000..75af8375
--- /dev/null
+++ b/scripts/extract_complex.py
@@ -0,0 +1,138 @@
+"""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"
+
+existing = set()
+dataset_path = Path(__file__).resolve().parent.parent / "data" / "classified_dataset.json"
+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")
+
+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}")
+ with urllib.request.urlopen(req, timeout=15) as resp:
+ return json.loads(resp.read())
+
+def extract_user_prompt(obs):
+ """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):
+ 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 messages: # forward iteration: first user message
+ if isinstance(msg, dict) and msg.get('role') == 'user':
+ content = msg.get('content', '')
+ 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
+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]
+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
new file mode 100644
index 00000000..cdb5f525
--- /dev/null
+++ b/scripts/extract_gapfill.py
@@ -0,0 +1,150 @@
+"""Gap-fill extraction: pull longer/older prompts targeting complex+ tiers."""
+import os, base64, json, urllib.request, time, re
+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):
+ """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):
+ 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 messages: # forward iteration: first user message
+ if isinstance(msg, dict) and msg.get('role') == 'user':
+ content = msg.get('content', '')
+ 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):
+ """Check if the prompt matches a list of trivial test patterns to filter out."""
+ 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 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+)...")
+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]
+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/extract_prompts.py b/scripts/extract_prompts.py
index 8c3e454d..028baa8d 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,34 +51,46 @@ 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
if isinstance(inp, str):
try:
inp = json.loads(inp)
- except:
+ except Exception:
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()
+
+ 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]}...")
diff --git a/scripts/reclassify_all.py b/scripts/reclassify_all.py
new file mode 100644
index 00000000..91ea6b61
--- /dev/null
+++ b/scripts/reclassify_all.py
@@ -0,0 +1,111 @@
+"""Re-run gemma4 classifier (with grammar) on all dataset prompts via router."""
+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']
+
+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
+- 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):
+ """Query the llama-server to classify the prompt complexity with grammar enforcement."""
+ 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(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())
+ 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'
+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('llm_tier') or 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'])
+
+total_results = len(results)
+print(f"\n{'='*60}")
+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 atomically
+combined = {
+ '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,
+}
+
+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
+
+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
new file mode 100644
index 00000000..26724ccc
--- /dev/null
+++ b/scripts/retry_errors.py
@@ -0,0 +1,111 @@
+"""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', ''):
+ 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]
+ 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):
+ """Query the direct model port to classify the prompt complexity, handling truncations."""
+ 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())
+ 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:
+ return tier
+ return "ERROR"
+
+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)
+
+# 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
+errors = 0
+
+for batch_start in range(0, len(error_indices), 5):
+ batch = error_indices[batch_start:batch_start + 5]
+ for idx in batch:
+ prompts_list = dataset.get('prompts', [])
+ prompt = prompts_list[idx].get('prompt') if idx < len(prompts_list) else ""
+ try:
+ tier = classify(prompt)
+ if idx < len(prompts_list):
+ prompts_list[idx]['tier'] = tier
+ 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]}")
+ 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.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]
+
+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
diff --git a/test_a2_verify.py b/test_a2_verify.py
index 13c872c0..42cde1e1 100644
--- a/test_a2_verify.py
+++ b/test_a2_verify.py
@@ -1,16 +1,18 @@
#!/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
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..becb686a 100644
--- a/test_circuit_breaker.py
+++ b/test_circuit_breaker.py
@@ -1,135 +1,176 @@
#!/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
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
+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.is_allowed()
assert b.tier == 0
- print("✓ Initial state: Tier 0, agy allowed")
+ assert b.google.is_allowed()
+ assert b.vendor.is_allowed()
+ 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 not b.google.is_allowed()
+
+ # Master breaker is still allowed because vendor is allowed (backward compatible fallback)
+ assert b.is_allowed()
+ 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(), "Probe should be granted"
+ assert b.google.probe_granted == True, "Probe flag should be set"
+ assert not b.google.is_allowed(), "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 not b.google.probe_granted
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 not b.google.probe_granted
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.record_success()
- assert b.tier == 0
- assert b.is_allowed() == True
+
+ 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()
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 not b.is_allowed() # both blocked
+
+ b.record_success()
+ assert b.google.tier == 0
+ assert b.vendor.tier == 0
+ assert b.is_allowed()
+ 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 +182,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..9198b583 100644
--- a/verify_breaker.py
+++ b/verify_breaker.py
@@ -1,17 +1,26 @@
#!/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()
-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'
+assert b.is_allowed(), 'Tier 0 should be open'
+
+for sub in (b.google, b.vendor):
+ assert sub.is_allowed()
+ sub.record_failure()
+ assert sub.tier == 1, 'Should be at Tier 1'
+ assert not sub.is_allowed(), 'Tier 1 should block (cooldown active)'
+ # Force cooldown expiry
+ sub.cooldown_until = 0
+ 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'
+
+assert b.tier == 2
print('All assertions passed')