"""MiniWeb Task Annotation Interface. Two modes: 1. Review Sites — browse each site, leave free-form feedback 2. Annotate Tasks — system samples N sites × M macros, annotator designs task The annotation blueprint is registered in the main MiniWeb app. Access at http://localhost:8080/annotate/ """ from helpers.paths import REPO_ROOT import json import os import random import uuid import pickle from datetime import datetime from pathlib import Path from flask import Blueprint, jsonify, redirect, render_template, request, session, url_for from miniweb.annotation.macro_locations import MACRO_LOCATIONS from miniweb.annotation import macros as _registry from miniweb.annotation import final_set as _final_set from miniweb.annotation.storage import ANNOTATIONS_DIR from miniweb.app import demo as _demo PROJECT_ROOT = REPO_ROOT SITES_DIR = PROJECT_ROOT / "src" / "miniweb" / "sites" # Annotator credentials: override via MINIWEB_ANNOTATORS env var # Format: "user1:pass1,user2:pass2,..." _DEFAULT_ANNOTATORS = {"annotator1": "miniweb", "annotator2": "miniweb", "annotator3": "miniweb", "annotator4": "miniweb"} def _load_annotator_credentials(): env = os.environ.get("MINIWEB_ANNOTATORS", "") if env: creds = {} for pair in env.split(","): if ":" in pair: user, pw = pair.split(":", 1) creds[user.strip()] = pw.strip() return creds if creds else _DEFAULT_ANNOTATORS return _DEFAULT_ANNOTATORS ANNOTATOR_CREDENTIALS = _load_annotator_credentials() annotation_bp = Blueprint( "annotation", __name__, template_folder="templates", static_folder="static", url_prefix="/annotate", ) # --------------------------------------------------------------------------- # Macro descriptions — verb + modality + explanation + example # --------------------------------------------------------------------------- # Derived from the canonical registry (data/macros.yaml) via src/miniweb/annotation/macros.py. # Retired-name entries are gone — resolve any name via _canon()/_registry.describe(). _MACRO_DESCRIPTIONS = _registry.descriptions() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _save_task(task): """Save task + trajectory to the shared annotations directory.""" from miniweb.annotation.storage import save_task as fs_save, generate_task_id task_id = task.get("task_id") or generate_task_id() task["task_id"] = task_id # annotator becomes a directory name and a URL path segment — strip # slashes/backslashes (a stray trailing backslash once forked the dataset in two) annotator = task.get("annotator", "anonymous").strip().strip("\\/") or "anonymous" task["annotator"] = annotator trajectory = task.pop("trajectory", []) # Primary storage: file system fs_save(annotator, task_id, task, trajectory) return task_id def _load_sites(): from miniweb.annotation.storage import list_tasks sites = [] # Count annotated tasks per site from file storage task_counts = {} for t in list_tasks(): for s in t.get("sites", []): sid = s["id"] if isinstance(s, dict) else s task_counts[sid] = task_counts.get(sid, 0) + 1 for site_json in sorted(SITES_DIR.glob("*/site.json")): if site_json.parent.name.startswith("_"): continue routes = site_json.parent / "routes.py" if not routes.exists() or routes.stat().st_size < 500: continue meta = json.loads(site_json.read_text()) meta["url"] = f"/sites/{meta['id']}/" meta["annotated_count"] = task_counts.get(meta["id"], 0) meta["review_count"] = 0 # Human-readable site type (specific kind of site, from its id), for # "Name (Type)" labels — e.g. "Tax Filing, DMV & Permits". Plus the broad # category for reference. from miniweb.app import SITE_CATEGORIES, site_type meta["category"] = SITE_CATEGORIES.get(meta["id"], "Other") meta["type"] = site_type(meta["id"]) sites.append(meta) return sites def _load_macros(): """Collect all unique (canonical) macros across all sites from MACRO_LOCATIONS.""" macros = set() for site_macros in MACRO_LOCATIONS.values(): macros.update(_canon(m) for m in site_macros.keys()) return sorted(macros) def _load_site_macros(site_id): """Load supported macros for a site. MACRO_LOCATIONS is the audited source of truth — it lists only macros whose UI elements have been verified to exist on each site. Falls back to README, then tasks.json for sites not yet audited. """ # MACRO_LOCATIONS is the audited source of truth if site_id in MACRO_LOCATIONS: macros = [_canon(m) for m in MACRO_LOCATIONS[site_id].keys()] if macros: return sorted(set(macros)) # Fallback to README spec for sites not yet in MACRO_LOCATIONS import re as _re readme = SITES_DIR / site_id / "doc" / "README.md" if readme.exists(): try: text = readme.read_text() # Try comma-separated format match = _re.search( r"## Target Macros[^\n]*\n\s*\n?([\w_,\s]+?)(?:\n\n|\n##|\Z)", text, _re.DOTALL, ) if match: macros = [m.strip() for m in match.group(1).split(",") if m.strip()] if macros: return sorted(set(macros)) # Try numbered list format: "1. **macro_name** -- ..." list_macros = _re.findall( r"\d+\.\s+\*\*(\w+)\*\*", text[text.find("## Target Macros"):] if "## Target Macros" in text else "" ) if list_macros: return sorted(set(list_macros)) except OSError: pass # Fallback to tasks.json tasks_file = SITES_DIR / site_id / "tasks.json" macros = set() if tasks_file.exists(): try: for t in json.loads(tasks_file.read_text()): for m in t.get("macros", []): macros.add(m) except (json.JSONDecodeError, OSError): pass return sorted(macros) _NA_FILE = ANNOTATIONS_DIR / "na_reports.json" # Skip-modal reports (reason + details) — same download-and-collect workflow # as na_reports.json _SKIP_REPORTS_FILE = ANNOTATIONS_DIR / "skip_reports.json" def _load_na_reports(): """Load NA reports from disk.""" if _NA_FILE.exists(): try: return json.loads(_NA_FILE.read_text()) except (json.JSONDecodeError, OSError): pass return [] def _save_na_report(sites, macro, annotator): """Append a single NA report.""" reports = _load_na_reports() reports.append({ "sites": sites, "macro": macro, "annotator": annotator, "timestamp": datetime.now().isoformat(), }) _NA_FILE.write_text(json.dumps(reports, indent=2)) def _get_na_macros(site_ids): """Get macros marked N/A for these sites (threshold: 2+ reports).""" from collections import Counter reports = _load_na_reports() site_set = set(site_ids) counts = Counter() for r in reports: if site_set & set(r.get("sites", [])): counts[r["macro"]] += 1 return {m for m, c in counts.items() if c >= 2} def _get_macro_coverage(): """Count how many annotated tasks cover each macro (from file storage).""" from miniweb.annotation.storage import list_tasks coverage = {} for t in list_tasks(): for m in t.get("macros", []): m = _canon(m) coverage[m] = coverage.get(m, 0) + 1 return coverage def _task_site_ids(task): """Normalize a stored task's sites to a list of site-id strings.""" ids = [] for s in task.get("sites", []): ids.append(s.get("id") if isinstance(s, dict) else s) return ids # Coverage floor: each macro should be demonstrated on K distinct sites via # single-site tasks (or on all its sites, when fewer than K support it). # K is adjustable from the coverage page; persisted in floor_config.json. _FLOOR_CONFIG_FILE = Path(__file__).parent / "floor_config.json" _DEFAULT_FLOOR_K = 3 def _get_floor_k(): try: k = int(json.loads(_FLOOR_CONFIG_FILE.read_text()).get("k", _DEFAULT_FLOOR_K)) return max(1, min(10, k)) except (OSError, ValueError, json.JSONDecodeError): return _DEFAULT_FLOOR_K # Retired macro names (merged 2026-07-08) -> canonical. Applied when READING # task files so historical annotations keep counting; task files themselves # are never rewritten. (filter_by_radio is site-scoped and NOT aliased — # it remains a valid macro on other sites.) _MACRO_ALIASES = _registry.alias_map() # Canonical-name resolver (alias chain with cycle guard) — from the registry. _canon = _registry.canon _CANONICAL_LOCATIONS = None def _canonical_macro_locations(): """MACRO_LOCATIONS re-keyed by canonical macro name, location lists aggregated across any alias keys that merge into the same macro. The UI samples canonical names, so lookups must be keyed the same way.""" global _CANONICAL_LOCATIONS if _CANONICAL_LOCATIONS is None: out = {} for site, macros in MACRO_LOCATIONS.items(): site_out = {} for m, locs in macros.items(): locs = [locs] if isinstance(locs, str) else list(locs or []) site_out.setdefault(_canon(m), []).extend(locs) out[site] = site_out _CANONICAL_LOCATIONS = out return _CANONICAL_LOCATIONS def _site_macro_pool(site_id): """Macros eligible for floor coverage on a site (excl. NA + navigation).""" pool = set(_load_site_macros(site_id)) - _get_na_macros([site_id]) pool.discard("navigate_by_route") return pool def _get_macro_site_coverage(): """macro -> set of site ids where a SINGLE-SITE task covers it.""" from miniweb.annotation.storage import list_tasks cov = {} for t in list_tasks(): ids = _task_site_ids(t) if len(ids) != 1: continue for m in t.get("macros", []): cov.setdefault(_canon(m), set()).add(ids[0]) return cov def _macro_k_targets(): """macro -> min(configured K, number of sites supporting it).""" k = _get_floor_k() supporting = {} for sid in MACRO_LOCATIONS: for m in _site_macro_pool(sid): supporting[m] = supporting.get(m, 0) + 1 return {m: min(k, n) for m, n in supporting.items()} def _get_cell_counts(): """Count existing tasks per (num_sites, num_macros) cell (from file storage).""" from miniweb.annotation.storage import list_tasks counts = {} for t in list_tasks(): n_sites = len(t.get("sites", [])) n_macros = len({_canon(m) for m in t.get("macros", [])}) key = (n_sites, n_macros) counts[key] = counts.get(key, 0) + 1 return counts def _load_graph_edges(): """Load cross-site edges from proposed_edges.csv, grouped by source.""" import csv from collections import defaultdict outgoing = defaultdict(list) incoming = defaultdict(list) edges_file = Path(__file__).parent / "proposed_edges.csv" if edges_file.exists(): with open(edges_file) as f: for row in csv.DictReader(f): if row.get("status") != "approved": continue src = row.get("source", "") tgt = row.get("target", "") evt = row.get("event_type", "") if src and tgt: outgoing[src].append((tgt, evt)) incoming[tgt].append((src, evt)) return outgoing, incoming # Edge types that imply specific macros the task should exercise # Edge-type -> hint macros to boost when that cross-site edge is present. # Canonical names only: the consumer matches these against the canonical macro # pool, so a retired name here would silently never fire. _EDGE_MACRO_HINTS = { "purchase": ["checkout_by_form", "add_by_button"], "payment": ["pay_by_form", "pay_by_dropdown"], "booking": ["book_by_form", "book_by_date_range"], "signup": ["submit_form", "authenticate_by_form"], "credential": ["authenticate_by_form"], "notification": ["extract_by_route"], "file_created": ["submit_form", "upload_by_upload"], "message": ["message_from_free_text"], "translate": ["translate_by_query", "compute_by_dropdown"], "share": ["share_by_dropdown", "share_by_toggle"], "crosspost": ["submit_form"], "shorten": ["submit_form"], "location": ["navigate_by_route", "search_by_query"], "directions": ["route_by_query"], "deadline": ["extract_by_date_range"], "query": ["search_by_query"], } def _generate_prompt(sites, coverage, force_single=False, n_sites_override=None): """Site-first prompt sampler. Strategy: 1. Pick the site with the most uncovered macros 2. Pick chain length using ratio 3:2:3:2 for lengths 1:2:3:4+ (weighted toward lengths the site still needs) 3. Pick uncovered macros for that site 4. For cross-site: expand along graph edges """ rng = random.Random() site_pool = list(sites) if not site_pool: return None site_map = {s["id"]: s for s in site_pool} # Load graph outgoing, incoming = _load_graph_edges() # --- Per-site coverage analysis --- from miniweb.annotation.storage import list_tasks all_tasks = list_tasks() # Count which macros are covered per site (by any task) site_macro_covered = {} # site_id -> set of covered macros site_chain_counts = {} # site_id -> {chain_len: count} for t in all_tasks: t_sites = [s["id"] if isinstance(s, dict) else s for s in t.get("sites", [])] t_macros = t.get("macros", []) chain_len = len(t_macros) for sid in t_sites: site_macro_covered.setdefault(sid, set()).update(t_macros) site_chain_counts.setdefault(sid, {}) site_chain_counts[sid][chain_len] = site_chain_counts[sid].get(chain_len, 0) + 1 # For each site, find uncovered macros site_uncovered = {} for s in site_pool: sid = s["id"] all_macros = set(_load_site_macros(sid)) covered = site_macro_covered.get(sid, set()) uncovered = all_macros - covered site_uncovered[sid] = uncovered # --- 1. Pick site: prefer sites whose uncovered macros are globally RARE (not # just numerous) — a site with 2 uncovered rare macros is more valuable for # global balance than one with 8 uncovered ubiquitous ones. --- site_weights = [] for s in site_pool: sid = s["id"] uncov = site_uncovered.get(sid, set()) if uncov: site_weights.append(sum(1.0 / (coverage.get(m, 0) + 1) for m in uncov)) else: site_weights.append(0.1) # small weight for fully covered sites total = sum(site_weights) norm_weights = [w / total for w in site_weights] seed = rng.choices(site_pool, weights=norm_weights, k=1)[0] sampled_sites = [seed] sampled_ids = {seed["id"]} edge_types_used = [] # --- 2. Pick chain length using ratio 3:2:3:2 for lengths 1:2:3:4 --- # Weight toward lengths the site still needs chain_counts = site_chain_counts.get(seed["id"], {}) # Target ratio for lengths 1..6. Still favors shorter chains, but includes the # longer 5- and 6-macro chains (weighted by scarcity: each length's target is # divided by its current count below, so under-filled lengths get priority). target_ratio = {1: 3, 2: 2, 3: 3, 4: 2, 5: 1, 6: 1} chain_weights = [] chain_options = [] for length, target_w in target_ratio.items(): if force_single and length > 3: continue current = chain_counts.get(length, 0) # Weight: target ratio / (current count + 1) — under-filled lengths get priority w = target_w / (current + 1) chain_weights.append(w) chain_options.append(length) if not chain_options: chain_options = [1] chain_weights = [1.0] total = sum(chain_weights) chain_weights = [w / total for w in chain_weights] n_macros = rng.choices(chain_options, weights=chain_weights, k=1)[0] # For single-site mode, cap at 3 if force_single: n_sites = 1 n_macros = min(n_macros, 3) elif n_sites_override: # Annotator explicitly chose the task's site count (1, 2, or 3). n_sites = max(1, min(3, int(n_sites_override))) if n_macros < n_sites: n_macros = n_sites else: # Longer chains often span sites, so make them progressively more likely to # be multi-site (single-site stays the majority). The longest chains may # reach 3 sites; shorter ones cap at 2. n_sites = 1 if n_macros >= 3: p_cross = min(0.6, 0.2 + 0.08 * (n_macros - 3)) # 3→.20 … 6→.44 if rng.random() < p_cross: n_sites = 2 if n_macros >= 5 and rng.random() < 0.35: n_sites = 3 if not force_single and n_sites >= 2: # Cross-site task — expand along graph edges from the seed to n_sites sites. candidates = [] # Direct outgoing for target, etype in outgoing.get(seed["id"], []): if target in site_map and target not in sampled_ids: # Weight: prefer under-annotated targets + non-hub sites t = site_map[target] hub_penalty = 0.3 if target in ("password-managers", "email") else 1.0 w = hub_penalty / (t.get("annotated_count", 0) + 1) candidates.append((t, w, etype, "outgoing")) # Direct incoming for source, etype in incoming.get(seed["id"], []): if source in site_map and source not in sampled_ids: s = site_map[source] hub_penalty = 0.3 if source in ("password-managers", "email") else 1.0 w = hub_penalty / (s.get("annotated_count", 0) + 1) candidates.append((s, w, etype, "incoming")) # Hub-mediated (seed and other site share a common target/source) if len(candidates) < 3: seed_targets = {t for t, _ in outgoing.get(seed["id"], [])} seed_sources = {s for s, _ in incoming.get(seed["id"], [])} hubs = seed_targets | seed_sources for hub in hubs: # Other sites that also connect to this hub for other, etype in outgoing.get(hub, []): if other in site_map and other not in sampled_ids and other != seed["id"]: o = site_map[other] w = 0.5 / (o.get("annotated_count", 0) + 1) candidates.append((o, w, f"via_{hub}", "hub")) for other, etype in incoming.get(hub, []): if other in site_map and other not in sampled_ids and other != seed["id"]: o = site_map[other] w = 0.5 / (o.get("annotated_count", 0) + 1) candidates.append((o, w, f"via_{hub}", "hub")) # Deduplicate candidates (keep highest weight per site) best = {} for site_obj, w, etype, direction in candidates: sid = site_obj["id"] if sid not in best or w > best[sid][1]: best[sid] = (site_obj, w, etype, direction) candidates = list(best.values()) # Pick additional sites from candidates for _ in range(n_sites - 1): if not candidates: # Fallback: random under-covered sites, filling all the way to the # requested count (previously stopped after one, so a forced # 3-site prompt could come back with only 2 sites). remaining = [s for s in site_pool if s["id"] not in sampled_ids] while remaining and len(sampled_sites) < n_sites: rw = [1.0 / (s.get("annotated_count", 0) + 1) for s in remaining] t = sum(rw) pick = rng.choices(remaining, weights=[w / t for w in rw], k=1)[0] sampled_sites.append(pick) sampled_ids.add(pick["id"]) remaining = [s for s in remaining if s["id"] != pick["id"]] break cs, ws, etypes, dirs = zip(*candidates) total = sum(ws) pick_idx = rng.choices(range(len(cs)), weights=[w/total for w in ws], k=1)[0] pick = cs[pick_idx] sampled_sites.append(pick) sampled_ids.add(pick["id"]) edge_types_used.append(etypes[pick_idx]) candidates = [(s, w, e, d) for s, w, e, d in candidates if s["id"] != pick["id"]] # 2b. Auto-include companion sites required for 2FA flows (skip in single-site mode): # - password-managers requires instant-messaging (SMS verification codes) # - any transaction site (banking edges) requires email (2FA confirmation) required_companions = {} if not force_single and "password-managers" in sampled_ids: required_companions["instant-messaging"] = "2fa_code" if not force_single: # Check if any sampled site sends payments/purchases to banking for sid in list(sampled_ids): for target, etype in outgoing.get(sid, []): if target == "banking" and etype in ("payment", "purchase", "trade"): required_companions["email"] = "2fa_verification" break # Also if banking itself is sampled (transactions need email 2FA) if "banking" in sampled_ids: required_companions["email"] = "2fa_verification" for companion_id, reason in required_companions.items(): if len(sampled_sites) >= (n_sites_override or 3): break # honor the annotator-chosen site count (default cap: 3) if companion_id not in sampled_ids and companion_id in site_map: sampled_sites.append(site_map[companion_id]) sampled_ids.add(companion_id) edge_types_used.append(reason) # Update n_sites to reflect added companions n_sites = len(sampled_sites) # Ensure n_macros >= n_sites if n_macros < n_sites: n_macros = n_sites # 3. Pick macros — combine site macros + edge-implied macros macro_pool = set() for s in sampled_sites: macro_pool.update(_load_site_macros(s["id"])) # Remove macros previously marked as N/A for these sites na_macros = _get_na_macros([s["id"] for s in sampled_sites]) macro_pool -= na_macros # Add edge-implied macros — only if they exist in the sampled sites' macro pool edge_macros = [] for etype in edge_types_used: etype_clean = etype.split("via_")[0] if etype.startswith("via_") else etype hints = _EDGE_MACRO_HINTS.get(etype_clean, []) for h in hints: if h in macro_pool: edge_macros.append(h) if not macro_pool: macro_pool = set(_load_macros()) macro_pool = sorted(macro_pool) if not macro_pool: return None # Weight: prefer macros that are uncovered on this site AND globally rare. The # uncovered boost is scaled by GLOBAL scarcity (1/(cov+1)), so a ubiquitous macro # that's merely uncovered on this one site (e.g. search at cov 51) doesn't out- # weigh a globally-rare macro (e.g. edit_by_cell at cov 0) — otherwise the common, # every-site macros dominate the corpus. site_uncov = site_uncovered.get(seed["id"], set()) edge_macro_set = set(edge_macros) macro_weights = [] for m in macro_pool: scarcity = 1.0 / (coverage.get(m, 0) + 1) w = (10.0 * scarcity) if m in site_uncov else scarcity if m in edge_macro_set: w *= 3.0 # boost edge-implied macros macro_weights.append(w) total = sum(macro_weights) macro_weights = [w / total for w in macro_weights] _QA_VERBS = {"extract", "compute", "count", "compare", "verify", "calculate", "rank", "list"} sampled_macros = [] has_qa = False remaining_macros = list(zip(macro_pool, macro_weights)) for _ in range(min(n_macros, len(remaining_macros))): if not remaining_macros: break # If we already have a QA macro, exclude other QA macros from candidates if has_qa: remaining_macros = [(m, w) for m, w in remaining_macros if m.split("_")[0] not in _QA_VERBS] if not remaining_macros: break ms, ws = zip(*remaining_macros) total = sum(ws) ws = [w / total for w in ws] pick = rng.choices(ms, weights=ws, k=1)[0] sampled_macros.append(pick) if pick.split("_")[0] in _QA_VERBS: has_qa = True remaining_macros = [(m, w) for m, w in remaining_macros if m != pick] # 4. Ensure every sampled site has at least 1 sampled macro it can perform sampled_macro_set = set(sampled_macros) for s in sampled_sites: site_macros = set(_load_site_macros(s["id"])) - na_macros if not (site_macros & sampled_macro_set): # This site has no overlap — add one of its macros site_only = sorted(site_macros - sampled_macro_set) if site_only: # Prefer under-covered macros from this site pick = min(site_only, key=lambda m: coverage.get(m, 0)) sampled_macros.append(pick) sampled_macro_set.add(pick) return { "sites": [{"id": s["id"], "name": s.get("name", s["id"]), "category": s.get("category", "Other"), "type": s.get("type", "")} for s in sampled_sites], "macros": sampled_macros, "num_sites": len(sampled_sites), "num_macros": len(sampled_macros), "edges": edge_types_used, } def _generate_site_prompt(site, rng=None): """Sampler for an annotator-chosen site. The annotator picks the site. This function picks: 1. Chain length using ratio 3:2:3:2 for lengths 1:2:3:4 (weighted toward lengths the site still needs) 2. Macros — strongly prefer uncovered macros on this site, all macros weighted equally (no difficulty bias) """ rng = rng or random.Random() site_id = site["id"] all_macros = sorted(set(_load_site_macros(site_id))) if not all_macros: return None # Get per-site coverage from miniweb.annotation.storage import list_tasks site_tasks = [t for t in list_tasks() if site_id in [s["id"] if isinstance(s, dict) else s for s in t.get("sites", [])]] # Which macros are already covered on this site? covered = set() chain_counts = {} # chain_len -> count for t in site_tasks: covered.update(t.get("macros", [])) cl = len(t.get("macros", [])) chain_counts[cl] = chain_counts.get(cl, 0) + 1 uncovered = [m for m in all_macros if m not in covered] # 1. Pick chain length: ratio for lengths 1..6 (favors shorter, includes 5/6) target_ratio = {1: 3, 2: 2, 3: 3, 4: 2, 5: 1, 6: 1} chain_options = [] chain_weights = [] for length, target_w in target_ratio.items(): current = chain_counts.get(length, 0) w = target_w / (current + 1) chain_options.append(length) chain_weights.append(w) total = sum(chain_weights) chain_weights = [w / total for w in chain_weights] n_macros = rng.choices(chain_options, weights=chain_weights, k=1)[0] # 2. Pick macros — strongly prefer uncovered, equal weight otherwise _QA_VERBS = {"extract", "compute", "count", "compare", "verify", "calculate", "rank", "list"} sampled = [] remaining = list(all_macros) has_qa = False for _ in range(min(n_macros, len(remaining))): if has_qa: remaining = [m for m in remaining if m.split("_")[0] not in _QA_VERBS] if not remaining: break ws = [] for m in remaining: if m in uncovered: ws.append(10.0) # uncovered — strong priority else: site_count = sum(1 for t in site_tasks if m in t.get("macros", [])) ws.append(1.0 / (site_count + 1)) total = sum(ws) ws = [w / total for w in ws] pick = rng.choices(remaining, weights=ws, k=1)[0] sampled.append(pick) if pick.split("_")[0] in _QA_VERBS: has_qa = True remaining = [m for m in remaining if m != pick] # Compute how many tasks still needed per chain length (lengths 1..6) total_macros = len(all_macros) import math batches = max(1, math.ceil(total_macros / 24)) targets = {1: 3 * batches, 2: 2 * batches, 3: 3 * batches, 4: 2 * batches, 5: 1 * batches, 6: 1 * batches} chain_remaining = {k: max(0, targets[k] - chain_counts.get(k, 0)) for k in targets} return { "sites": [{"id": site_id, "name": site.get("name", site_id), "category": site.get("category", "Other"), "type": site.get("type", "")}], "macros": sampled, "num_sites": 1, "num_macros": len(sampled), "edges": [], "chosen_site": True, "total_macros": total_macros, "uncovered_macros": len(uncovered), "chain_counts": chain_counts, "chain_remaining": chain_remaining, } # --------------------------------------------------------------------------- # Auth # --------------------------------------------------------------------------- _AUTH_EXEMPT = {"/annotate/login", "/annotate/api/auto_login", "/annotate/api/auto_logout"} # The public demo (MINIWEB_DEMO=1, app/demo.py): no login, the released tasks prefilled, nothing saved. These POSTs # only touch the visitor's own session (site logins, 2FA) or read (macro search, verifier runs); every other write # is refused. The pages and APIs of the audits, the site reviews, the accounts and the macro graph are hidden, and # so are the reviewers' tools (the Verifier Builder, Task Review, and the playback facts, which give answers away). _DEMO_WRITES = {"/annotate/api/auto_login", "/annotate/api/auto_logout", "/annotate/api/toggle_2fa", "/annotate/api/macro_search", "/annotate/api/run_task_verifier", "/annotate/api/run_verifiers"} _DEMO_HIDDEN = ("/annotate/accounts", "/annotate/grader-audit", "/annotate/task-audit", "/annotate/review", "/annotate/graph", "/annotate/api/grader_audit/", "/annotate/api/task_audit/", "/annotate/api/review", "/annotate/api/graph/", "/annotate/verify", "/annotate/task-review", "/annotate/api/task_review/", "/annotate/playback", "/annotate/api/playback_facts") TRAJECTORIES_URL = "https://minhphd.github.io/miniweb/trajectories.html" # every agent episode of the paper _DEMO_BANNER = ( '
' 'Read-only demo: explore freely, nothing you change is saved
' '' # the editor asks for the annotator's name before it opens (it runs at DOMContentLoaded, after this) "") def _demo_gate(): session["annotator_authenticated"] = True session.setdefault("annotator_name", "guest") path = request.path if path in ("/annotate/login", "/annotate/logout") or path.startswith(_DEMO_HIDDEN): if path.startswith("/annotate/api/"): return jsonify({"error": "Not available in the demo"}), 404 return redirect(url_for("annotation.index")) if request.method not in ("GET", "HEAD", "OPTIONS") and path not in _DEMO_WRITES: return jsonify({"error": "This is a read-only demo: nothing you change is saved."}), 403 return None @annotation_bp.before_request def _require_annotator_login(): if _demo.ENABLED: return _demo_gate() if request.path in _AUTH_EXEMPT: return None if session.get("annotator_authenticated"): if (request.method not in ("GET", "HEAD", "OPTIONS") and _final_set.locked() and not _final_set.write_allowed(request.path)): return jsonify({"error": "The annotation tool is locked to the paper's final set " "(data/final_set.yaml); edits are disabled."}), 403 return None # API routes get JSON 401, HTML routes get redirect if request.path.startswith("/annotate/api/"): return jsonify({"error": "Not authenticated"}), 401 return redirect(url_for("annotation.annotator_login")) @annotation_bp.after_request def _final_set_banner(resp): """Mark every annotation page as read-only while the final set is locked (not API responses, such as a recorded page the Task Audit shows in a frame).""" if ((_final_set.locked() or _demo.ENABLED) and resp.mimetype == "text/html" and resp.status_code == 200 and not resp.direct_passthrough and "/api/" not in request.path): html = resp.get_data(as_text=True) if "" in html: banner = _DEMO_BANNER if _demo.ENABLED else _final_set.banner() resp.set_data(html.replace("", banner + "", 1)) return resp @annotation_bp.route("/login", methods=["GET", "POST"]) def annotator_login(): error = None if request.method == "POST": username = request.form.get("username", "").strip() password = request.form.get("password", "").strip() if username in ANNOTATOR_CREDENTIALS and ANNOTATOR_CREDENTIALS[username] == password: session["annotator_authenticated"] = True session["annotator_name"] = username return redirect(url_for("annotation.index")) error = "Invalid credentials" return render_template("annotate_login.html", error=error) @annotation_bp.route("/logout") def annotator_logout(): session.pop("annotator_authenticated", None) session.pop("annotator_name", None) return redirect(url_for("annotation.annotator_login")) # --------------------------------------------------------------------------- # Routes # --------------------------------------------------------------------------- @annotation_bp.route("/") def index(): from miniweb.annotation.storage import get_stats sites = _load_sites() ann_stats = get_stats() macros = _load_macros() coverage = _get_macro_coverage() cell_counts = _get_cell_counts() # per-macro task counts, least-covered first — surfaced on the dashboard so # annotators see gaps at a glance (replaces the standalone Coverage page) macro_counts = sorted(((m, coverage.get(m, 0)) for m in macros), key=lambda x: (x[1], x[0])) return render_template("index.html", sites=sites, task_count=ann_stats["total_tasks"], macros=macros, coverage=coverage, cell_counts=cell_counts, macro_counts=macro_counts, demo=_demo.ENABLED, trajectories_url=TRAJECTORIES_URL) @annotation_bp.route("/task") def annotate(): """Single annotation interface — system samples sites × macros. With ?rerecord=[&annotator=] the prompt is built from that saved task instead of sampled, and the task's design data is shipped to the template so the editor opens prefilled; saving then replaces the original recording (see api_create_task's replace branch). """ sites = _load_sites() rerecord = None rerecord_id = request.args.get("rerecord", "").strip() if _demo.ENABLED and not rerecord_id: # the demo opens a released task, filled in, rather than an empty prompt from miniweb.annotation.storage import list_tasks tasks = [t for t in list_tasks() if not _final_set.locked() or t.get("task_id") in _final_set.tasks()] if tasks: t = random.choice(tasks) return redirect(url_for("annotation.annotate", rerecord=t["task_id"], annotator=t.get("annotator", ""))) if rerecord_id: from miniweb.annotation.storage import get_annotators, load_task annotator = request.args.get("annotator", "").strip() candidates = [annotator] if annotator else get_annotators() task = None for ann in candidates: task = load_task(ann, rerecord_id) if task: annotator = ann break if not task: return redirect(url_for("annotation.verify")) site_map = {s["id"]: s for s in sites} task_site_ids = [s["id"] if isinstance(s, dict) else s for s in task.get("sites", [])] prompt = { "sites": [{"id": sid, "name": site_map.get(sid, {}).get("name", sid), "category": site_map.get(sid, {}).get("category", "Other"), "type": site_map.get(sid, {}).get("type", "")} for sid in task_site_ids], "macros": task.get("macros", []), # instance ids for duplicate-macro tasks; absent/empty for the common # case (the client then derives ids 1:1 from the base names) "macro_instances": task.get("macro_instances", []), "num_sites": len(task_site_ids), "num_macros": len(task.get("macros", [])), "edges": [], } rerecord = { "task_id": rerecord_id, "annotator": annotator, "instruction": task.get("instruction", ""), "expected_answer": task.get("expected_answer") or "", "answer_type": task.get("answer_type", "string"), "alternatives": task.get("alternatives", ""), "expected_outcome": task.get("expected_outcome", ""), "macro_edges": task.get("macro_edges", []), "macro_positions": task.get("macro_positions", {}), "macro_subtasks": task.get("macro_subtasks", {}), "macro_operations": task.get("macro_operations", {}), "qa_answers": task.get("qa_answers", {}), "starting_url": task.get("starting_url", ""), "requires_login": task.get("requires_login", True), "missing_sites": [sid for sid in task_site_ids if sid not in site_map], } else: coverage = _get_macro_coverage() prompt = _generate_prompt(sites, coverage) return render_template("annotate.html", sites=sites, prompt=prompt, rerecord=rerecord, macro_descriptions=_MACRO_DESCRIPTIONS, macro_locations=_canonical_macro_locations(), operations=_registry.operations(), macro_coverage=_get_macro_coverage(), all_macros=[{"key": m, "group": _registry.group_of(m), "description": (_MACRO_DESCRIPTIONS.get(m) or {}).get("description", m)} for m in _registry.all_canonical()]) @annotation_bp.route("/review") def review(): """Website review mode — browse each site, leave feedback.""" sites = _load_sites() return render_template("review.html", sites=sites) @annotation_bp.route("/verify") def verify(): """Verifier builder — load a saved task and build verification checks.""" from collections import Counter from miniweb.annotation.storage import list_tasks, ANNOTATIONS_DIR task_id = request.args.get("task_id") annotator = request.args.get("annotator") # Always list every annotator's tasks — `annotator` only scopes task loading. sort = request.args.get("sort", "newest") tasks = list_tasks(newest_first=(sort != "oldest")) from miniweb.annotation.quality import summary as quality_summary for t in tasks: vf = ANNOTATIONS_DIR / (t.get("annotator") or "anonymous") / t.get("task_id", "") / "verifier.json" t["has_verifier"] = vf.exists() t["quality"] = quality_summary(vf.parent, t) t["has_feedback"] = False if t["has_verifier"]: try: t["has_feedback"] = bool(json.loads(vf.read_text()).get("feedback")) except (json.JSONDecodeError, OSError): pass annotator_counts = sorted( Counter(t.get("annotator") or "?" for t in tasks).items(), key=lambda kv: (-kv[1], kv[0])) task = None if task_id: from miniweb.annotation.storage import load_task # Find the task across annotators if annotator: task = load_task(annotator, task_id) else: from miniweb.annotation.storage import get_annotators for ann in get_annotators(): task = load_task(ann, task_id) if task: break if task: task["quality"] = quality_summary(ANNOTATIONS_DIR / (task.get("annotator") or annotator or "anonymous") / task_id, task) return render_template("verify.html", tasks=tasks, task=task, sort=sort, annotator_counts=annotator_counts, op_labels={op: info.get("label", op) for op, info in _registry.operations().items()}, macro_descriptions=_MACRO_DESCRIPTIONS) @annotation_bp.route("/graph") def graph(): """Interactive site affinity graph — visualize and edit cross-site flows.""" from miniweb.annotation.site_affinities import SITE_AFFINITY_GROUPS sites = _load_sites() return render_template("graph.html", sites=sites, affinities=SITE_AFFINITY_GROUPS) @annotation_bp.route("/accounts") def accounts_page(): """Credential lookup — every seeded login account across all sites.""" from miniweb.app import db sites = [] for meta in _load_sites(): users = db.query(meta["id"], "users", limit=200) rows = [] for u in users or []: rows.append({ "id": u.get("id"), "username": u.get("username") or "", "email": u.get("email") or "", "password": "" if u.get("password") in (None, "") else str(u["password"]), "name": u.get("name") or u.get("full_name") or u.get("display_name") or "", "role": u.get("role") or "", }) if rows: rows.sort(key=lambda r: (r["id"] is None, r["id"])) sites.append({"id": meta["id"], "name": meta.get("name") or meta["id"], "url": meta["url"], "users": rows}) return render_template("accounts.html", sites=sites) # --- Macro template builder ------------------------------------------------ @annotation_bp.route("/api/macro_sheet.csv") def api_macro_sheet_csv(): """Download the current macro + reasoning-op set as a CSV (same columns as docs/refined_macro_set.csv), generated live from the registry.""" import csv import io buf = io.StringIO() w = csv.writer(buf) w.writerow(["category", "group", "tag", "description", "span_start", "span_end", "warning"]) group_order = list(_registry.groups().keys()) for m in sorted(_final_set.keep_primitives(_registry.all_canonical()), key=lambda m: (group_order.index(_registry.group_of(m)), m)): e = _registry.entry(m) w.writerow(["base macro", e.get("group", ""), m, e.get("description", ""), e.get("span_start", ""), e.get("span_end", ""), e.get("warning", "")]) for op, info in _registry.operations().items(): # reasoning ops carry a `check` (how it's graded) rather than a design warning w.writerow(["reasoning op", "reasoning", info.get("label", op), info.get("desc", ""), "", "", info.get("check", "")]) from flask import Response return Response(buf.getvalue(), mimetype="text/csv", headers={"Content-Disposition": "attachment; filename=refined_macro_set.csv"}) # task verifier — fill the scaffold's open fields from the trajectory _VERIFIER_FILL_MODEL = "gemini-3.5-flash" def _span_actions(task, traj, macro): """The concrete recorded actions inside a macro's span — the strongest signal for what its open target/value should be.""" from miniweb.annotation import verifier_scaffold as vs acts = [e for e in traj if e.get("type") == "action"] out = [] # `macro` is canonical; aggregate every pre-merge span that maps to it for orig, span in (task.get("macro_spans") or {}).items(): if _canon(orig.split("#", 1)[0]) != _canon(macro.split("#", 1)[0]): continue for idx in vs._span_indices(span, len(acts)): a = acts[idx] out.append({k: a[k] for k in ("action", "target", "value", "text", "option_text", "url", "method") if a.get(k) not in (None, "")}) return out @annotation_bp.route("/api/suggest_task_verifier", methods=["POST"]) def api_suggest_task_verifier(): """Fill the OPEN fields of a task's universal verifier scaffolds from its trajectory. Body: {task_id, annotator?}. Every macro scaffolds the same 3 canonical components (page_visited + FE affordance + backend gate); the LLM (gemini-3.5-flash) fills their open params, grounded in the whole trajectory (observations reduced to axtree only). """ from miniweb.annotation import verifier_scaffold as vs from miniweb.app.llm import call_llm data = request.get_json(silent=True) or {} task_id = data.get("task_id") annotator = data.get("annotator") if not task_id: return jsonify({"error": "task_id required"}), 400 _dir, task, traj = _load_saved_task(annotator, task_id) if not task: return jsonify({"error": "task not found"}), 404 macros = task.get("macros") or [] draft = vs.scaffold_task(macros, task) if not draft["macros"]: return jsonify({"error": "this task has no macros tagged", "macros": {}}), 200 # per-macro context: description + the actions the human took in its span macro_ctx = [] for macro in draft["macros"]: desc = (_MACRO_DESCRIPTIONS.get(macro) or {}).get("description", "") macro_ctx.append({"macro": macro, "description": desc, "span_actions": _span_actions(task, traj, macro)}) slots_for_llm = [{"id": s["id"], "macro": s["macro"], "check": s["check_type"], "param": s["param"], "fixed": s["fixed"], "label": s.get("label", ""), "context": s.get("group_labels", [])} for s in draft["slots"]] from pathlib import Path as _P from miniweb.evaluation.trajectory import merge_server_log reduced = vs.reduce_trajectory_for_llm( merge_server_log(traj, _P(str(_dir)) / "server_log.json")) system_prompt = ( "You fill in the OPEN variables of a web task's verifier scaffolds.\n" "Every macro carries the same 3 canonical checks — page_visited, action_included\n" "(the FE affordance) and request_made (the backend gate); the structure is fixed.\n" "Your ONLY job is to supply concrete values for the open variables, read off the\n" "recorded trajectory. Do NOT invent checks, change structure, or fill a value the\n" "trajectory does not support.\n\n" "Each slot names its macro, the check type, the param to fill, the FIXED sibling\n" "params, and a `label` (what this check asserts). Use those labels as the intent:\n" "they tell you what value the slot is meant to capture. Ground each macro's slots\n" "in ITS tagged span_actions; for request_made prefer the span's mutating call and\n" "use the relative /sites/... path.\n" "Guidance by param:\n" " target -> the human-readable component the action hit (from that macro's span action's `target`)\n" " value -> the value entered/selected (the span action's value/text/option_text)\n" " url -> the request path or page URL involved (from network events or observations)\n" " method -> the HTTP method of the mutating request\n" " status -> the response status code of that request\n" " body_fields -> an object of the key payload fields that were submitted {field: value}\n" " expected -> the answer the task produces (use the task answer, or the value visible\n" " on the page the macro ends on). For qa_answer/reasoning_contains this is\n" " the value to look for; leave `mode` and `leaf` alone (leaf is auto-set).\n" " alternatives -> other acceptable surface forms of the expected answer (e.g. a name and\n" " its email), as a list; [] if none\n\n" "Return ONLY JSON: {\"fills\": {\"\": , ...}}. Use the exact slot ids.\n" "A value may be a string, number, list, or object. Omit a slot only if the trajectory\n" "genuinely has no value for it." ) user_prompt = json.dumps({ "instruction": task.get("instruction", ""), "answer": task.get("answer"), "macros": macro_ctx, "slots_to_fill": slots_for_llm, "trajectory": reduced, }, ensure_ascii=False, default=str) raw = call_llm(user_prompt, system=system_prompt, max_tokens=4000, temperature=0.2, json_mode=True, model=_VERIFIER_FILL_MODEL) if not raw: return jsonify({"error": f"LLM unavailable ({_VERIFIER_FILL_MODEL})"}), 503 try: fills = (json.loads(raw) or {}).get("fills", {}) except (json.JSONDecodeError, TypeError): return jsonify({"error": "LLM returned malformed JSON", "raw": raw[:2000]}), 502 # map slot id -> (macro, path, param), then fill each macro's tree by_id = {s["id"]: s for s in draft["slots"]} per_macro_fills = {} for sid, value in (fills or {}).items(): s = by_id.get(sid) if not s: continue per_macro_fills.setdefault(s["macro"], []).append( {"path": s["path"], "param": s["param"], "value": value}) filled = {} for macro, tree in draft["macros"].items(): filled[macro] = vs.fill_open(tree, per_macro_fills.get(macro, [])) vs.inject_qa_leaf(filled, task) # resolve qa_answer's leaf/chained per this task return jsonify({ "macros": filled, "slots": draft["slots"], "filled_count": len(fills or {}), "slot_count": len(draft["slots"]), "model": _VERIFIER_FILL_MODEL, }) def _load_test_trajectory(annotator, task_id, which): """Return (trajectory, answer, note) for the requested test source. Network events come from RECORDED evidence only: the trajectory's own (client-captured) network events, unioned with the session's server-side request log when one was saved next to the recording. Nothing is synthesized from actions — a request either has a witness or it doesn't count.""" from miniweb.annotation.storage import ANNOTATIONS_DIR from miniweb.evaluation.trajectory import merge_server_log tdir = ANNOTATIONS_DIR / annotator / task_id if which == "walk": wf = tdir / "verification_walk.json" if not wf.exists(): return None, "", "no verification walk recorded for this task" d = json.loads(wf.read_text()) return d.get("trajectory", []), d.get("answer", ""), "verification walk" if which == "agent": from pathlib import Path from miniweb.evaluation.trajectory import extract_final_reasoning rd = Path("src/miniweb/evaluation/results") / f"recorded_{task_id}" ar = rd / "trajectory.json" if not ar.exists(): return None, "", "no agent run recorded for this task" traj = json.loads(ar.read_text()) traj = merge_server_log(traj, rd / "server_log.json") # no-op if absent reasoning = extract_final_reasoning(rd) if reasoning: traj = traj + [{"type": "reasoning", "text": reasoning}] # the agent's answer is its final_result (for answer_matches, if used) answer = "" rp = rd / "result.json" if rp.exists(): answer = (json.loads(rp.read_text()) or {}).get("final_result", "") or "" return traj, answer, "agent attempt" # default: gold human trajectory + the recording session's server log tf = tdir / "trajectory.json" task = json.loads((tdir / "task.json").read_text()) if (tdir / "task.json").exists() else {} traj = json.loads(tf.read_text()) if tf.exists() else [] traj = merge_server_log(traj, tdir / "server_log.json") # no-op if absent # the human's reported answer IS the expected answer (perfect-trace assumption) answer = task.get("answer") or task.get("expected_answer") or "" return traj, answer, "gold (human)" @annotation_bp.route("/api/run_task_verifier", methods=["POST"]) def api_run_task_verifier(): """Sandbox: run a task verifier spec against a chosen trajectory. Body: {task_id, annotator, which: gold|walk|agent, macros?}. If `macros` is omitted the saved verifier.json is used. Returns the per-macro report. """ from miniweb.evaluation.verifiers import verify_task data = request.get_json(silent=True) or {} task_id = data.get("task_id") annotator = data.get("annotator") or "anonymous" which = data.get("which") or "gold" from miniweb.annotation import quality if which not in quality.SOURCES: return jsonify({"error": "unknown test source"}), 400 if not task_id: return jsonify({"error": "task_id required"}), 400 macros = data.get("macros") if macros is None: from miniweb.annotation.storage import ANNOTATIONS_DIR vf = ANNOTATIONS_DIR / annotator / task_id / "verifier.json" if vf.exists(): macros = json.loads(vf.read_text()).get("macros", {}) else: macros = {} if not macros: return jsonify({"error": "no verifier spec — suggest or save one first"}), 400 traj, answer, note = _load_test_trajectory(annotator, task_id, "gold" if which in ("empty", "answer_only") else which) if which in ("empty", "answer_only"): traj, note = [], which.replace("_", " ") + " negative check" if which == "empty": answer = "" if traj is None: return jsonify({"error": note, "which": which}), 404 # resolve qa_answer leaf/chained from the task graph before running from miniweb.annotation.storage import ANNOTATIONS_DIR from miniweb.annotation import verifier_scaffold as vs tjson = ANNOTATIONS_DIR / annotator / task_id / "task.json" tdata = {} if tjson.exists(): tdata = json.loads(tjson.read_text()) vs.inject_qa_leaf(macros, tdata) vs.refresh_expected(macros, tdata) # never grade against a stale expected report = verify_task({"task_id": task_id, "macros": macros}, traj, answer, question=tdata.get("instruction", "")) report["which"] = which report["source"] = note report["action_count"] = sum(1 for e in traj if e.get("type") == "action") saved = quality.read_json(tjson.parent / "verifier.json").get("macros", {}) report["evidence_saved"] = bool(saved) and macros == quality.effective_macros(tdata, saved) if report["evidence_saved"] and not _demo.ENABLED: quality.record_run(tjson.parent, tdata, macros, which, report) report["quality"] = quality.summary(tjson.parent, tdata) return jsonify(report) @annotation_bp.route("/api/task_quality//", methods=["GET", "POST"]) def api_task_quality(annotator, task_id): """Persist attributed AI findings without changing human review decisions.""" from miniweb.annotation import quality from miniweb.annotation.storage import ANNOTATIONS_DIR directory = (ANNOTATIONS_DIR / annotator / task_id).resolve() if directory.parent.parent != ANNOTATIONS_DIR.resolve() or not (directory / "task.json").is_file(): return jsonify({"error": "Task not found"}), 404 task = json.loads((directory / "task.json").read_text()) if request.method == "POST": try: quality.save_ai_review(directory, task, request.get_json(silent=True) or {}, session.get("annotator_name", "anonymous")) except ValueError as exc: return jsonify({"error": str(exc)}), 400 return jsonify(quality.summary(directory, task)) @annotation_bp.route("/api/review_document") def api_review_document(): from flask import abort, send_file relative = Path(request.args.get("path", "")) root = PROJECT_ROOT / "data" path = (PROJECT_ROOT / relative).resolve() if (not path.is_relative_to(root) or len(path.relative_to(root).parts) < 2 or not path.relative_to(root).parts[0].startswith(("task_review_", "task_repairs_", "verifier_audit_", "railway_pulls")) or path.suffix not in (".md", ".json", ".csv") or not path.is_file()): abort(404) return send_file(path, mimetype="text/plain") @annotation_bp.route("/api/task_verifier//", methods=["GET", "POST"]) def api_task_verifier(annotator, task_id): """Load/save a task's filled verifier spec (verifier.json next to the task).""" from miniweb.annotation.storage import ANNOTATIONS_DIR vf = ANNOTATIONS_DIR / annotator / task_id / "verifier.json" if request.method == "GET": if vf.exists(): spec = json.loads(vf.read_text()) # verifier.json freezes state at build time; never serve the builder # stale data when task.json has since changed (re-record / retag): tjson = vf.parent / "task.json" if tjson.exists(): from miniweb.annotation.verifier_scaffold import (default_scaffold, refresh_expected) tdata = json.loads(tjson.read_text()) # 1. expected answer may have changed refreshed = refresh_expected(spec.get("macros") or {}, tdata) if refreshed: spec["expected_refreshed"] = refreshed # 2. the macro TAGS may have changed — reconcile the macro set; # newly-tagged macros get the universal 3-check scaffold want = list(dict.fromkeys(_canon(m) for m in (tdata.get("macros") or []))) have = spec.get("macros") or {} removed = [m for m in have if m not in want] for m in removed: have.pop(m) added = [m for m in want if m not in have] for m in added: have[m] = default_scaffold(m, tdata) # keep the builder's card order = the task's tag order spec["macros"] = {m: have[m] for m in want if m in have} if removed: spec["macros_removed"] = removed if added: spec["macros_added"] = added return jsonify(spec) return jsonify({"task_id": task_id, "macros": {}}) data = request.get_json(silent=True) or {} if "macros" in data: from miniweb.annotation.review_mode import ensure_before ensure_before(vf.parent) # Merge-save: only overwrite keys present in the payload, so fields written # by other tools (built_by, archetype_v2, ...) survive a partial POST. spec = {} if vf.exists(): try: spec = json.loads(vf.read_text()) except (json.JSONDecodeError, OSError): spec = {} spec["task_id"] = task_id for key in ("macros", "model", "feedback", "feedback_at"): if key in data: spec[key] = data[key] if "macros" in data: # a human saving through the builder IS the redesign event — stamp it from datetime import datetime as _dt actor = session.get("annotator_name", "anonymous") from miniweb.annotation.quality import AI_ACTORS spec["built_by"] = "ai-builder-v2" if actor in AI_ACTORS else "human-builder-v2" spec["saved_by"] = actor spec["saved_at"] = _dt.now().isoformat() vf.parent.mkdir(parents=True, exist_ok=True) vf.write_text(json.dumps(spec, indent=2, default=str)) return jsonify({"status": "ok"}) # --- API --- @annotation_bp.route("/api/tasks", methods=["GET"]) def api_list_tasks(): from miniweb.annotation.storage import list_tasks annotator = request.args.get("annotator") tasks = list_tasks(annotator) if _final_set.locked(): keep = _final_set.tasks() tasks = [t for t in tasks if t.get("task_id") in keep] return jsonify(tasks) @annotation_bp.route("/api/screenshot///") def api_screenshot(annotator, task_id, filename): """Serve a screenshot file from the task's directory.""" from flask import send_from_directory from miniweb.annotation.storage import ANNOTATIONS_DIR task_dir = ANNOTATIONS_DIR / annotator / task_id return send_from_directory(str(task_dir), filename) @annotation_bp.route("/api/update_task_field", methods=["POST"]) def api_update_task_field(): """Update a single field on a saved task.""" data = request.get_json(silent=True) or {} task_id = data.get("task_id") annotator = data.get("annotator", "anonymous") field = data.get("field") value = data.get("value") if not task_id or not field: return jsonify({"error": "task_id and field required"}), 400 from miniweb.annotation.storage import ANNOTATIONS_DIR task_file = ANNOTATIONS_DIR / annotator / task_id / "task.json" if not task_file.exists(): return jsonify({"error": "Task not found"}), 404 if field != "review_tag": # a human decision is not a correction from miniweb.annotation.review_mode import ensure_before ensure_before(task_file.parent) task_data = json.loads(task_file.read_text()) task_data[field] = value task_file.write_text(json.dumps(task_data, indent=2, default=str)) return jsonify({"status": "ok", "field": field}) @annotation_bp.route("/api/tasks", methods=["POST"]) def api_create_task(): data = request.get_json(silent=True) or {} if not data.get("instruction") or not data.get("sites"): return jsonify({"error": "instruction and sites required"}), 400 # Multi-macro tasks must ship a dependency graph that touches every macro # (mirrors the client-side check in annotate.html so the API can't bypass it). # `macros` is a list of BASE names (duplicates allowed); when duplicates are # present the client also sends `macro_instances` (per-occurrence ids) and the # edges/subtasks are keyed by those ids, so validate against the instance list. macros = data.get("macros") or [] macro_instances = data.get("macro_instances") or [] connectivity_keys = macro_instances if macro_instances else macros if len(connectivity_keys) >= 2: edges = data.get("macro_edges") or [] linked = {e.get("from") for e in edges} | {e.get("to") for e in edges} orphans = [m for m in connectivity_keys if m not in linked] if orphans: return jsonify({"error": "macro dependencies incomplete", "unconnected_macros": orphans}), 400 # Register any proposed (new-to-registry) macros into data/macros.yaml so # they become real, reusable macros — not just per-task strings. global _MACRO_DESCRIPTIONS registered, reg_errors = [], {} _defs = data.get("proposed_macro_defs") or {} for nm in (data.get("proposed_macros") or []): if _registry.is_known(nm): continue d = _defs.get(nm) or {} try: _registry.register_macro(nm, d.get("group", ""), d.get("description", ""), d.get("span_start", ""), d.get("span_end", "")) registered.append(nm) except (ValueError, OSError) as e: reg_errors[nm] = str(e) if registered: _MACRO_DESCRIPTIONS = _registry.descriptions() # refresh derived cache sites_list = data.get("sites", []) task = { "task_id": f"{'_'.join(s['id'] if isinstance(s, dict) else s for s in sites_list[:2])}_{uuid.uuid4().hex[:6]}", "site": sites_list[0]["id"] if isinstance(sites_list[0], dict) else sites_list[0], "sites": [s["id"] if isinstance(s, dict) else s for s in sites_list], "instruction": data["instruction"], "macros": data.get("macros", []), "num_sites": len(sites_list), # NOTE: `macro_instances` is added below ONLY for duplicate-macro tasks, so # tasks without duplicates serialize identically to the pre-duplicate schema. "num_macros": len(data.get("macros", [])), "expected_answer": data.get("expected_answer"), "answer_type": data.get("answer_type", "string"), "alternatives": data.get("alternatives", ""), "expected_outcome": data.get("expected_outcome", ""), "macro_edges": data.get("macro_edges", []), "macro_positions": data.get("macro_positions", {}), "macro_subtasks": data.get("macro_subtasks", {}), "macro_spans": data.get("macro_spans", {}), "macro_operations": data.get("macro_operations", {}), "proposed_macros": data.get("proposed_macros", []), "proposed_macro_defs": data.get("proposed_macro_defs", {}), "qa_answers": data.get("qa_answers", {}), "starting_url": data.get("starting_url", ""), "requires_login": data.get("requires_login", True), "trajectory": data.get("trajectory", []), "server_log": data.get("server_log", []), "beacon_log": data.get("beacon_log", []), "agent_result": data.get("agent_result"), "annotator": data.get("annotator", "anonymous"), } # Duplicate-macro tasks: persist the per-occurrence instance ids so reload can # reconstruct the distinct nodes (their subtasks/ops/edges are keyed by these # ids). Only stored when the macro list actually repeats a name — a task with # no duplicates omits the field entirely and stays byte-compatible with the # existing (gold) tasks. GRADING NOTE: for duplicate tasks the QA leaf/terminal # detection in verifier_scaffold.leaf_macros compares BASE `macros` against the # instance-id edge endpoints, so per-occurrence leaf grading isn't resolved for # duplicates; non-duplicate tasks (all gold) are unaffected. if macro_instances and len(macro_instances) == len(task["macros"]) \ and any(i != b for i, b in zip(macro_instances, task["macros"])): task["macro_instances"] = macro_instances # Re-record: replace an existing task in place. The new recording keeps # the original task_id and annotator dir (external references stay # stable); the previous version moves to the recoverable .trash. replace_id = data.get("replace_task_id") trashed_from = None if replace_id: from miniweb.annotation.storage import ANNOTATIONS_DIR, get_annotators, trash_task orig_annotator = data.get("replace_annotator") or next( (a for a in get_annotators() if (ANNOTATIONS_DIR / a / replace_id).exists()), None) if not orig_annotator or not (ANNOTATIONS_DIR / orig_annotator / replace_id).exists(): return jsonify({"error": "Task to replace not found"}), 404 task["task_id"] = replace_id task["annotator"] = orig_annotator task["rerecorded_at"] = datetime.now().isoformat() task["rerecorded_by"] = data.get("annotator", "anonymous") # Carry over reviewer-authored data from the old task: the refined # instruction and the review note survive; the approve/reject status # and the verified/stale tag reset — the recording just changed, so # it needs review again. try: old = json.loads((ANNOTATIONS_DIR / orig_annotator / replace_id / "task.json").read_text()) if old.get("instruction_ambiguous"): task["instruction_ambiguous"] = old["instruction_ambiguous"] # Keep the AI findings as history; their hashes mark them outdated. if old.get("ai_review"): task["ai_review"] = old["ai_review"] if old.get("review", {}).get("note"): task["review"] = {"status": "pending", "note": old["review"]["note"], "reviewer": old["review"].get("reviewer", ""), "reviewed_at": ""} except (OSError, json.JSONDecodeError): pass # trash-then-save: moving the whole old dir away also removes stale # screenshots/ and logs that a plain overwrite would leave behind trashed_from = trash_task(orig_annotator, replace_id) try: task_id = _save_task(task) except Exception: if trashed_from: # restore the original so a failed replace never loses the task import shutil shutil.move(trashed_from, str(ANNOTATIONS_DIR / task["annotator"] / replace_id)) raise if trashed_from: # the verifier spec lives in a sidecar file — preserve it across the # replace (the reviewer's checks usually still apply to the redo) old_vf = Path(trashed_from) / "verifier.json" if old_vf.exists(): import shutil shutil.copy2(old_vf, ANNOTATIONS_DIR / task["annotator"] / task_id / "verifier.json") # Derive axtree + screenshot from the recorded HTML snapshots so saved # observations match the backfilled/replayed data format (async). from miniweb.annotation.observations import schedule_completion schedule_completion(task["annotator"], task_id, request.host_url.rstrip("/")) # Record N/A macros for future sampling improvement na_macros = data.get("macros_not_applicable", {}) sites_list = data.get("sites", []) site_ids = [s["id"] if isinstance(s, dict) else s for s in sites_list] annotator = data.get("annotator", "anonymous") for macro in na_macros: _save_na_report(site_ids, macro, annotator) return jsonify({"task_id": task_id, "status": "saved", "registered_macros": registered, "macro_errors": reg_errors}) @annotation_bp.route("/api/sites") def api_sites(): return jsonify(_load_sites()) @annotation_bp.route("/api/macros") def api_macros(): return jsonify(_final_set.keep_primitives(_load_macros())) @annotation_bp.route("/api/macro_descriptions") def api_macro_descriptions(): return jsonify(_MACRO_DESCRIPTIONS) def _macro_search_keyword(query, exclude, limit): """Offline relevance ranker over the macro catalog. Token-overlap scoring across name + description + verb + modality + example, with a boost for hits in the macro name. Used both as the no-API-key fallback and to seed the LLM shortlist. Returns a ranked list of macro-name strings. """ import re terms = [t for t in re.split(r"[^a-z0-9]+", query.lower()) if len(t) > 1] if not terms: return [] scored = [] for name, meta in _MACRO_DESCRIPTIONS.items(): if name in exclude: continue name_l = name.lower() blob = " ".join([ name_l.replace("_", " "), (meta.get("description") or "").lower(), (meta.get("verb") or "").lower(), (meta.get("modality") or "").lower(), (meta.get("example") or "").lower(), ]) score = 0 for t in terms: if t in name_l: score += 3 score += blob.count(t) if score: scored.append((score, name)) scored.sort(key=lambda s: (-s[0], s[1])) return [name for _, name in scored[:limit]] @annotation_bp.route("/api/macro_search", methods=["POST"]) def api_macro_search(): """Semantic (LLM) search over the macro catalog by natural-language query. Body: {query: str, exclude?: [str], limit?: int} Returns {mode: "semantic"|"keyword", results: [{key, description, verb, modality}]}. Falls back to the offline keyword ranker when no LLM is configured or the model response can't be used, so the box always returns something useful. """ data = request.get_json(silent=True) or {} query = (data.get("query") or "").strip() exclude = set(data.get("exclude") or []) if _final_set.locked(): exclude |= set(_MACRO_DESCRIPTIONS) - _final_set.primitives() try: limit = max(1, min(15, int(data.get("limit", 8)))) except (ValueError, TypeError): limit = 8 if not query: return jsonify({"error": "query required"}), 400 def _pack(names): out = [] for n in names: meta = _MACRO_DESCRIPTIONS.get(n) if not meta: continue out.append({"key": n, "description": meta.get("description") or n, "verb": meta.get("verb", ""), "modality": meta.get("modality", "")}) return out # Shortlist with the offline ranker to keep the prompt small, but always # give the model the full catalog names so it can pick beyond keyword hits. shortlist = _macro_search_keyword(query, exclude, 40) catalog_lines = [] for name, meta in _MACRO_DESCRIPTIONS.items(): if name in exclude: continue catalog_lines.append( f"{name}: {meta.get('description','')} " f"[{meta.get('verb','')}/{meta.get('modality','')}]") catalog = "\n".join(catalog_lines) system_prompt = ( "You help an annotator find UI-skill macros by meaning.\n" "Each macro is a primitive skill (e.g. filter_by_slider, " "create_by_form). Given a natural-language description of what the " "annotator wants to do, return the most relevant macros, most " "relevant first.\n" "Match on intent and UI mechanism, not just shared words — e.g. " "'drag a price range' -> filter_by_slider, 'find papers about a " "topic' -> search_by_semantic.\n" "Only choose macro names from the provided catalog. " f"Return ONLY a JSON array of up to {limit} macro-name strings, " 'no prose. Example: ["filter_by_slider", "sort_by_column"]' ) user_prompt = f"Annotator wants: {query}\n\nMacro catalog:\n{catalog}" result = _call_llm(system_prompt, user_prompt) if result: try: import re m = re.search(r"\[.*\]", result, re.DOTALL) if m: names = json.loads(m.group()) names = [n for n in names if isinstance(n, str) and n in _MACRO_DESCRIPTIONS and n not in exclude] if names: return jsonify({"mode": "semantic", "results": _pack(names[:limit])}) except Exception: pass # fall through to keyword ranking return jsonify({"mode": "keyword", "results": _pack(shortlist[:limit])}) @annotation_bp.route("/api/schema/") def api_site_schema(site_id): """Return DB schema for a site: collections and their fields.""" from miniweb.app import db import sqlite3 conn = db.get_conn() prefix = site_id.replace("-", "_") tables = conn.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name LIKE ?", (f"{prefix}%",), ).fetchall() schema = {} for (table_name,) in tables: coll = table_name[len(prefix) + 1:] # strip prefix_ if not coll or coll.startswith("fts_"): continue cols = conn.execute(f"PRAGMA table_info([{table_name}])").fetchall() fields = [{"name": c[1], "type": c[2]} for c in cols] # Get 2 sample values per field try: sample = conn.execute(f"SELECT * FROM [{table_name}] LIMIT 2").fetchall() if sample: for f in fields: vals = [] for row in sample: v = row[next(i for i, c in enumerate(cols) if c[1] == f["name"])] if v is not None: vals.append(str(v)[:50]) f["samples"] = vals except Exception: pass schema[coll] = fields return jsonify(schema) @annotation_bp.route("/api/macro_locations/") def api_macro_locations(site_id): """Return macro-to-location map for a specific site (canonical keys).""" return jsonify(_canonical_macro_locations().get(site_id, {})) @annotation_bp.route("/api/site_macros/") def api_site_macros(site_id): return jsonify(_final_set.keep_primitives(_load_site_macros(site_id))) @annotation_bp.route("/api/coverage") def api_coverage(): return jsonify(_get_macro_coverage()) @annotation_bp.route("/api/cell_counts") def api_cell_counts(): cells = _get_cell_counts() return jsonify({f"{n},{m}": c for (n, m), c in cells.items()}) @annotation_bp.route("/api/prompt") def api_prompt(): """Get a new prompt. ?site= — annotator-chosen site: only macros are sampled, with the per-site coverage floor prioritized (single-site task base). ?single=1 — random single-site prompt (legacy behavior). ?n_sites=N — force the task's site count (1, 2, or 3). (neither) — graph-aware multi-site sampling (mixed site counts). """ sites = _load_sites() site_id = (request.args.get("site") or "").strip() if site_id: site = next((s for s in sites if s["id"] == site_id), None) if not site: return jsonify({"error": f"unknown site: {site_id}"}), 404 return jsonify(_generate_site_prompt(site)) coverage = _get_macro_coverage() single = request.args.get("single") == "1" n_sites = request.args.get("n_sites", type=int) prompt = _generate_prompt(sites, coverage, force_single=single, n_sites_override=n_sites) return jsonify(prompt) @annotation_bp.route("/api/floor_k", methods=["GET", "POST"]) def api_floor_k(): """Get or set the coverage-floor K (sites per macro).""" if request.method == "POST": data = request.get_json(silent=True) or {} try: k = max(1, min(10, int(data.get("k")))) except (TypeError, ValueError): return jsonify({"error": "k must be an integer 1-10"}), 400 _FLOOR_CONFIG_FILE.write_text(json.dumps({"k": k})) return jsonify({"k": k}) return jsonify({"k": _get_floor_k()}) @annotation_bp.route("/api/site_floors") def api_site_floors(): """Per-site K-floor status: how many macros still need THIS site. covered/total where total = site's macro pool and covered = macros that are either already demonstrated here or globally at their K-site target. """ global_cov = _get_macro_site_coverage() k_targets = _macro_k_targets() floors = {} for s in _load_sites(): sid = s["id"] pool = _site_macro_pool(sid) needy = sum(1 for m in pool if sid not in global_cov.get(m, set()) and len(global_cov.get(m, set())) < k_targets.get(m, _get_floor_k())) floors[sid] = {"covered": len(pool) - needy, "total": len(pool)} return jsonify(floors) @annotation_bp.route("/api/auto_login", methods=["POST"]) def api_auto_login(): """Log into the specified sites as user 1 (Alex Rivera). Sets session["user_id"] and any site-specific session keys so the sites render as logged-in when the browser loads them. """ data = request.get_json(silent=True) or {} site_ids = data.get("sites", []) if not site_ids: return jsonify({"error": "sites list required"}), 400 from miniweb.app import db # Clear the no-autologin flag so before_request auto-login works normally session.pop("_no_autologin", None) # Site-specific session keys for sites that don't use the generic "user_id" SITE_SESSION_KEYS = { "conference-review-submission": "conf_review_uid", "health-portals": "health_user_id", "insurance-loans": "il_user_id", "instant-messaging": "im_user_id", "job-sites": "job_sites_user_id", "live": "live_user_id", "qa-knowledge": "qa_user_id", "university-academic": "ua_user", "version-control": "vc_user_id", "video": "user_id", # also needs username + display_name "weather": "weather_user_id", "team-chat-workspace": "user_id", # uses root_user_id "remote-calls": "user_id", # uses root_user_id, not the rc-u-XXX id } logged_in = [] for site_id in site_ids: try: users = db.query(site_id, "users", limit=5) if not users: continue # The user agents are signed in as (id or root_user_id 1); else Alex Rivera; else the first user. # A name match alone picked Carlos Rivera (id 5) on SportsHub, so recordings showed another user's data. user = next((u for u in users if u.get("id") in (1, "1") or u.get("root_user_id") in (1, "1")), None) for u in ([] if user else users): name = " ".join(str(v) for v in [ u.get("name", ""), u.get("display_name", ""), u.get("username", "") ]).lower() if "alex" in name or "rivera" in name: user = u break if not user: user = users[0] uid = user.get("id", user.get("root_user_id", 1)) # Sites that read session["user_id"] as a root_user_id if site_id in ("team-chat-workspace", "remote-calls"): uid = user.get("root_user_id", uid) # Namespaced per-site key (src/miniweb/app/__init__ swaps this into # session["user_id"] on requests to /sites//...) session[f"_uid_{site_id}"] = uid # Generic key too, for the current request cycle session["user_id"] = uid # Set site-specific session key if site_id in SITE_SESSION_KEYS: key = SITE_SESSION_KEYS[site_id] if site_id == "university-academic": session[key] = user.get("net_id", user.get("username", "")) elif site_id == "version-control": session[key] = user.get("root_user_id", uid) session["vc_username"] = user.get("username", "") session["vc_name"] = user.get("name", "") elif site_id in ("team-chat-workspace", "remote-calls"): session[key] = user.get("root_user_id", uid) elif site_id == "video": session[key] = uid session["username"] = user.get("username", "") session["display_name"] = user.get("display_name", user.get("name", "")) else: session[key] = uid logged_in.append(site_id) except Exception as exc: import traceback traceback.print_exc() return jsonify({"logged_in": logged_in}) @annotation_bp.route("/api/auto_logout", methods=["POST"]) def api_auto_logout(): """Clear all login-related session keys so sites render as logged out.""" # All known login session keys login_keys = [ "user_id", "conf_review_uid", "health_user_id", "health_pending_verify_id", "il_user_id", "im_user_id", "job_sites_user_id", "live_user_id", "qa_user_id", "ua_user", "vc_user_id", "vc_username", "vc_name", "weather_user_id", "username", "display_name", ] cleared = [] for key in login_keys: if key in session: session.pop(key) cleared.append(key) # Namespaced per-site keys (src/miniweb/app/__init__ per-site session isolation) for key in [k for k in session if k.startswith("_uid_")]: session.pop(key) cleared.append(key) # Prevent the before_request auto-login from re-setting user_id session["_no_autologin"] = True return jsonify({"cleared": cleared}) @annotation_bp.route("/api/toggle_2fa", methods=["POST"]) def api_toggle_2fa(): """Enable or disable 2FA for the current session.""" data = request.get_json(silent=True) or {} if data.get("disable"): session["_disable_2fa"] = True else: session.pop("_disable_2fa", None) return jsonify({"disabled": bool(session.get("_disable_2fa"))}) @annotation_bp.route("/api/reset_tasks", methods=["POST"]) def api_reset_tasks(): """Delete annotation tasks from file storage. Requires annotator name.""" from miniweb.annotation.storage import list_tasks, delete_task, get_annotators data = request.get_json(silent=True) or {} annotator = data.get("annotator", "") if not annotator: return jsonify({"error": "Must specify annotator", "annotators": get_annotators()}), 400 tasks = list_tasks(annotator) count = 0 for t in tasks: if delete_task(annotator, t["task_id"]): count += 1 return jsonify({"message": f"Deleted {count} tasks for '{annotator}'.", "deleted": count, "annotator": annotator}) @annotation_bp.route("/api/annotators") def api_annotators(): from miniweb.annotation.storage import get_annotators, get_stats return jsonify({"annotators": get_annotators(), "stats": get_stats()}) @annotation_bp.route("/api/tasks/") def api_get_task(task_id): from miniweb.annotation.storage import load_task annotator = request.args.get("annotator", "") if not annotator: # Try all annotators from miniweb.annotation.storage import get_annotators for ann in get_annotators(): task = load_task(ann, task_id) if task: return jsonify(task) return jsonify({"error": "Task not found"}), 404 task = load_task(annotator, task_id) if not task: return jsonify({"error": "Task not found"}), 404 return jsonify(task) @annotation_bp.route("/api/tasks/", methods=["DELETE"]) def api_delete_task(task_id): """Move a task to the recoverable .trash (per-task delete from review).""" from miniweb.annotation.storage import ANNOTATIONS_DIR, get_annotators, trash_task annotator = request.args.get("annotator", "") if not annotator: annotator = next((a for a in get_annotators() if (ANNOTATIONS_DIR / a / task_id).exists()), "") dest = trash_task(annotator, task_id) if annotator else None if not dest: return jsonify({"error": "Task not found"}), 404 return jsonify({"status": "trashed", "task_id": task_id, "trash_path": dest}) @annotation_bp.route("/api/report", methods=["POST"]) def api_report(): """Save a skip report (reason + details) and record any NA macros.""" data = request.get_json(silent=True) or {} sites = data.get("sites", []) annotator = data.get("annotator", "anonymous") # Persist the full skip report — reason/details were previously discarded report = { "sites": sites, "macros": data.get("macros", []), "reason": data.get("reason", ""), "details": data.get("details", ""), "instruction": data.get("instruction", ""), "annotator": annotator, "timestamp": datetime.now().isoformat(), } reports = [] if _SKIP_REPORTS_FILE.exists(): try: reports = json.loads(_SKIP_REPORTS_FILE.read_text()) except (json.JSONDecodeError, OSError): reports = [] reports.append(report) _SKIP_REPORTS_FILE.write_text(json.dumps(reports, indent=2)) # Persist NA macro reports na_macros = data.get("macros_not_applicable", {}) saved_na = 0 for macro, _site_info in na_macros.items(): _save_na_report(sites, macro, annotator) saved_na += 1 return jsonify({"status": "reported", "na_saved": saved_na, "skip_saved": True}) @annotation_bp.route("/api/routes/") def api_routes(site_id): """Extract API routes from a site's routes.py for annotator reference.""" import re as _re routes_file = SITES_DIR / site_id / "routes.py" if not routes_file.exists(): return jsonify([]) try: content = routes_file.read_text() except OSError: return jsonify([]) results = [] func_pattern = _re.compile( r'@\w+\.route\(["\']([^"\']+)["\']\s*' r'(?:,\s*methods=\[([^\]]*)\])?\)' r'(.*?)(?=\n@\w+\.route|\nclass |\Z)', _re.DOTALL, ) for match in func_pattern.finditer(content): path = match.group(1) methods_raw = match.group(2) body = match.group(3) if "/api/" not in path: continue if methods_raw: methods = [m.strip().strip("'\"") for m in methods_raw.split(",")] else: methods = ["GET"] func_match = _re.search(r'def\s+(\w+)\s*\(', body) func_name = func_match.group(1) if func_match else "" docstring = "" doc_match = _re.search(r'"""(.+?)"""', body, _re.DOTALL) if not doc_match: doc_match = _re.search(r"'''(.+?)'''", body, _re.DOTALL) if doc_match: docstring = doc_match.group(1).strip().split("\n")[0].strip() params = list(dict.fromkeys( _re.findall(r'request\.args\.get\(["\'](\w+)', body) )) body_fields = list(dict.fromkeys( _re.findall(r'data(?:\.get)?\(?["\'](\w+)', body) )) response_fields = list(dict.fromkeys( _re.findall(r'jsonify\(\{["\'](\w+)', body) ))[:6] for method in methods: desc = docstring or _generate_route_description( method, path, func_name, params, body_fields, ) results.append({ "method": method, "path": path, "description": desc, "params": params if method == "GET" else [], "body_fields": body_fields if method != "GET" else [], "returns": response_fields, }) return jsonify(results) def _generate_route_description(method, path, func_name, params, body_fields): parts = [p for p in path.replace("/api/", "").split("/") if p and not p.startswith("<")] resource = parts[0].replace("_", " ").replace("-", " ") if parts else "resource" action = parts[-1].replace("_", " ").replace("-", " ") if len(parts) > 1 else "" has_id = "<" in path is_sub = len(parts) > 1 if method == "GET": if action == "search": return f"Keyword search across {resource}" if action == "semantic": return f"Semantic/natural-language search across {resource}" if action == "stats": return f"Get aggregate statistics for {resource}" if action == "export": return f"Export {resource} data" if action in ("count", "counts"): return f"Count {resource}" if is_sub and action: return f"Get {action} for {resource.rstrip('s')}" if has_id: return f"Get single {resource.rstrip('s')} by ID" if params: filterable = ", ".join(params[:5]) return f"List {resource} (filter: {filterable})" return f"List all {resource}" elif method == "POST": if "login" in path or "auth" in path: return "Authenticate and create session" if "register" in path or "signup" in path: return "Register new user account" if is_sub and action: act = action.capitalize() if action in ("follow", "save", "star", "bookmark", "like", "subscribe", "favorite", "flag", "report"): return f"Toggle {action} on {resource.rstrip('s')}" return f"{act} on {resource.rstrip('s')}" if body_fields: return f"Create {resource.rstrip('s')} ({', '.join(body_fields[:4])})" return f"Create new {resource.rstrip('s')}" elif method == "PUT": return f"Update {resource.rstrip('s')}" elif method == "DELETE": return f"Delete {resource.rstrip('s')} by ID" return f"{method} {resource}" @annotation_bp.route("/api/graph/edges") def api_graph_edges(): """Get all graph edges from proposed_edges.csv.""" import csv edges = [] edges_file = Path(__file__).parent / "proposed_edges.csv" if edges_file.exists(): with open(edges_file) as f: for i, row in enumerate(csv.DictReader(f)): if row.get("status") == "approved": edges.append({ "id": i, "source": row.get("source", ""), "target": row.get("target", ""), "event_type": row.get("event_type", ""), "kind": row.get("kind", ""), }) return jsonify(edges) @annotation_bp.route("/api/graph/edges", methods=["POST"]) def api_add_graph_edge(): """Add a new edge.""" data = request.get_json(silent=True) or {} source = data.get("source", "") target = data.get("target", "") event_type = data.get("event_type", "") if not source or not target or not event_type: return jsonify({"error": "source, target, and event_type required"}), 400 edge_id = None return jsonify({"id": edge_id, "status": "saved"}) @annotation_bp.route("/api/graph/edges/", methods=["DELETE"]) def api_delete_graph_edge(edge_id): """Delete a custom edge.""" return jsonify({"status": "deleted"}) @annotation_bp.route("/api/graph/positions") def api_graph_positions(): """Get saved node positions.""" return jsonify({}) @annotation_bp.route("/api/graph/positions", methods=["POST"]) def api_save_graph_positions(): """Save node positions.""" return jsonify({"status": "saved"}) @annotation_bp.route("/api/graph/seed", methods=["POST"]) def api_seed_graph(): """Seed built-in edges from site_affinities.py event flow map.""" builtin_edges = [ # purchase ("e-commerce", "banking", "purchase"), ("e-commerce", "email", "purchase"), ("auctions-p2p-marketplaces", "banking", "purchase"), ("auctions-p2p-marketplaces", "email", "purchase"), ("books-comics", "banking", "purchase"), ("books-comics", "email", "purchase"), ("ticketing-events", "banking", "purchase"), ("ticketing-events", "email", "purchase"), # payment ("crowdfunding-donations", "banking", "payment"), ("tax-filing-dmv-permits", "banking", "payment"), ("flights-hotels", "banking", "payment"), ("insurance-loans", "banking", "payment"), # trade ("brokerage", "banking", "trade"), # booking ("flights-hotels", "calendar-todo", "booking"), ("flights-hotels", "email", "booking"), ("health-portals", "calendar-todo", "booking"), ("health-portals", "email", "booking"), ("remote-calls", "calendar-todo", "booking"), ("remote-calls", "email", "booking"), ("ticketing-events", "calendar-todo", "booking"), ("ticketing-events", "email", "booking"), # signup ("auctions-p2p-marketplaces", "email", "signup"), ("auctions-p2p-marketplaces", "password-managers", "signup"), ("forums", "email", "signup"), ("forums", "password-managers", "signup"), ("health-portals", "email", "signup"), ("health-portals", "password-managers", "signup"), ("live", "email", "signup"), ("live", "password-managers", "signup"), ("qa-knowledge", "email", "signup"), ("qa-knowledge", "password-managers", "signup"), ("real-estate-buy-rent", "email", "signup"), ("real-estate-buy-rent", "password-managers", "signup"), ("ticketing-events", "email", "signup"), # subscribe / inquiry ("business-company", "email", "subscribe"), ("business-company", "email", "inquiry"), # file_created ("documents", "cloud-storage-file-transfer", "file_created"), ("documents", "email", "file_created"), ("handwritten-notes-whiteboards", "cloud-storage-file-transfer", "file_created"), ("spreadsheets-slides", "cloud-storage-file-transfer", "file_created"), ("insurance-loans", "cloud-storage-file-transfer", "file_created"), # message ("dating", "instant-messaging", "message"), # 2FA ("password-managers", "instant-messaging", "2fa_code"), ] count = 0 for src, tgt, evt in builtin_edges: count += 1 return jsonify({"seeded": count}) @annotation_bp.route("/api/refine", methods=["POST"]) def api_refine_instruction(): """Refine an annotator's task instruction using an LLM.""" data = request.get_json(silent=True) or {} instruction = data.get("instruction", "").strip() if not instruction: return jsonify({"error": "No instruction provided"}), 400 sites = data.get("sites", []) macros = data.get("macros", []) macro_context = "" if macros: descs = [] for m in macros[:6]: d = _MACRO_DESCRIPTIONS.get(m, {}) if d: descs.append(f"- {m}: {d.get('description', '')}") if descs: macro_context = "\n\nThe task should exercise these interaction patterns:\n" + "\n".join(descs) site_context = "" if sites: site_context = f"\n\nTarget sites: {', '.join(sites)}" system_prompt = ( "Rewrite this task instruction so it sounds like a person casually asking " "their assistant to do something on a website. Keep it natural and conversational " "— like a Slack message or a quick verbal request, not a formal test case.\n\n" "Rules:\n" "- Use 'I need you to...', 'Can you...', 'Go to... and...', 'Find me...' etc.\n" "- Be specific about what to do but don't over-explain obvious steps\n" "- Include concrete values (names, numbers, categories) when the draft has them\n" "- One short paragraph, no bullet points or numbered steps\n" "- Don't mention 'the website' or 'the page' — just say what to do\n\n" "Output ONLY the rewritten instruction, nothing else." + site_context + macro_context ) refined = _call_llm(system_prompt, instruction) if refined: return jsonify({"refined": refined}) return jsonify({"error": "LLM unavailable — check OPENAI_API_KEY in .env"}), 503 def _call_llm(system_prompt, user_prompt): """Call LLM via the shared model-routing helper (default model).""" from miniweb.app.llm import call_llm return call_llm(user_prompt, system=system_prompt, max_tokens=500, temperature=0.4) def _site_data_samples(site_id, max_collections=5, rows=3, maxlen=48): """A compact snapshot of REAL entities on a site, so a drafted task references real items (product names, payees, subreddits …) instead of generic ones.""" from miniweb.app import db try: colls = [r["collection"] for r in db.execute( "SELECT collection FROM site_registry WHERE site=? ORDER BY collection", (site_id,))] except Exception: colls = [] skip = {"users", "reddit_users", "cc_users", "sessions"} colls = [c for c in colls if c not in skip and not c.startswith("fts_")][:max_collections] label_fields = ("name", "title", "subject", "subreddit", "reference", "payee_name", "merchant", "product_name", "id") out = [] for c in colls: try: recs = db.query(site_id, c, limit=rows) except Exception: continue items = [] for r in recs: lab = next((str(r[f]) for f in label_fields if r.get(f) not in (None, "")), None) if lab is None: lab = str(next((v for v in r.values() if isinstance(v, str) and v), "")) if lab: items.append(lab[:maxlen]) if items: out.append(f" {c}: " + "; ".join(items)) return "\n".join(out) @annotation_bp.route("/api/draft_task", methods=["POST"]) def api_draft_task(): """Draft a concrete, realistic task from the sampled macros — feeding the LLM each macro's enriched UI location + REAL site data + the macro/op meaning, so the annotator edits a specific draft instead of writing from scratch.""" import json as _json data = request.get_json(silent=True) or {} site_ids = [s["id"] if isinstance(s, dict) else s for s in (data.get("sites") or [])] macros = [m for m in (data.get("macros") or []) if m] if not site_ids or not macros: return jsonify({"error": "sites and macros required"}), 400 locs = _canonical_macro_locations() ops = _registry.operations() from miniweb.app import discover_sites brand = {s["id"]: s.get("name", s["id"]) for s in discover_sites()} interactions = [] used_sites = [] # order-preserving list of sites actually hosting a macro site_load = {sid: 0 for sid in site_ids} for m in macros: base, _, op = m.partition(".") desc = (_MACRO_DESCRIPTIONS.get(base) or {}).get("description", base) # Assign the macro to a requested site that actually hosts it, spreading # macros across sites (least-loaded first) so a cross-site draft gets a real # A-on-site-X / B-on-site-Y structure instead of collapsing onto one site. candidates = [sid for sid in site_ids if (locs.get(sid) or {}).get(base)] if candidates: host = min(candidates, key=lambda s: (site_load[s], site_ids.index(s))) w = (locs.get(host) or {}).get(base) where = w[0] if isinstance(w, list) else str(w) else: host, where = site_ids[0], "" site_load[host] += 1 entry = {"site": brand.get(host, host), "macro": base, "does": desc, "where": where or "(unspecified)"} if op: entry["reasoning"] = (ops.get(op) or {}).get("desc", op) interactions.append(entry) if host not in used_sites: used_sites.append(host) cross_site = len(site_ids) > 1 or len(used_sites) > 1 site_data = "\n".join(f"{brand.get(sid, sid)} ({sid}):\n{_site_data_samples(sid)}" for sid in site_ids) site_names = ", ".join(brand.get(sid, sid) for sid in site_ids) system = ( "You draft a realistic web task for a benchmark annotator. Given a list of " "interactions (each with WHAT it does, WHERE it lives, and the SITE it's on) " "and REAL data currently on those sites, write ONE natural task a person would " "ask their assistant that WEAVES ALL the interactions into a single coherent " "goal (not a list of unrelated actions). It must be concrete and specific, " "REFERENCING REAL ENTITIES/VALUES from the data and the exact options named in " "each 'where'. If an interaction has a reasoning op, the task must genuinely " "require that reasoning. Keep the instruction to 1-2 sentences, conversational. " "For each macro's subtask, write the CONCRETE step to take on the page (the " "specific control + the specific value/entity) — do NOT restate the generic " 'macro definition. Reply ONLY JSON: {"instruction": "", "subtasks": ' '{"": ""}}.') if cross_site: system += ( " IMPORTANT — this is a CROSS-SITE task spanning multiple sites (each " "interaction names the SITE it happens on). Write ONE goal that genuinely " "REQUIRES MOVING BETWEEN these sites: carry a real value/result from one " "site to the next (e.g. read/compute something on one site and use that " "exact value to act on another). Name that carried value explicitly in the " "instruction. The agent moves between sites via the portal (search bar / " "opening the other site) — do not assume they are the same site. Every " "listed site must be essential to completing the task.") user = _json.dumps({"sites": site_names, "cross_site": cross_site, "interactions": interactions, "real_site_data": site_data}) from miniweb.app.llm import call_llm raw = call_llm(user, system=system, max_tokens=600, temperature=0.5, json_mode=True) if not raw: return jsonify({"error": "LLM unavailable"}), 503 try: return jsonify(_json.loads(raw)) except (ValueError, TypeError): return jsonify({"instruction": raw.strip(), "subtasks": {}}) def _collect_playback_facts(): """Every mini-player video across sites + the facts hidden in its stream. Reproduces each site's player identity key (page path + "|" + data-title) and duration exactly as the templates set them, then reads the deterministic timeline from app.playback — so annotators see WHAT info appears at WHAT span without scrubbing, and can design search_by_playback tasks against it. """ import json as _json from miniweb.app import db from miniweb.app.playback import generate_timeline def fmt(sec): return f"{int(sec) // 60}:{int(sec) % 60:02d}" rows = [] def add(site, key_path, title, duration, url=None, flavor="video"): try: tl = generate_timeline(key_path + "|" + (title or ""), int(duration), flavor) except Exception: return facts = [{"label": s["label"], "value": s["value"], "start": fmt(s["start"]), "end": fmt(s["end"])} for s in tl["segments"] if s.get("kind") == "fact"] rows.append({"site": site, "url": url or key_path, "title": title, "duration": fmt(tl["duration"]), "facts": facts}) def _j(v, default): if isinstance(v, str): try: return _json.loads(v) except (ValueError, TypeError): return default return v if v is not None else default try: # StreamHub for v in db.query("video", "videos", sort="id", limit=200): add("video", f"/sites/video/watch/{v['id']}", v.get("title", ""), v.get("duration_seconds") or 0) except Exception: pass try: # PixShare video posts for p in db.query("multimedia-posting", "posts", where={"type": "video"}, limit=200): add("multimedia-posting", f"/sites/multimedia-posting/post/{p['id']}", (p.get("caption") or "Video post")[:60], p.get("video_duration") or 30) except Exception: pass try: # Remote-calls recordings for r in db.query("remote-calls", "recordings", limit=200): add("remote-calls", f"/sites/remote-calls/recording/{r['id']}", r.get("title", ""), (r.get("duration_minutes") or 0) * 60) except Exception: pass try: # Sports highlights (finished matches only — others show no player) teams = {t["id"]: t.get("name", "") for t in db.query("sports-esports", "teams", limit=500)} for m in db.query("sports-esports", "matches", where={"status": "final"}, limit=200): home = teams.get(m.get("home_team_id"), "") away = teams.get(m.get("away_team_id"), "") add("sports-esports", f"/sites/sports-esports/match/{m['id']}/highlights", f"{home} vs {away}", 3540) except Exception: pass try: # Project homepages video resources for r in db.query("project-homepages", "resources", where={"type": "video"}, limit=200): dm = r.get("duration_minutes") add("project-homepages", f"/sites/project-homepages/resource/{r['id']}", r.get("title", ""), (dm if dm else 15) * 60) except Exception: pass try: # SoundShelf podcast episodes (audio-flavored stream) for e in db.query("podcasts-audiobooks", "episodes", sort="id", limit=200): add("podcasts-audiobooks", f"/sites/podcasts-audiobooks/episode/{e['id']}", e.get("title", ""), (e.get("duration_minutes") or 0) * 60, flavor="audio") except Exception: pass try: # SoundWave tracks (playbar key is track-identity, not page path) albums = {a["id"]: a for a in db.query("music", "albums", limit=500)} for t in db.query("music", "tracks", sort="id", limit=200): album = albums.get(t.get("album_id")) url = f"/sites/music/album/{album['id']}" if album else "/sites/music/" add("music", f"/sites/music/track/{t['id']}", t.get("title", ""), t.get("duration_seconds") or 210, url=url, flavor="audio") except Exception: pass try: # Course lectures (player key = course page path + lecture title) for course in db.query("course-sites-classrooms", "courses", limit=100): for mod in _j(course.get("modules"), []): for lesson in mod.get("lessons", []): if lesson.get("type") != "lecture": continue mins = lesson.get("duration_min") or 0 if not mins: mins = 8 + (len(lesson.get("title") or "") % 17) add("course-sites-classrooms", f"/sites/course-sites-classrooms/course/{course['id']}", "Lecture: " + (lesson.get("title") or ""), mins * 60) except Exception: pass return rows @annotation_bp.route("/playback") def playback_facts_page(): """Lookup table: which video hides what fact at what span.""" return render_template("playback_facts.html", rows=_collect_playback_facts()) @annotation_bp.route("/api/playback_facts") def api_playback_facts(): return jsonify(_collect_playback_facts()) @annotation_bp.route("/macro-browser") def macro_browser_page(): """Step through every task's sub-trajectory for one macro (or one task's macros).""" return render_template("macro_browser.html") @annotation_bp.route("/grader-audit") def grader_audit_page(): """Blind labeling of agent episodes for the grader audit (src/miniweb/annotation/grader_audit.py).""" return render_template("grader_audit.html") @annotation_bp.route("/api/grader_audit/cases") def api_grader_audit_cases(): from miniweb.annotation import grader_audit me = session.get("annotator_name", "anonymous") ids, overlap = grader_audit.order() mine = grader_audit.labels(me) return jsonify({"labeler": me, "cases": [{"id": c, "overlap": c in overlap, "labeled": c in mine} for c in ids]}) @annotation_bp.route("/api/grader_audit/case/") def api_grader_audit_case(case_id): from miniweb.annotation import grader_audit c = grader_audit.case(case_id) if c is None: return jsonify({"error": "unknown case"}), 404 c["my_label"] = grader_audit.labels(session.get("annotator_name", "anonymous")).get(case_id) return jsonify(c) @annotation_bp.route("/api/grader_audit/img//") def api_grader_audit_img(case_id, name): import re from flask import send_from_directory from miniweb.annotation import grader_audit d = grader_audit.case_dir(case_id) if d is None or not re.fullmatch(r"s\d{2,3}\.jpg", name): return "not found", 404 return send_from_directory(str(d), name, max_age=86400) @annotation_bp.route("/api/grader_audit/label", methods=["POST"]) def api_grader_audit_label(): from miniweb.annotation import grader_audit data = request.get_json(silent=True) or {} try: rec = grader_audit.save(session.get("annotator_name", "anonymous"), data.get("case_id"), data.get("primitives"), data.get("notes", "")) except ValueError as e: return jsonify({"error": str(e)}), 400 return jsonify({"status": "ok", "label": rec}) @annotation_bp.route("/api/grader_audit/labels.json") def api_grader_audit_labels(): from miniweb.annotation import grader_audit return jsonify(grader_audit.all_labels()) @annotation_bp.route("/task-audit") def task_audit_page(): """A reviewer's pass over every final task, its demonstration and the review's changes (miniweb/annotation/task_audit.py).""" return render_template("task_audit.html") @annotation_bp.route("/api/task_audit/tasks") def api_task_audit_tasks(): from miniweb.annotation import task_audit mine = task_audit.labels(session.get("annotator_name", "anonymous")) rows = [{k: r[k] for k in ("task_id", "site", "split", "status", "changed_fields", "verifier_repaired", "rerecorded", "walk")} | {"labeled": r["task_id"] in mine} for r in task_audit.entries()] return jsonify({"labeler": session.get("annotator_name", "anonymous"), "tasks": rows}) @annotation_bp.route("/api/task_audit/task/") def api_task_audit_task(task_id): from miniweb.annotation import task_audit c = task_audit.case(task_id) if c is None: return jsonify({"error": "unknown task"}), 404 c["my_label"] = task_audit.labels(session.get("annotator_name", "anonymous")).get(task_id) return jsonify(c) @annotation_bp.route("/api/task_audit/snapshot///") def api_task_audit_snapshot(task_id, source, index): """A step's page as recorded (HTML), shown in a sandboxed frame for recordings without screenshots; scripts are stripped and blocked.""" import re from flask import make_response from miniweb.annotation import task_audit html = task_audit.snapshot(task_id, source, index) if html is None: return "No page snapshot was recorded for this step.", 404 resp = make_response(re.sub(r"(?is)", "", html)) resp.headers["Content-Type"] = "text/html; charset=utf-8" resp.headers["Content-Security-Policy"] = "script-src 'none'; object-src 'none'" return resp @annotation_bp.route("/api/task_audit/label", methods=["POST"]) def api_task_audit_label(): from miniweb.annotation import task_audit data = request.get_json(silent=True) or {} try: rec = task_audit.save(session.get("annotator_name", "anonymous"), data.get("task_id"), data.get("task"), data.get("changes"), data.get("notes", "")) except ValueError as e: return jsonify({"error": str(e)}), 400 return jsonify({"status": "ok", "label": rec}) @annotation_bp.route("/api/task_audit/labels.json") def api_task_audit_labels(): from miniweb.annotation import task_audit return jsonify(task_audit.all_labels()) @annotation_bp.route("/api/macro_browser/index") def api_macro_browser_index(): from miniweb.annotation import macro_browser return jsonify(macro_browser.index()) @annotation_bp.route("/api/macro_browser/task//") def api_macro_browser_task(annotator, task_id): from miniweb.annotation import macro_browser source = "walk" if request.args.get("source") == "walk" else "gold" detail = macro_browser.task_detail(annotator, task_id, source) if detail is None: return jsonify({"error": "Task not found"}), 404 return jsonify(detail) @annotation_bp.route("/task-review") def task_review_page(): """One-screen task review: queues, before/now comparison, recording, verifier, decisions.""" return render_template("task_review.html", reviewer=session.get("annotator_name", "anonymous")) @annotation_bp.route("/api/task_review/index") def api_task_review_index(): from miniweb.annotation import review_mode return jsonify(review_mode.index()) @annotation_bp.route("/api/task_review/task//") def api_task_review_task(annotator, task_id): from miniweb.annotation import review_mode detail = review_mode.detail(annotator, task_id) if detail is None: return jsonify({"error": "Task not found"}), 404 return jsonify(detail) @annotation_bp.route("/api/infer_macros", methods=["POST"]) def api_infer_macros(): """Reverse annotation: the annotator writes the task in natural language FIRST, and the LLM decomposes it into ordered macro steps (base macro + optional reasoning op + a concrete subtask per step) — the inverse of draft_task (macros -> instruction). The UI loads the steps onto the graph canvas as a sequential chain for the annotator to correct. Body: {instruction: str, sites?: [site ids]} Returns {mode: "llm", steps: [{macro, operation, subtask}], unknown: [..]} where `unknown` lists model-suggested names not in the registry (dropped). """ import json as _json import re as _re data = request.get_json(silent=True) or {} instruction = (data.get("instruction") or "").strip() if len(instruction) < 10: return jsonify({"error": "instruction required (>= 10 chars)"}), 400 site_ids = [s["id"] if isinstance(s, dict) else s for s in (data.get("sites") or [])] ops = _registry.operations() catalog = "\n".join(f"{name}: {meta.get('description', '')}" for name, meta in _MACRO_DESCRIPTIONS.items()) ops_text = "\n".join(f"{op}: {(info or {}).get('desc', '')}" for op, info in ops.items()) # Per-site availability biases the model toward macros that actually exist # on the sampled sites; it may still pick outside — the annotator corrects. site_avail = "" if site_ids: locs = _canonical_macro_locations() parts = [f"{sid}: {', '.join(sorted((locs.get(sid) or {}).keys()))}" for sid in site_ids if locs.get(sid)] if parts: site_avail = ("\n\nMacros known to exist on the task's sites " "(prefer these):\n" + "\n".join(parts)) system = ( "You decompose a web task (written in natural language by an annotator) " "into an ORDERED sequence of macro steps.\n" "A step = a base macro (a physical UI interaction from the catalog) plus " "an OPTIONAL reasoning operation, attached only when the step reads or " "derives a value.\n" "Rules:\n" "- Use ONLY macro names from the catalog and ONLY the listed operations.\n" "- One step per distinct interaction, in execution order.\n" "- If the task requires reporting a value/answer back to the human, the " "final step is report_information with the fitting operation.\n" "- Intermediate reasoning is NOT its own step: attach its operation to " "the macro it happens on.\n" "- For each step write a short CONCRETE subtask: what to do on the page, " "with the specific values/entities from the instruction.\n" 'Reply ONLY JSON: {"steps": [{"macro": "", ' '"operation": "", "subtask": ""}]}') user = (f"Task instruction:\n{instruction}\n\nMacro catalog:\n{catalog}" f"\n\nReasoning operations:\n{ops_text}{site_avail}") from miniweb.app.llm import call_llm raw = call_llm(user, system=system, max_tokens=700, temperature=0.2, json_mode=True) if not raw: return jsonify({"error": "LLM unavailable"}), 503 try: parsed = _json.loads(raw) except (ValueError, TypeError): m = _re.search(r"\{.*\}", raw, _re.DOTALL) try: parsed = _json.loads(m.group()) if m else None except (ValueError, TypeError): parsed = None if not isinstance(parsed, dict): return jsonify({"error": "unparseable LLM response"}), 502 steps, unknown = [], [] for s in (parsed.get("steps") or []): if not isinstance(s, dict): continue name = str(s.get("macro") or "").strip() if not name: continue if not _registry.is_known(name): unknown.append(name) continue op = str(s.get("operation") or "").strip() if op and not _registry.is_operation(op): op = "" steps.append({"macro": _registry.canon(name), "operation": op, "subtask": str(s.get("subtask") or "").strip()}) if not steps: return jsonify({"error": "no usable steps inferred", "unknown": unknown}), 502 return jsonify({"mode": "llm", "steps": steps, "unknown": unknown}) @annotation_bp.route("/api/llm_models") def api_llm_models(): """Supported models per provider, with configuration status.""" from miniweb.app.llm import list_models, DEFAULT_MODEL, _get_env return jsonify({ "default": _get_env("LLM_MODEL") or DEFAULT_MODEL, "providers": list_models(), }) @annotation_bp.route("/api/make_ambiguous", methods=["POST"]) def api_make_ambiguous(): """Rewrite an instruction to be more ambiguous using LLM.""" data = request.get_json(silent=True) or {} instruction = data.get("instruction", "").strip() if not instruction: return jsonify({"error": "No instruction provided"}), 400 # Rewrite VOICE ONLY. The task already has a recorded trajectory and an # expected outcome, so the rewrite must not change what the agent has to do: # every name, id, amount and date stays. What changes is how the request # sounds — a colleague asking a favour (the house style, modelled on the # dataset's strongest instructions) instead of a spec that walks the agent # through the UI. system_prompt = ( "You rephrase a web-task instruction so it sounds like a real person asking " "a colleague for a favour, instead of a spec.\n\n" "DIFFICULTY MUST NOT CHANGE. The task already has a recorded trajectory and " "a verifier — the agent must end up doing exactly the same work.\n" "- KEEP every specific: names, emails, IDs, amounts, dates, filenames, " "credentials, text to type, the sort order, the filter values.\n" "- Do NOT replace a name with a description ('Priya Sharma' must NOT become " "'the person who ran the deploy') — that adds a lookup the original task " "never asked for.\n" "- Do NOT add or remove any step.\n" "- If the instruction names a required mechanism ('using the slider'), keep it: " "the verifier checks it.\n\n" "WHAT YOU MAY CHANGE — only the wording:\n" "- Drop UI hand-holding: 'go to the History page', 'click the Filter button', " "'open the dropdown'. Say what is wanted, not which widget to press.\n" "- Make it conversational: how someone would actually ask. A short reason for " "the request is welcome if it does not add work.\n" "- 1-3 sentences.\n\n" "The house style (real examples):\n" " 'Hey, what course is Dr. Balazinska teaching? Can you give me the official " "course name?'\n" " \"Let's start to file a new claim - can you select auto for the policy?\"\n" " 'I want to talk with the person running the Architecture Sync before-hand to " "ask them some questions. Can you make a new calendar event, then invite whoever " "owns that meeting for 30 minutes before that meeting starts.'\n" " (note: the last one keeps its own indirection because the ORIGINAL task was " "written that way — do not invent indirection that was not there.)\n\n" "Output ONLY the rephrased instruction, nothing else." ) result = _call_llm(system_prompt, instruction) if result: return jsonify({"ambiguous": result}) return jsonify({"error": "LLM unavailable"}), 503 @annotation_bp.route("/api/suggest_tags", methods=["POST"]) def api_suggest_tags(): """Use LLM to suggest macro span tags for a trajectory.""" data = request.get_json(silent=True) or {} task_id = data.get("task_id") annotator = data.get("annotator", "anonymous") if not task_id: return jsonify({"error": "task_id required"}), 400 # Support inline mode (from annotate UI) or saved task mode (from verify UI) if data.get("_inline"): macros = data.get("macros", []) actions = data.get("actions", []) instruction = data.get("instruction", "") else: from miniweb.annotation.storage import load_task task = load_task(annotator, task_id) if not task: return jsonify({"error": "Task not found"}), 404 macros = task.get("macros", []) trajectory = task.get("trajectory", []) instruction = task.get("instruction", "") # Build a compact action list for the LLM actions = [] for i, e in enumerate(trajectory): if e.get("type") == "action": verb = e.get("action", "") target = (e.get("target", ""))[:60] val = e.get("text", e.get("value", "")) val_str = f' "{str(val)[:30]}"' if val else "" actions.append(f"{len(actions) + 1}. {verb} {target}{val_str}") if not actions or not macros: return jsonify({"error": "No actions or macros to tag"}), 400 system_prompt = ( "You are tagging a web interaction trajectory with macro labels.\n\n" "Each macro is a primitive skill (like filter_by_slider, create_by_form, etc.).\n" "A span is [start_index, end_index] (inclusive, 1-indexed) indicating which actions belong to that macro.\n" "Actions are numbered starting from 1.\n\n" "Rules:\n" "- Every macro must get exactly one span\n" "- Spans can overlap if a macro encompasses sub-macros (e.g., extract_by_extremum covers the full flow)\n" "- Navigation actions (clicking nav links) at the start should be untagged\n" "- Include the submit/confirm action at the end of form-based macros\n" "- For QA macros (extract/compute) with no visible actions, use the span where the answer is visible\n\n" "Output ONLY a JSON object mapping macro names to [start, end] spans. Example:\n" '{"filter_by_dropdown": [1, 5], "extract_by_route": [6, 10]}' ) user_prompt = ( f"Task instruction: {instruction}\n\n" f"Macros to tag: {macros}\n\n" f"Action trajectory:\n" + "\n".join(actions) ) result = _call_llm(system_prompt, user_prompt) if not result: return jsonify({"error": "LLM unavailable"}), 503 try: # Parse JSON from response import re json_match = re.search(r'\{[^}]+\}', result) if json_match: spans = json.loads(json_match.group()) return jsonify({"suggested_spans": spans}) return jsonify({"error": "Could not parse LLM response", "raw": result}), 500 except Exception as e: return jsonify({"error": str(e), "raw": result}), 500 @annotation_bp.route("/api/build_verifiers", methods=["POST"]) def api_build_verifiers(): """Build verifier configs: deterministic checks + LLM for expected_outcome. 1. Auto-generate grounding, action, and state checks from the trajectory 2. If expected_outcome is provided, call GPT-5.5 to parse it into extra checks 3. If feedback + existing_configs, call GPT-5.5 to fix the existing configs 4. Auto-run all checks and return results """ from flask import current_app data = request.get_json(silent=True) or {} client = current_app.test_client() for key, val in request.cookies.items(): client.set_cookie(key, val) # 1. Fetch request log and enrich trajectory log_resp = client.get("/_admin/log") log_data = log_resp.get_json() if log_resp.status_code == 200 else {} request_log = log_data.get("entries", []) if isinstance(log_data, dict) else log_data trajectory = data.get("trajectory", []) enriched = _enrich_trajectory(trajectory, request_log) macro_spans = data.get("macro_spans", {}) expected_answer = data.get("expected_answer", "") expected_outcome = data.get("expected_outcome", "") feedback = data.get("feedback", "") existing_configs = data.get("existing_configs") # 2. Gather admin API schema per site schemas = {} sites_list = data.get("sites", []) for site in sites_list: sid = site["id"] if isinstance(site, dict) else site r = client.get(f"/_admin/files/{sid}") if r.status_code == 200: schemas[sid] = r.get_json() # 3. If feedback + existing configs → fix mode via LLM if feedback and existing_configs: configs = _llm_fix_verifiers(existing_configs, feedback, schemas) if not configs: configs = existing_configs # fallback: keep originals else: # 4. Auto-generate deterministic checks per macro span configs = _auto_generate_verifiers(enriched, macro_spans, expected_answer) # 5. If expected_outcome provided, augment with LLM-parsed checks if expected_outcome: extra = _llm_parse_outcome(expected_outcome, macro_spans, schemas, enriched) if extra: _merge_llm_checks(configs, extra) # 6. Auto-run from miniweb.annotation.evaluators import run_task_eval passed, results = run_task_eval( configs, server_url=request.host_url.rstrip("/"), flask_client=client, trajectory=trajectory, agent_answer=expected_answer, ) return jsonify({"configs": configs, "results": results, "passed": passed}) def _auto_generate_verifiers(enriched_trajectory, macro_spans, expected_answer): """Deterministically generate verifier configs from trajectory data.""" configs = [] macro_names = list(macro_spans.keys()) for macro, span in macro_spans.items(): if not span or len(span) < 2: continue start, end = span checks = [] # Extract actions and observations in this span action_idx = 0 span_urls = set() span_actions = [] span_api_calls = [] for entry in enriched_trajectory: action_idx += 1 if action_idx < start or action_idx > end: continue url = entry.get("url", "") if url and url.startswith("/sites/"): span_urls.add(url.split("?")[0]) # strip query params span_actions.append(entry) for call in entry.get("api_calls", []): span_api_calls.append(call) # Grounding: require visited URLs if span_urls: checks.append({ "type": "grounding", "required_urls": sorted(span_urls), "description": f"Visited {len(span_urls)} page(s) during {macro}", }) # Action checks: verify key actions occurred for act in span_actions: action_type = act.get("action", "") target = act.get("target", "") if action_type in ("click", "type", "select", "navigate", "tab_switch"): checks.append({ "type": "action_performed", "action": action_type, "target_contains": target[:50] if target else "", "description": f"{action_type} on {target[:40]}", }) break # one key action per macro is enough # State checks from API calls (POST/PUT/DELETE = mutations) for call in span_api_calls: method = call.get("method", "") path = call.get("path", "") if method in ("POST", "PUT", "DELETE") and "/api/" in path: # Convert site API path to admin path # /sites/forums/api/posts -> /_admin/data/forums/posts parts = path.split("/api/") if len(parts) == 2: site_part = parts[0].replace("/sites/", "") collection = parts[1].split("/")[0].split("?")[0] admin_endpoint = f"/_admin/data/{site_part}/{collection}" if method == "POST": checks.append({ "type": "state_query", "endpoint": admin_endpoint, "params": {}, "check": "not_empty", "description": f"Records exist in {site_part}/{collection} after {method}", }) elif method == "DELETE": body = call.get("body", {}) item_id = body.get("id", "") if item_id: checks.append({ "type": "record_absent", "endpoint": admin_endpoint, "params": {"id": str(item_id)}, "description": f"Record {item_id} deleted from {collection}", }) break # one state check per macro # Answer match: if this is the last macro and answer is provided if expected_answer and macro == macro_names[-1]: match_type = "number" if expected_answer.replace(".", "").replace("-", "").isdigit() else "string" checks.append({ "type": "answer_match", "expected": expected_answer, "match_type": match_type, "description": f"Answer matches '{expected_answer}'", }) configs.append({ "macro": macro, "span": span, "checks": checks, }) return configs def _llm_parse_outcome(expected_outcome, macro_spans, schemas, trajectory): """Call LLM to parse expected_outcome text into extra check configs.""" schema_summary = {} for site, colls in schemas.items(): if isinstance(colls, list): schema_summary[site] = colls[:15] # Extract API calls from trajectory to show LLM what endpoints exist api_examples = [] for entry in trajectory: for call in entry.get("api_calls", []): path = call.get("path", "") if "/api/" in path: example = {"method": call.get("method"), "path": path, "status": call.get("status")} if call.get("body"): example["body_fields"] = list(call["body"].keys()) if isinstance(call["body"], dict) else [] if call.get("response") and isinstance(call["response"], list) and call["response"]: example["response_fields"] = list(call["response"][0].keys()) if isinstance(call["response"][0], dict) else [] elif call.get("response") and isinstance(call["response"], dict): example["response_fields"] = list(call["response"].keys()) api_examples.append(example) if len(api_examples) >= 10: break prompt = f"""Parse this expected outcome into verifier check configs for a web benchmark. Expected outcome (written by annotator): {expected_outcome} Macro spans (which macro each action range belongs to): {json.dumps(macro_spans)} Available admin API collections per site: {json.dumps(schema_summary)} The admin API endpoint format is: /_admin/data// It accepts query params to filter (e.g. ?user_id=1&category=Groceries) and returns a JSON array of records. API calls observed in the trajectory: {json.dumps(api_examples, indent=2)} For each assertion in the expected outcome, output a JSON object with ALL required fields: {{ "macro": "which_macro_this_check_belongs_to (must match a key in macro_spans)", "check": {{ "type": "state_query", "endpoint": "/_admin/data//", "params": {{"field": "value"}}, "check": "contains|count_equals|count_gt|equals|not_empty", "expected": "the expected value", "description": "human-readable description" }} }} Other valid check types: - record_exists: {{"type":"record_exists", "endpoint":"/_admin/data/...", "params":{{}}, "description":"..."}} - record_absent: {{"type":"record_absent", "endpoint":"/_admin/data/...", "params":{{}}, "description":"..."}} - count_equals: {{"type":"count_equals", "endpoint":"/_admin/data/...", "params":{{}}, "expected":5, "description":"..."}} IMPORTANT RULES: - Every check MUST have "endpoint" with a valid /_admin/data// path - Use the admin API collections list and trajectory API calls to determine correct site and collection names - The "macro" field MUST exactly match one of the keys in macro_spans - Each assertion should go to the macro that is responsible for it - Only generate checks for things the annotator explicitly mentioned — do not invent extra checks - For "contains" checks, use {{"check": "contains", "expected": "the text to find"}} - For record existence, use params to filter (e.g. {{"body": "15:34"}}) based on what was in the API call Output a JSON array of these objects. If nothing to add, output []. Output ONLY the JSON array, no explanation.""" result = _call_openai_simple("gpt-4.1-nano", prompt, max_tokens=1500) if not result: return None try: text = result.strip() if text.startswith("```"): text = text.split("```")[1] if text.startswith("json"): text = text[4:] return json.loads(text.strip()) except (json.JSONDecodeError, IndexError): return None def _llm_fix_verifiers(existing_configs, feedback, schemas): """Call GPT-5.5 to fix existing verifier configs based on feedback.""" schema_summary = {} for site, colls in schemas.items(): if isinstance(colls, list): schema_summary[site] = colls[:15] prompt = f"""Fix these verifier configs based on the annotator's feedback. Current configs: {json.dumps(existing_configs, indent=2)} Feedback: {feedback} Available admin API collections: {json.dumps(schema_summary)} Apply ONLY the changes described in the feedback. Keep everything else unchanged. Output the full corrected JSON array. Output ONLY the JSON array.""" result = _call_openai_simple("gpt-4.1-nano", prompt, max_tokens=2000) if not result: return None try: text = result.strip() if text.startswith("```"): text = text.split("```")[1] if text.startswith("json"): text = text[4:] return json.loads(text.strip()) except (json.JSONDecodeError, IndexError): return None def _merge_llm_checks(configs, extra_checks): """Merge LLM-generated checks into existing macro configs.""" for item in extra_checks: macro = item.get("macro", "") check = item.get("check", {}) if not macro or not check: continue # Find the matching macro config target = next((c for c in configs if c["macro"] == macro), None) if target: target["checks"].append(check) elif configs: # Append to last config if macro not found configs[-1]["checks"].append(check) def _call_openai_simple(model, prompt, max_tokens=1500): """LLM call via shared Groq/OpenAI/Gemini helper.""" from miniweb.app.llm import call_llm try: return call_llm(prompt, max_tokens=max_tokens, temperature=0.2) except Exception: import traceback traceback.print_exc() return None _verifier_processes = {} # job_id -> subprocess.Popen @annotation_bp.route("/api/verifier_status/") def api_verifier_status(job_id): """Check if a verifier builder job is done.""" import pathlib as _pl result_file = _pl.Path(__file__).resolve().parent / "verifier_jobs" / f"{job_id}_verifiers.json" # Check subprocess status proc = _verifier_processes.get(job_id) if proc and proc.poll() is None: return jsonify({"status": "building", "ready": False}) if not result_file.exists(): return jsonify({"status": "not_found", "ready": False}) try: configs = json.loads(result_file.read_text()) if not isinstance(configs, list): return jsonify({"status": "invalid_output", "ready": False, "error": "LLM output was not a JSON array"}) except (json.JSONDecodeError, OSError) as e: return jsonify({"status": "error", "ready": False, "error": str(e)}) return jsonify({"status": "done", "ready": True, "configs": configs}) @annotation_bp.route("/api/load_verifiers", methods=["POST"]) def api_load_verifiers(): """Load verifier configs from file (written by the verifier builder).""" from flask import current_app data = request.get_json(silent=True) or {} result_file = data.get("result_file", "") if not result_file: return jsonify({"error": "No result_file specified"}), 400 import pathlib as _pl path = _pl.Path(result_file) if not path.exists(): return jsonify({"error": "File not found. Build verifiers first.", "ready": False}) try: configs = json.loads(path.read_text()) except (json.JSONDecodeError, OSError) as e: return jsonify({"error": f"Failed to read file: {e}", "ready": False}) # Auto-run the loaded configs client = current_app.test_client() for key, val in request.cookies.items(): client.set_cookie(key, val) from miniweb.annotation.evaluators import run_task_eval passed, results = run_task_eval( configs, server_url=request.host_url.rstrip("/"), flask_client=client, trajectory=data.get("trajectory", []), agent_answer=data.get("expected_answer", ""), ) return jsonify({"configs": configs, "results": results, "passed": passed, "ready": True}) # --------------------------------------------------------------------------- # Verification walk — the annotator walks the task a SECOND time in the verify # panel, producing an independent trajectory of the same task. # # Two trajectories are worth more than one: whatever appears in both is # load-bearing (the checks a verifier should assert), whatever appears in only # one is incidental (a stray scroll, a fumbled field, a different route to the # same page). A second walk that cannot reproduce the outcome also means the # task is broken against current data — which is how silent drift gets caught. # --------------------------------------------------------------------------- def _load_saved_task(annotator, task_id): """Load (dir, task.json, trajectory.json) for a task; searches all annotators.""" from miniweb.annotation.storage import ANNOTATIONS_DIR pattern = f"{annotator}/{task_id}" if annotator else f"*/{task_id}" for d in ANNOTATIONS_DIR.glob(pattern): task = json.loads((d / "task.json").read_text()) tf = d / "trajectory.json" traj = json.loads(tf.read_text()) if tf.exists() else [] return d, task, traj return None, None, [] def _action_key(a): """Identity of an action for cross-walk comparison. Uses verb + semantic target + value. Deliberately NOT the selector: recorded selectors are often bare tags ("a", "select") with no discriminating power, while `target` carries the readable description. """ val = a.get("value") or a.get("text") or a.get("option_text") or "" target = (a.get("target") or "").strip().lower() return (a.get("action", ""), target, str(val).strip().lower()) @annotation_bp.route("/api/verification_walk", methods=["POST"]) def api_verification_walk(): """Save a second walk of a task and compare it to the original. Body: {task_id, annotator?, trajectory: [...], answer?} Writes verification_walk.json next to the task and returns the comparison. """ data = request.get_json(silent=True) or {} task_id = data.get("task_id") if not task_id: return jsonify({"error": "task_id required"}), 400 d, task, original = _load_saved_task(data.get("annotator"), task_id) if task is None: return jsonify({"error": "task not found"}), 404 walk = data.get("trajectory") or [] walk_actions = [e for e in walk if e.get("type") == "action"] orig_actions = [e for e in original if e.get("type") == "action"] orig_keys = [_action_key(a) for a in orig_actions] walk_keys = [_action_key(a) for a in walk_actions] orig_set, walk_set = set(orig_keys), set(walk_keys) common = [k for k in orig_keys if k in walk_set] seen = set() common = [k for k in common if not (k in seen or seen.add(k))] walker = session.get("annotator_name", "anonymous") payload = { "task_id": task_id, "answer": data.get("answer", ""), "recorded_at": datetime.now().isoformat(), "recorded_by": walker, "trajectory": walk, } (d / "verification_walk.json").write_text(json.dumps(payload, indent=1)) task["verification_walk_by"] = {"by": walker, "at": payload["recorded_at"]} (d / "task.json").write_text(json.dumps(task, indent=2, default=str)) def fmt(keys): return [{"action": k[0], "target": k[1][:60], "value": k[2][:30]} for k in keys] return jsonify({ "task_id": task_id, "original_actions": len(orig_actions), "walk_actions": len(walk_actions), # in BOTH walks -> essential; a verifier should assert these "in_both": fmt(common), # in one walk only -> incidental; asserting these would fail a valid path "original_only": fmt([k for k in dict.fromkeys(orig_keys) if k not in walk_set]), "walk_only": fmt([k for k in dict.fromkeys(walk_keys) if k not in orig_set]), "answer_original": task.get("expected_answer", ""), "answer_walk": data.get("answer", ""), "answers_agree": _norm_answer(task.get("expected_answer", "")) == _norm_answer(data.get("answer", "")) if data.get("answer") else None, }) def _norm_answer(s): import re as _re s = str(s or "").lower() return " ".join(_re.sub(r"[^a-z0-9\s.]+", " ", s).split()) @annotation_bp.route("/api/task_status", methods=["POST"]) def api_task_status(): """Approve or reject a task after review. Body: {task_id, annotator?, status: "approved"|"rejected", note?} Records the decision on task.json — rejected tasks stay on disk (the recording is still evidence) but are excluded from eval task loading. """ data = request.get_json(silent=True) or {} task_id = data.get("task_id") status = data.get("status") if status not in ("approved", "rejected", "pending"): return jsonify({"error": "status must be approved, rejected or pending"}), 400 d, task, _ = _load_saved_task(data.get("annotator"), task_id) if task is None: return jsonify({"error": "task not found"}), 404 task["review"] = { "status": status, "note": data.get("note", ""), "reviewer": session.get("annotator_name", "anonymous"), "reviewed_at": datetime.now().isoformat(), } (d / "task.json").write_text(json.dumps(task, indent=2, default=str)) return jsonify({"task_id": task_id, "review": task["review"]}) @annotation_bp.route("/api/run_verifiers", methods=["POST"]) def api_run_verifiers(): """Execute verifier configs — uses server request log for agent-agnostic checks.""" from flask import current_app data = request.get_json(silent=True) or {} configs = data.get("configs", []) client = current_app.test_client() for key, val in request.cookies.items(): client.set_cookie(key, val) # Fetch both request log AND action beacon log request_log = [] action_log = [] try: log_resp = client.get("/_admin/log") if log_resp.status_code == 200: request_log = log_resp.get_json().get("entries", []) beacon_resp = client.get("/_admin/beacon") if beacon_resp.status_code == 200: action_log = beacon_resp.get_json().get("entries", []) except Exception: pass from miniweb.annotation.evaluators import run_task_eval passed, results = run_task_eval( configs, server_url=request.host_url.rstrip("/"), flask_client=client, trajectory=data.get("trajectory", []), agent_answer=data.get("expected_answer", ""), request_log=request_log, action_log=action_log, ) return jsonify({"passed": passed, "results": results}) @annotation_bp.route("/api/agent_validate", methods=["POST"]) def api_agent_validate(): """Run a browser-use agent with headless Chrome to attempt the task. Spawns a real browser session, navigates the site, and uses ChatLLM (Groq/OpenAI/Gemini) as the LLM to decide actions. After the agent finishes, we check the server request log + agent's answer against the verifiers. """ import asyncio import sys from flask import current_app data = request.get_json(silent=True) or {} instruction = data.get("instruction", "") configs = data.get("configs", []) sites = data.get("sites", []) if not instruction: return jsonify({"error": "Missing instruction"}), 400 site_ids = [s["id"] if isinstance(s, dict) else s for s in sites] base = request.host_url.rstrip("/") site_url = f"{base}/sites/{site_ids[0]}" if site_ids else base # Clear request log before agent runs client = current_app.test_client() for key, val in request.cookies.items(): client.set_cookie(key, val) client.post("/_admin/log/clear") # Run browser-use agent agent_answer = "" agent_steps = 0 agent_errors = [] agent_response = "" agent_cookies = {} # capture agent's session cookies for verifier async def _run_agent(): nonlocal agent_answer, agent_steps, agent_errors, agent_response, agent_cookies # Import browser-use components # Ensure browser_use is findable (installed in ~/.local) import site as _site _user_site = str(Path.home() / ".local/lib/python3.11/site-packages") if _user_site not in sys.path: sys.path.insert(0, _user_site) from miniweb.evaluation.agents import ChatLLM from browser_use import Agent, BrowserSession llm = ChatLLM() browser_session = BrowserSession(headless=True) try: await browser_session.start() page = await browser_session.get_current_page() await page.goto(site_url) await asyncio.sleep(2) task_prompt = ( f"You are on a web application at {site_url}. " f"Your task: {instruction}" ) agent = Agent( task=task_prompt, llm=llm, browser_session=browser_session, use_vision=False, max_steps=15, ) history = await asyncio.wait_for(agent.run(), timeout=120) agent_steps = len(history.history) agent_answer = history.final_result() or "" agent_errors = history.errors() or [] agent_response = f"Completed in {agent_steps} steps" # Capture agent's cookies so verifiers use the agent's session try: cookies = await page.context.cookies() for c in cookies: agent_cookies[c["name"]] = c["value"] except Exception: pass except asyncio.TimeoutError: agent_errors = ["Agent timed out after 120s"] agent_response = "Timed out" except Exception as e: agent_errors = [str(e)] agent_response = f"Error: {e}" finally: try: await browser_session.kill() except Exception: pass # Run the async agent in a new event loop try: loop = asyncio.new_event_loop() loop.run_until_complete(_run_agent()) loop.close() except Exception as e: agent_errors = [str(e)] agent_response = f"Loop error: {e}" # Create a new test client with the AGENT's cookies (isolated session) agent_client = current_app.test_client() for name, value in agent_cookies.items(): agent_client.set_cookie(name, value) # Fetch request log using agent's session request_log = [] try: log_resp = agent_client.get("/_admin/log") if log_resp.status_code == 200: request_log = log_resp.get_json().get("entries", []) except Exception: pass # Run verifiers against agent's session state from miniweb.annotation.evaluators import run_task_eval agent_passed, agent_results = run_task_eval( configs, flask_client=agent_client, trajectory=[], agent_answer=agent_answer, request_log=request_log, ) return jsonify({ "agent_passed": agent_passed, "agent_results": agent_results, "agent_answer": agent_answer or "(no answer)", "agent_response": agent_response[:500], "agent_steps": agent_steps, "agent_errors": agent_errors[:3], "request_count": len(request_log), }) @annotation_bp.route("/api/perturb_verifiers", methods=["POST"]) def api_perturb_verifiers(): """Run verifiers against 2 perturbed versions to validate they fail correctly. Perturbation 1: Wrong answer — swap expected answer to something different. Perturbation 2: Missing actions — remove key actions from the trajectory. Returns the original + 2 perturbed results with diffs highlighted. """ from flask import current_app data = request.get_json(silent=True) or {} configs = data.get("configs", []) traj = data.get("trajectory", []) answer = data.get("expected_answer", "") client = current_app.test_client() for key, val in request.cookies.items(): client.set_cookie(key, val) from miniweb.annotation.evaluators import run_task_eval # Original (should pass) orig_passed, orig_results = run_task_eval( configs, flask_client=client, trajectory=traj, agent_answer=answer) # Perturbation 1: Wrong answer wrong_answer = _perturb_answer(answer) p1_passed, p1_results = run_task_eval( configs, flask_client=client, trajectory=traj, agent_answer=wrong_answer) # Perturbation 2: Remove actions — drop half the actions from trajectory instruction = data.get("instruction", "") thin_traj = _perturb_trajectory(traj, instruction) p2_passed, p2_results = run_task_eval( configs, flask_client=client, trajectory=thin_traj, agent_answer=answer) # Build trajectory summaries for diff display def _traj_summary(t): out = [] for e in t: if e.get("type") != "action": continue a = e.get("action", "") target = e.get("target", "") if a == "type" and e.get("text"): label = f'{a} "{e["text"][:25]}"' elif a == "select" and e.get("option_text"): label = f'{a} "{e["option_text"]}"' elif a == "tab_switch": label = f'switch to {e.get("to_tab", "")}' else: label = f'{a} {target[:30]}' entry = {"label": label, "perturbed": bool(e.get("_perturbed"))} if e.get("_original"): entry["original"] = e["_original"] if e.get("_changed_to"): entry["changed_to"] = e["_changed_to"] out.append(entry) return out return jsonify({ "original": {"passed": orig_passed, "results": orig_results, "answer": answer, "actions": _count_actions(traj), "trajectory": _traj_summary(traj)}, "perturbation_1": {"passed": p1_passed, "results": p1_results, "label": "Wrong answer", "answer": wrong_answer, "trajectory": _traj_summary(traj), "diff": f'"{answer}" \u2192 "{wrong_answer}"'}, "perturbation_2": {"passed": p2_passed, "results": p2_results, "label": "Missing actions", "actions": _count_actions(thin_traj), "trajectory": _traj_summary(thin_traj), "diff": f'{_count_actions(traj)} \u2192 {_count_actions(thin_traj)} actions'}, }) def _perturb_answer(answer): """Generate a plausible but wrong answer using LLM.""" if not answer: return "wrong_answer" prompt = f"""Given this correct answer to a web task: "{answer}" Generate a plausible but WRONG answer that: - Is the same type (number if number, name if name, etc.) - Looks realistic (not obviously garbage) - Is clearly different from the correct answer Output ONLY the wrong answer, nothing else.""" result = _call_openai_simple("gpt-4.1-nano", prompt, max_tokens=50) if result: return result.strip().strip('"').strip("'") # Fallback if no API key try: n = float(answer) return str(int(n + 3) if n == int(n) else round(n + 2.5, 2)) except (ValueError, TypeError): pass return answer[::-1] if len(answer) > 1 else "X" def _perturb_trajectory(traj, instruction=""): """Generate a subtly wrong trajectory using LLM — keeps structure but changes key actions.""" # Build a summary of the trajectory for the LLM actions = [] for e in traj: if e.get("type") != "action": continue a = e.get("action", "") t = e.get("target", "") detail = "" if a == "type" and e.get("text"): detail = f': "{e["text"][:30]}"' elif a == "select" and e.get("option_text"): detail = f': "{e["option_text"]}"' elif a == "tab_switch": detail = f': {e.get("to_tab", "")}' actions.append(f'{a} {t[:40]}{detail}') if len(actions) < 2: # Too short to perturb meaningfully return [e for e in traj if e.get("type") == "observation" or e.get("action") in ("navigate", "tab_switch")] actions_text = "\n".join(f"{i+1}. {a}" for i, a in enumerate(actions)) prompt = f"""Here is a sequence of browser actions for a web task: Task: {instruction[:200] if instruction else '(not provided)'} Actions: {actions_text} Create a WRONG version of this action sequence that would fail the task. Make subtle but meaningful changes: - Change a dropdown selection to a wrong value - Type a different search query - Click a different button - Navigate to a wrong page - Skip a critical step Keep 60-80% of actions the same. Change 2-3 key actions. Output the modified sequence as numbered lines in the same format. Output ONLY the list.""" result = _call_openai_simple("gpt-4.1-nano", prompt, max_tokens=500) if result: # Parse the LLM output back into trajectory entries perturbed = [] lines = [l.strip() for l in result.strip().split("\n") if l.strip()] orig_actions = [e for e in traj if e.get("type") == "action"] for i, line in enumerate(lines): # Remove numbering line = line.lstrip("0123456789. ").strip() if i < len(orig_actions): entry = dict(orig_actions[i]) # Check if LLM changed this action orig_desc = actions[i] if i < len(actions) else "" if line.lower() != orig_desc.lower(): entry["_perturbed"] = True entry["_original"] = orig_desc entry["_changed_to"] = line perturbed.append(entry) else: perturbed.append({"type": "action", "action": "click", "target": line, "_perturbed": True, "_changed_to": line}) # Add observations back result_traj = [] act_idx = 0 for e in traj: if e.get("type") == "observation": result_traj.append(e) elif e.get("type") == "action": if act_idx < len(perturbed): result_traj.append(perturbed[act_idx]) act_idx += 1 return result_traj # Fallback: simple removal return [e for e in traj if e.get("type") == "observation" or e.get("action") in ("navigate", "tab_switch")] def _count_actions(traj): return sum(1 for e in traj if e.get("type") == "action") def _enrich_trajectory(trajectory, request_log): """Attach API calls from request log to each action by timestamp.""" actions = [e for e in trajectory if e.get("type") == "action"] if not actions or not request_log: return [ {**a, "api_calls": []} for a in actions ] enriched = [] log_idx = 0 for i, action in enumerate(actions): act_time = action.get("timestamp", "") next_time = actions[i + 1]["timestamp"] if i + 1 < len(actions) else "9999" api_calls = [] while log_idx < len(request_log): entry = request_log[log_idx] entry_time = entry.get("timestamp", "") if entry_time < act_time: log_idx += 1 continue if entry_time >= next_time: break api_calls.append(entry) log_idx += 1 enriched.append({**action, "api_calls": api_calls}) return enriched # --- Website Review --- # Override on deployments (e.g. Railway) to point at a mounted volume so # runtime-submitted reviews survive redeploys; defaults to the repo copy. _REVIEWS_DIR = Path( os.environ.get("MINIWEB_REVIEWS_DIR", str(SITES_DIR.parent / "data" / "reviews")) ).resolve() def _review_file(site_id): # site_id comes from the URL — sanitize to a plain name to keep paths safe safe = "".join(c for c in site_id if c.isalnum() or c in "-_") return _REVIEWS_DIR / f"{safe}.json" def _load_reviews(site_id): f = _review_file(site_id) if f.exists(): try: return json.loads(f.read_text()) except (json.JSONDecodeError, OSError): pass return [] @annotation_bp.route("/api/review/", methods=["POST"]) def api_submit_review(site_id): """Submit free-form feedback for a site. Stored in data/reviews/.json.""" data = request.get_json(silent=True) or {} feedback = (data.get("feedback") or "").strip() if not feedback: return jsonify({"error": "feedback required"}), 400 reviews = _load_reviews(site_id) reviews.append({ "annotator": data.get("annotator", "anonymous"), "feedback": feedback, "timestamp": datetime.now().isoformat(), }) _REVIEWS_DIR.mkdir(parents=True, exist_ok=True) _review_file(site_id).write_text(json.dumps(reviews, indent=1)) return jsonify({"status": "saved", "count": len(reviews)}) @annotation_bp.route("/api/review_status") def api_review_status(): """List sites with their review counts.""" sites = _load_sites() result = [] for s in sites: result.append({ "id": s["id"], "name": s.get("name", s["id"]), "review_count": len(_load_reviews(s["id"])), }) return jsonify(result) @annotation_bp.route("/api/reviews/") def api_get_reviews(site_id): """Get all reviews for a site.""" return jsonify(_load_reviews(site_id)) # Dashboard removed — use index page and verify page instead