[Haik]: removed more unused code, now gonna remove the prompt templates feature entirely

This commit is contained in:
haikdc
2026-03-30 22:37:44 -07:00
parent 8be1e48341
commit 32093e0149
10 changed files with 112 additions and 212 deletions
+1 -87
View File
@@ -9,7 +9,7 @@ from backend.apps.outputs.helpers import _validate_against_schema
from backend.apps.outputs.executor import execute_backend_code
from backend.apps.common.model_registry import resolve_model_id as _resolve_model
from backend.apps.outputs.models import (
VibeCodeRequest, AutoRunRequest, AutoRunAgentRequest,
AutoRunRequest, AutoRunAgentRequest,
)
logger = logging.getLogger(__name__)
@@ -21,92 +21,6 @@ def _get_anthropic_client():
return get_anthropic_client(load_settings())
VIBE_CODE_SYSTEM_PROMPT = """\
You are an expert at building self-contained HTML/JS/CSS applications that run in an iframe.
The user will describe what they want, and you will generate:
1. **frontend_code**: A complete HTML document. React 18 is available via esm.sh CDN.
- Use: <script type="importmap">{"imports":{"react":"https://esm.sh/react@18","react-dom/client":"https://esm.sh/react-dom@18/client"}}</script>
- Input data is at window.OUTPUT_INPUT (object), backend result at window.OUTPUT_BACKEND_RESULT.
2. **input_schema**: A JSON Schema object defining the structured input.
3. **backend_code** (optional): Python code where input_data is a global dict and result is a global dict to assign to.
4. **name**: A short name for the view.
5. **description**: A one-sentence description.
6. **message**: A brief explanation of what you did/changed.
Return ONLY valid JSON with these keys. No markdown fences, no extra text.\
"""
async def vibe_code(body: VibeCodeRequest):
from backend.apps.analytics.collector import record as _analytics
_analytics("feature.used", {"feature": "vibe_code.used"})
try:
import anthropic
except ImportError:
return {
"message": "anthropic SDK not installed. Install with: pip install anthropic",
"frontend_code": body.current_frontend_code,
"backend_code": body.current_backend_code,
"input_schema": body.current_schema,
}
context_parts = []
if body.current_frontend_code:
context_parts.append(f"Current frontend code:\n```html\n{body.current_frontend_code}\n```")
if body.current_backend_code:
context_parts.append(f"Current backend code:\n```python\n{body.current_backend_code}\n```")
if body.current_schema:
context_parts.append(f"Current input schema:\n```json\n{body.current_schema}\n```")
if body.name:
context_parts.append(f"Current name: {body.name}")
if body.description:
context_parts.append(f"Current description: {body.description}")
user_message = body.prompt
if context_parts:
user_message = "\n\n".join(context_parts) + "\n\nUser request: " + body.prompt
client = _get_anthropic_client()
from backend.apps.common.llm_helpers import _resolve_model as _resolve_9r
from backend.apps.settings.settings import load_settings as _ls
try:
resp = await client.messages.create(
model=_resolve_9r("claude-sonnet-4-20250514", _ls()), max_tokens=8000,
system=VIBE_CODE_SYSTEM_PROMPT,
messages=[{"role": "user", "content": user_message}],
)
raw = resp.content[0].text.strip()
if raw.startswith("```"):
raw = raw.split("\n", 1)[1] if "\n" in raw else raw[3:]
if raw.endswith("```"):
raw = raw[:-3]
result = json.loads(raw)
return {
"message": result.get("message", "View updated."),
"frontend_code": result.get("frontend_code", body.current_frontend_code),
"backend_code": result.get("backend_code", body.current_backend_code),
"input_schema": result.get("input_schema", body.current_schema),
"name": result.get("name", body.name),
"description": result.get("description", body.description),
}
except json.JSONDecodeError:
return {
"message": "I generated code but couldn't parse the response. Please try again.",
"frontend_code": body.current_frontend_code,
"backend_code": body.current_backend_code,
"input_schema": body.current_schema,
}
except Exception as e:
logger.exception("Vibe code generation failed")
return {
"message": f"Error: {str(e)}",
"frontend_code": body.current_frontend_code,
"backend_code": body.current_backend_code,
"input_schema": body.current_schema,
}
AUTO_RUN_SYSTEM_PROMPT = """\
You generate structured JSON data matching a given schema.
The user provides a prompt describing what data to generate and a JSON Schema.
+1 -10
View File
@@ -168,13 +168,4 @@ class WorkspaceSeedRequest(BaseModel):
@model_validator(mode="before")
@classmethod
def _migrate_flat_fields(cls, data: Any) -> Any:
return _migrate_legacy_files(data, allow_schema_json=True)
class VibeCodeRequest(BaseModel):
prompt: str
current_frontend_code: str = ""
current_backend_code: str = ""
current_schema: str = ""
name: str = ""
description: str = ""
return _migrate_legacy_files(data, allow_schema_json=True)
-1
View File
@@ -42,7 +42,6 @@ _store = JsonStore(Output, DATA_DIR, not_found_detail="Output not found")
_load_all = _store.load_all
_save = _store.save
_load = _store.load
load_output = _store.load_or_none
# -- File serving --
-98
View File
@@ -25,108 +25,10 @@ def _check_9router() -> bool:
except Exception:
return False
def validate_credentials(settings: AppSettings, provider: str = "anthropic") -> None:
"""Raise ValueError if credentials are missing for the given provider.
Allows through if 9Router is running as a fallback.
Handles both display names ('Anthropic') and lowercase ('anthropic').
"""
p = provider.lower().strip()
# 9Router or GitHub Copilot providers don't need traditional credentials
if p in ("9router", "github copilot", "copilot"):
return
# If 9Router is running, all providers are accessible
if _check_9router():
return
if p == "anthropic":
if getattr(settings, "connection_mode", "own_key") == "managed":
if not getattr(settings, "openswarm_auth_token", None):
raise ValueError("Open Swarm account not connected. Sign in via Settings -> API.")
return
if settings.anthropic_api_key:
return
raise ValueError("Anthropic API key not configured. Set it in Settings, or connect a subscription.")
elif p == "openai":
if settings.openai_api_key:
return
raise ValueError("OpenAI API key not configured. Set it in Settings, or connect a subscription.")
elif p in ("gemini", "google"):
if getattr(settings, "google_api_key", None):
return
raise ValueError("Google API key not configured. Set it in Settings, or connect a subscription.")
elif p == "openrouter":
if getattr(settings, "openrouter_api_key", None):
return
raise ValueError("OpenRouter API key not configured. Set it in Settings.")
elif p in ("xai", "meta", "deepseek", "mistral", "qwen", "cohere"):
# These route through OpenRouter — need either OpenRouter key or 9Router
if getattr(settings, "openrouter_api_key", None):
return
raise ValueError(f"{provider} requires an OpenRouter API key, or connect a subscription via 9Router.")
else:
# Custom provider — check if it exists in custom_providers
for cp in getattr(settings, "custom_providers", []):
if cp.name.lower() == p:
return
# Unknown provider — allow through (create_provider will handle the error)
return
def get_provider_credentials(settings: AppSettings, provider: str) -> dict[str, str]:
"""Return credential dict for a specific provider."""
p = provider.lower().strip()
validate_credentials(settings, provider)
if p in ("anthropic", "claude"):
if getattr(settings, "connection_mode", "own_key") == "managed":
return {
"auth_token": getattr(settings, "openswarm_auth_token", "") or "",
"base_url": getattr(settings, "openswarm_proxy_url", None) or OPENSWARM_DEFAULT_PROXY_URL,
}
return {"api_key": settings.anthropic_api_key or ""}
if p in ("openai", "codex"):
return {"api_key": settings.openai_api_key or ""}
if p in ("gemini", "google", "gemini-cli"):
return {"api_key": getattr(settings, "google_api_key", "") or ""}
if p == "openrouter":
return {"api_key": getattr(settings, "openrouter_api_key", "") or ""}
# Custom provider
for cp in getattr(settings, "custom_providers", []):
if cp.name.lower() == p:
return {"api_key": cp.api_key, "base_url": cp.base_url}
raise ValueError(f"No credentials for provider: {provider}")
# ---------------------------------------------------------------------------
# Legacy helpers (kept for backward compat during migration)
# ---------------------------------------------------------------------------
def get_agent_sdk_env(settings: AppSettings) -> dict[str, str]:
"""Return the env dict for ClaudeAgentOptions based on connection mode.
DEPRECATED: Use create_provider() from providers.registry instead.
"""
validate_credentials(settings, "anthropic")
if getattr(settings, "connection_mode", "own_key") == "managed":
proxy_url = getattr(settings, "openswarm_proxy_url", None) or OPENSWARM_DEFAULT_PROXY_URL
return {
"ANTHROPIC_AUTH_TOKEN": getattr(settings, "openswarm_auth_token", ""),
"ANTHROPIC_BASE_URL": proxy_url,
}
return {"ANTHROPIC_API_KEY": settings.anthropic_api_key}
def get_anthropic_client(settings: AppSettings) -> anthropic.AsyncAnthropic:
"""Return a configured AsyncAnthropic client based on connection mode.
+8 -8
View File
@@ -20,6 +20,13 @@ DEFAULT_SYSTEM_PROMPT = (
)
class CustomProvider(BaseModel):
name: str
base_url: str
api_key: str = ""
models: list[dict[str, Any]] = Field(default_factory=list)
class AppSettings(BaseModel):
default_system_prompt: Optional[str] = DEFAULT_SYSTEM_PROMPT
default_folder: Optional[str] = None
@@ -35,7 +42,7 @@ class AppSettings(BaseModel):
openai_api_key: Optional[str] = None
google_api_key: Optional[str] = None
openrouter_api_key: Optional[str] = None
custom_providers: list["CustomProvider"] = Field(default_factory=list)
custom_providers: list[CustomProvider] = Field(default_factory=list)
# Dashboard / UI preferences
auto_select_mode_on_new_agent: bool = False
expand_new_chats_in_dashboard: bool = True
@@ -53,10 +60,3 @@ class AppSettings(BaseModel):
analytics_opt_in: bool = True
installation_id: Optional[str] = None
first_opened_at: Optional[str] = None # ISO timestamp of first app open
class CustomProvider(BaseModel):
name: str
base_url: str
api_key: str = ""
models: list[dict[str, Any]] = Field(default_factory=list)
-1
View File
@@ -2,7 +2,6 @@ import os
import json
import logging
import re
from pathlib import Path
from contextlib import asynccontextmanager
from fastapi import HTTPException
from backend.config.Apps import SubApp
+58
View File
@@ -0,0 +1,58 @@
"""Class-level dead code detection with framework awareness.
Pydantic BaseModel subclasses are auto-whitelisted: every annotated field is
part of the serialization schema and therefore intentionally "used".
Non-framework classes are skipped for now (tier 2 — future cross-referencing).
"""
from __future__ import annotations
import ast
from pathlib import Path
from . import is_excepted, is_excluded
FRAMEWORK_BASES = {"BaseModel"}
def _is_framework_model(cls: ast.ClassDef) -> bool:
return any(
(isinstance(b, ast.Name) and b.id in FRAMEWORK_BASES)
or (isinstance(b, ast.Attribute) and b.attr in FRAMEWORK_BASES)
for b in cls.bases
)
def run_class_check(
root: Path,
exceptions: dict[str, list[str]],
excludes: list[str],
) -> list[str]:
"""Analyse classes in backend Python files and return errors."""
errors: list[str] = []
backend = root / "backend"
if not backend.is_dir():
return errors
for pyfile in sorted(backend.rglob("*.py")):
if is_excluded(pyfile, root, excludes):
continue
rel = str(pyfile.relative_to(root))
if is_excepted(rel, "classes", exceptions):
continue
try:
source = pyfile.read_text()
tree = ast.parse(source, filename=rel)
except (OSError, SyntaxError):
continue
for node in ast.walk(tree):
if not isinstance(node, ast.ClassDef):
continue
if _is_framework_model(node):
continue
# Tier 2 placeholder: non-framework classes are skipped until
# cross-reference analysis is implemented.
return errors
+34 -2
View File
@@ -1,10 +1,16 @@
"""Vulture dead-code detection runner."""
"""Vulture dead-code detection runner.
Class-body findings (fields, methods inside a class) are filtered out here
and handled separately by checks/classes.py which understands Pydantic.
"""
from __future__ import annotations
import ast
import re
import shutil
import subprocess
from functools import lru_cache
from pathlib import Path
from . import is_excepted
@@ -12,6 +18,30 @@ from . import is_excepted
CONFIG_DIR = Path(__file__).resolve().parent.parent / "config"
@lru_cache(maxsize=64)
def _class_line_ranges(filepath: str) -> list[tuple[int, int]]:
"""Return (start, end) line ranges for all class bodies in *filepath*."""
try:
tree = ast.parse(Path(filepath).read_text())
except (OSError, SyntaxError):
return []
ranges: list[tuple[int, int]] = []
for node in ast.walk(tree):
if isinstance(node, ast.ClassDef):
end = max(getattr(n, "lineno", node.lineno) for n in ast.walk(node))
ranges.append((node.lineno, end))
return ranges
def _is_inside_class(filepath: str, lineno: int) -> bool:
"""True when *lineno* is strictly inside a class body.
The class declaration line itself (``class Foo:``) is *not* considered
inside, so vulture's "unused class" findings still pass through.
"""
return any(start < lineno <= end for start, end in _class_line_ranges(filepath))
def run_vulture(
root: Path, min_confidence: int, error_threshold: int,
exceptions: dict[str, list[str]],
@@ -34,7 +64,7 @@ def run_vulture(
cmd.extend([
"--min-confidence", str(min_confidence),
"--exclude", ".venv,__pycache__,data,uv-bin",
"--ignore-decorators", "@*.router.*",
"--ignore-decorators", "@*.router.*,@*.websocket,@pytest.fixture,@pytest.fixture*",
"--ignore-names", "cls",
])
@@ -53,6 +83,8 @@ def run_vulture(
filepath, lineno, message = m.groups()
if is_excepted(filepath, "vulture", exceptions):
continue
if _is_inside_class(str(root / filepath), int(lineno)):
continue
conf = re.search(r"\((\d+)% confidence\)", message)
confidence = int(conf.group(1)) if conf else 0
severity = "error" if confidence >= error_threshold else "warning"
+4 -2
View File
@@ -6,7 +6,8 @@
"vulture": true,
"eslint": false,
"knip": false,
"endpoints": true
"endpoints": true,
"classes": true
},
"rules": {
"max-file-lines": 250,
@@ -41,6 +42,7 @@
],
"no-nested-imports": ["linter/lint.py"],
"vulture": [],
"endpoints": ["backend/apps/health/*"]
"endpoints": ["backend/apps/health/*"],
"classes": []
}
}
+6 -3
View File
@@ -16,6 +16,7 @@ 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
from checks.classes import run_class_check
SCRIPT_DIR = Path(__file__).resolve().parent
CONFIG_FILE = SCRIPT_DIR / "config" / "config.json"
@@ -26,7 +27,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], list[str]]:
def run_checks(root: Path) -> tuple[list[str], 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"]
@@ -83,8 +84,9 @@ def run_checks(root: Path) -> tuple[list[str], list[str], list[str], list[str],
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 []
class_errors = run_class_check(root, exceptions, excludes) if enabled.get("classes", True) else []
return sorted(structural_errors), sorted(vulture_errors), sorted(eslint_errors), sorted(knip_errors), sorted(endpoint_errors)
return sorted(structural_errors), sorted(vulture_errors), sorted(eslint_errors), sorted(knip_errors), sorted(endpoint_errors), sorted(class_errors)
def _print_section(name: str, errors: list[str]) -> None:
@@ -97,13 +99,14 @@ 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],
endpoint_errors: list[str], class_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)
_print_section("classes", class_errors)
def watch_loop(root: Path) -> None: