[Haik]: removed unused endpoints, also made linter compatible with endpoint parsing

This commit is contained in:
haikdc
2026-03-30 22:00:36 -07:00
parent 23945ad079
commit 8be1e48341
10 changed files with 223 additions and 108 deletions
-10
View File
@@ -123,16 +123,6 @@ async def update_session(session_id: str, body: dict):
await agent_manager.update_session(session_id, **body)
return {"ok": True}
@agents.router.get("/sessions/{session_id}/branches")
async def get_branches(session_id: str):
session = agent_manager.get_session(session_id)
if not session:
raise HTTPException(status_code=404, detail="Session not found")
return {
"branches": {k: v.model_dump(mode="json") for k, v in session.branches.items()},
"active_branch_id": session.active_branch_id,
}
@agents.router.post("/sessions/{session_id}/duplicate")
async def duplicate_session(session_id: str, body: dict = {}):
try:
+1 -39
View File
@@ -141,42 +141,4 @@ async def usage_summary():
stats = compute_session_stats(sessions)
nine_router_stats = await get_usage_stats() if _9r_running() else None
return enrich_with_nine_router(stats, nine_router_stats)
@analytics.router.get("/cost-breakdown")
async def cost_breakdown(period: str = "7d"):
"""Get detailed cost breakdown from 9Router."""
from backend.apps.nine_router import get_usage_stats, is_running as _9r_running
if not _9r_running():
return {"available": False, "by_model": {}, "by_provider": {}}
stats = await get_usage_stats(period)
if not stats:
return {"available": False, "by_model": {}, "by_provider": {}}
return {
"available": True,
"period": period,
"total_cost": stats.get("totalCost", 0),
"total_requests": stats.get("totalRequests", 0),
"total_prompt_tokens": stats.get("totalPromptTokens", 0),
"total_completion_tokens": stats.get("totalCompletionTokens", 0),
"by_model": stats.get("byModel", {}),
"by_provider": stats.get("byProvider", {}),
}
@analytics.router.get("/status")
async def analytics_status():
return {"status": "posthog", "enabled": True}
@analytics.router.post("/event")
async def record_event(body: dict):
"""Accept analytics events from the frontend (e.g. feature.time_spent)."""
event_type = body.get("event_type", "")
properties = body.get("properties", {})
if event_type:
record(event_type, properties,
session_id=body.get("session_id"),
dashboard_id=body.get("dashboard_id"))
return {"ok": True}
return enrich_with_nine_router(stats, nine_router_stats)
-1
View File
@@ -230,7 +230,6 @@ async def execute_output(body: OutputExecute):
# -- AI generation routes --
outputs.router.add_api_route("/vibe-code", ai_generation.vibe_code, methods=["POST"])
outputs.router.add_api_route("/auto-run", ai_generation.auto_run_output, methods=["POST"])
outputs.router.add_api_route("/auto-run-agent", ai_generation.auto_run_agent, methods=["POST"])
outputs.router.add_api_route("/auto-run-agent/{session_id}", ai_generation.cleanup_auto_run_agent, methods=["DELETE"])
-5
View File
@@ -91,11 +91,6 @@ async def update_settings(body: AppSettings):
return {"ok": True, "settings": body.model_dump()}
@settings.router.get("/default-system-prompt")
async def get_default_system_prompt():
return {"default_system_prompt": DEFAULT_SYSTEM_PROMPT}
@settings.router.post("/reset-system-prompt")
async def reset_system_prompt():
current = load_settings()
@@ -95,40 +95,6 @@ async def subscriptions_poll(body: dict):
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@subscriptions.router.post("/exchange")
async def subscriptions_exchange(body: dict):
"""Exchange OAuth code for tokens via 9Router."""
from backend.apps.nine_router import exchange_oauth
provider = body.get("provider", "")
code = body.get("code", "")
redirect_uri = body.get("redirect_uri", "")
code_verifier = body.get("code_verifier", "")
state = body.get("state", "")
if not provider or not code:
raise HTTPException(status_code=400, detail="provider and code required")
try:
result = await exchange_oauth(provider, code, redirect_uri, code_verifier, state)
if result.get("success"):
from backend.apps.analytics.collector import record as _analytics
_analytics("subscription.connected", {"provider": provider})
return result
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@subscriptions.router.get("/models")
async def subscriptions_models():
"""List all models available through connected subscriptions."""
from backend.apps.nine_router import is_running, get_models
if not is_running():
return {"models": []}
models = await get_models()
return {"models": models}
@subscriptions.router.post("/disconnect")
async def subscriptions_disconnect(body: dict):
"""Disconnect a subscription provider via 9Router."""
-14
View File
@@ -60,17 +60,3 @@ async def update_template(template_id: str, body: PromptTemplateUpdate):
async def delete_template(template_id: str):
_delete(template_id)
return {"ok": True}
@templates.router.post("/render")
async def render_template(body: dict):
template_id = body.get("template_id", "")
values = body.get("values", {})
template = _load(template_id)
rendered = template.template
for field in template.fields:
placeholder = "{{" + field.name + "}}"
value = values.get(field.name, field.default or "")
rendered = rendered.replace(placeholder, str(value))
from backend.apps.analytics.collector import record as _analytics
_analytics("feature.used", {"feature": "template.used"})
return {"rendered": rendered}
+207
View File
@@ -0,0 +1,207 @@
"""Orphaned endpoint detection — cross-references backend routes with usage.
Extracts all registered API routes from the backend (decorator and add_api_route
patterns) and checks whether each route's static path segments appear in the
frontend source or in other backend files (e.g. MCP servers that call endpoints
internally). Routes with no matching reference anywhere are flagged.
Limitations (v1):
- Routes that end with a path parameter (e.g. /{id}) and have no trailing
static segment are skipped — they're too ambiguous to match.
- WebSocket routes in main.py are not checked.
- Backend-only endpoints (health checks, OAuth callbacks) should be excluded
via the exceptions list or endpoint-ignore-routes in config.json.
"""
from __future__ import annotations
import fnmatch
import re
from pathlib import Path
from . import is_excepted
_DECORATOR_RE = re.compile(
r"@(\w+)\.router\.\w+\(\s*[\"']([^\"']+)[\"']"
)
_ADD_ROUTE_RE = re.compile(
r"(\w+)\.router\.add_api_route\(\s*[\"']([^\"']+)[\"']"
)
_SUBAPP_RE = re.compile(
r"(\w+)\s*=\s*SubApp\(\s*[\"']([^\"']+)[\"']"
)
_FUNC_DEF_RE = re.compile(r"\s*(?:async\s+)?def\s+(\w+)")
_ADD_ROUTE_FUNC_RE = re.compile(r"add_api_route\([^,]+,\s*(?:\w+\.)*(\w+)")
_TEMPLATE_ASSIGN_RE = re.compile(
r"""(?:export\s+)?(?:const|let|var)\s+(\w+)\s*=\s*`([^`]*)`"""
)
_STRING_ASSIGN_RE = re.compile(
r"""(?:export\s+)?(?:const|let|var)\s+(\w+)\s*=\s*(['"])(.*?)\2"""
)
_TEMPLATE_REF_RE = re.compile(r"\$\{(\w+)\}")
def _static_tail(route_path: str) -> str:
"""Return the trailing contiguous static segments of a route path.
>>> _static_tail("/sessions/{id}/message")
'/message'
>>> _static_tail("/usage-summary")
'/usage-summary'
>>> _static_tail("/{id}")
''
"""
parts = route_path.strip("/").split("/")
tail: list[str] = []
for part in reversed(parts):
if part.startswith("{"):
break
tail.append(part)
tail.reverse()
return "/" + "/".join(tail) if tail else ""
def _resolve_frontend_vars(files: list[tuple[str, str]]) -> dict[str, str]:
"""Collect const/let/var string assignments across files and resolve refs.
Handles patterns like:
const API_BASE = "/api";
const WORKSPACE_API = `${API_BASE}/outputs/workspace`;
"""
raw: dict[str, str] = {}
for _, text in files:
for m in _STRING_ASSIGN_RE.finditer(text):
raw.setdefault(m.group(1), m.group(3))
for m in _TEMPLATE_ASSIGN_RE.finditer(text):
raw.setdefault(m.group(1), m.group(2))
resolved = dict(raw)
for _ in range(5):
changed = False
for name, val in list(resolved.items()):
new_val = _TEMPLATE_REF_RE.sub(
lambda m: resolved.get(m.group(1), m.group(0)), val
)
if new_val != val:
resolved[name] = new_val
changed = True
if not changed:
break
return resolved
def _expand_template_refs(text: str, resolved: dict[str, str]) -> str:
"""Replace ``${VAR}`` references in *text* with resolved values."""
return _TEMPLATE_REF_RE.sub(
lambda m: resolved.get(m.group(1), m.group(0)), text
)
def _find_func_name(lines: list[str], decorator_idx: int) -> str:
for j in range(decorator_idx + 1, min(decorator_idx + 5, len(lines))):
m = _FUNC_DEF_RE.match(lines[j])
if m:
return m.group(1)
return ""
def run_endpoint_check(
root: Path,
exceptions: dict[str, list[str]],
ignore_routes: list[str] | None = None,
) -> list[str]:
"""Find backend API endpoints with no matching frontend or backend reference."""
backend_dir = root / "backend"
frontend_dir = root / "frontend" / "src"
if not backend_dir.exists() or not frontend_dir.exists():
return []
_ignore_routes = ignore_routes or []
var_to_name: dict[str, str] = {}
for py in backend_dir.rglob("*.py"):
if ".venv" in py.parts:
continue
for m in _SUBAPP_RE.finditer(py.read_text(errors="ignore")):
var_to_name[m.group(1)] = m.group(2)
routes: list[tuple[str, str, str, int, str]] = []
for py in backend_dir.rglob("*.py"):
if ".venv" in py.parts:
continue
text = py.read_text(errors="ignore")
lines = text.splitlines()
rel = str(py.relative_to(root))
for i, line in enumerate(lines):
m = _DECORATOR_RE.search(line)
if m:
var, path = m.group(1), m.group(2)
name = var_to_name.get(var)
if name:
func = _find_func_name(lines, i)
routes.append((name, path, rel, i + 1, func))
m2 = _ADD_ROUTE_RE.search(line)
if m2:
var, path = m2.group(1), m2.group(2)
name = var_to_name.get(var)
if name:
fm = _ADD_ROUTE_FUNC_RE.search(line)
func = fm.group(1) if fm else ""
routes.append((name, path, rel, i + 1, func))
frontend_files: list[tuple[str, str]] = []
for ext in ("*.ts", "*.tsx"):
for f in frontend_dir.rglob(ext):
frontend_files.append((str(f.relative_to(root)), f.read_text(errors="ignore")))
backend_files: list[tuple[str, str]] = []
for py in backend_dir.rglob("*.py"):
if ".venv" in py.parts:
continue
backend_files.append((str(py.relative_to(root)), py.read_text(errors="ignore")))
resolved_vars = _resolve_frontend_vars(frontend_files)
errors: list[str] = []
for subapp_name, route_path, filepath, lineno, func_name in routes:
if is_excepted(filepath, "endpoints", exceptions):
continue
full_path = f"{subapp_name}{route_path}"
if any(fnmatch.fnmatch(full_path, p) for p in _ignore_routes):
continue
tail = _static_tail(route_path)
if not tail:
continue
found = False
for _fe_path, fe_text in frontend_files:
expanded = _expand_template_refs(fe_text, resolved_vars)
if full_path in expanded:
found = True
break
if subapp_name in expanded and tail in expanded:
found = True
break
if not found:
for be_path, be_text in backend_files:
if be_path == filepath:
continue
if full_path in be_text:
found = True
break
if not found:
label = func_name or route_path
errors.append(
f"{filepath}:{lineno}:1: warning: "
f"[endpoints] orphaned endpoint '{label}' "
f"(/api/{full_path}) — no frontend or backend reference found"
)
return sorted(errors)
+2
View File
@@ -34,6 +34,8 @@ def run_vulture(
cmd.extend([
"--min-confidence", str(min_confidence),
"--exclude", ".venv,__pycache__,data,uv-bin",
"--ignore-decorators", "@*.router.*",
"--ignore-names", "cls",
])
try:
+6 -3
View File
@@ -5,14 +5,16 @@
"no-nested-imports": false,
"vulture": true,
"eslint": false,
"knip": false
"knip": false,
"endpoints": true
},
"rules": {
"max-file-lines": 250,
"max-folder-items": 7,
"vulture-min-confidence": 1,
"vulture-error-threshold": 1,
"no-nested-imports": true
"no-nested-imports": true,
"endpoint-ignore-routes": ["*/callback", "*/callback/*"]
},
"include_extensions": [".py", ".ts", ".tsx", ".js", ".jsx"],
"exclude": [
@@ -38,6 +40,7 @@
"backend"
],
"no-nested-imports": ["linter/lint.py"],
"vulture": []
"vulture": [],
"endpoints": ["backend/apps/health/*"]
}
}
+7 -2
View File
@@ -15,6 +15,7 @@ from checks.structural import check_file_lines, check_folder_items, check_nested
from checks.vulture import run_vulture
from checks.eslint import run_eslint
from checks.knip import run_knip
from checks.endpoints import run_endpoint_check
SCRIPT_DIR = Path(__file__).resolve().parent
CONFIG_FILE = SCRIPT_DIR / "config" / "config.json"
@@ -25,7 +26,7 @@ def load_config() -> dict[str, Any]:
return json.load(f)
def run_checks(root: Path) -> tuple[list[str], list[str], list[str], list[str]]:
def run_checks(root: Path) -> tuple[list[str], list[str], list[str], list[str], list[str]]:
config = load_config()
enabled: dict[str, bool] = config.get("enabled", {})
rules: dict[str, int] = config["rules"]
@@ -80,8 +81,10 @@ def run_checks(root: Path) -> tuple[list[str], list[str], list[str], list[str]]:
eslint_errors = run_eslint(root) if enabled.get("eslint", True) else []
knip_errors = run_knip(root) if enabled.get("knip", True) else []
endpoint_ignore_routes: list[str] = rules.get("endpoint-ignore-routes", [])
endpoint_errors = run_endpoint_check(root, exceptions, endpoint_ignore_routes) if enabled.get("endpoints", True) else []
return sorted(structural_errors), sorted(vulture_errors), sorted(eslint_errors), sorted(knip_errors)
return sorted(structural_errors), sorted(vulture_errors), sorted(eslint_errors), sorted(knip_errors), sorted(endpoint_errors)
def _print_section(name: str, errors: list[str]) -> None:
@@ -94,11 +97,13 @@ def _print_section(name: str, errors: list[str]) -> None:
def print_results(
structural_errors: list[str], vulture_errors: list[str],
eslint_errors: list[str], knip_errors: list[str],
endpoint_errors: list[str],
) -> None:
_print_section("structural", structural_errors)
_print_section("vulture", vulture_errors)
_print_section("eslint", eslint_errors)
_print_section("knip", knip_errors)
_print_section("endpoints", endpoint_errors)
def watch_loop(root: Path) -> None: