feat: backend servis iskeletini ve yönetim uçlarını ekle

backend/README.md

@@ -0,0 +1,17 @@
+# WiseClaw Backend
+
+FastAPI service for WiseClaw. The backend now includes:
+
+- SQLite persistence through SQLAlchemy
+- runtime/admin settings endpoints
+- Ollama integration status endpoint
+- Telegram polling runtime scaffold
+
+## Run locally
+
+```bash
+python3.12 -m venv .venv312
+source .venv312/bin/activate
+pip install .
+uvicorn app.main:app --reload
+```

backend/app/__init__.py

@@ -0,0 +1 @@
+

backend/app/admin/__init__.py

@@ -0,0 +1 @@
+

backend/app/admin/routes.py

@@ -0,0 +1,78 @@
+from fastapi import APIRouter, Depends
+from pydantic import BaseModel
+from sqlalchemy.orm import Session
+
+from app.admin.services import AdminService
+from app.db import get_session
+from app.llm.ollama_client import OllamaClient
+from app.models import MemoryRecord, OllamaStatus, RuntimeSettings, TelegramStatus, UserRecord
+
+router = APIRouter(prefix="/admin", tags=["admin"])
+
+
+class SecretPayload(BaseModel):
+    key: str
+    value: str
+
+
+def get_admin_service(session: Session = Depends(get_session)) -> AdminService:
+    return AdminService(session)
+
+
+@router.get("/dashboard")
+def get_dashboard(service: AdminService = Depends(get_admin_service)):
+    return service.dashboard()
+
+
+@router.get("/settings", response_model=RuntimeSettings)
+def get_settings(service: AdminService = Depends(get_admin_service)):
+    return service.get_runtime_settings()
+
+
+@router.put("/settings", response_model=RuntimeSettings)
+def put_settings(payload: RuntimeSettings, service: AdminService = Depends(get_admin_service)):
+    return service.update_runtime_settings(payload)
+
+
+@router.get("/users", response_model=list[UserRecord])
+def get_users(service: AdminService = Depends(get_admin_service)):
+    return service.list_users()
+
+
+@router.post("/users", response_model=UserRecord)
+def post_user(payload: UserRecord, service: AdminService = Depends(get_admin_service)):
+    return service.save_user(payload)
+
+
+@router.get("/memory", response_model=list[MemoryRecord])
+def get_memory(service: AdminService = Depends(get_admin_service)):
+    return service.list_memory()
+
+
+@router.delete("/memory")
+def delete_memory(service: AdminService = Depends(get_admin_service)):
+    service.clear_memory()
+    return {"status": "ok"}
+
+
+@router.get("/secrets/{key}")
+def get_secret(key: str, service: AdminService = Depends(get_admin_service)):
+    return {"key": key, "masked": service.get_secret_mask(key)}
+
+
+@router.post("/secrets")
+def post_secret(payload: SecretPayload, service: AdminService = Depends(get_admin_service)):
+    service.save_secret(payload.key, payload.value)
+    return {"status": "ok"}
+
+
+@router.get("/integrations/ollama", response_model=OllamaStatus)
+async def get_ollama_status(service: AdminService = Depends(get_admin_service)):
+    runtime = service.get_runtime_settings()
+    client = OllamaClient(runtime.ollama_base_url)
+    return await client.status(runtime.default_model)
+
+
+@router.get("/integrations/telegram", response_model=TelegramStatus)
+def get_telegram_status(service: AdminService = Depends(get_admin_service)):
+    return service.telegram_status()

backend/app/admin/services.py

@@ -0,0 +1,142 @@
+from datetime import datetime
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from app.db import (
+    AuditLogORM,
+    AuthorizedUserORM,
+    MemoryItemORM,
+    SecretORM,
+    SettingORM,
+    ToolStateORM,
+    list_recent_logs,
+)
+from app.config import get_settings
+from app.models import DashboardSnapshot, MemoryRecord, RuntimeSettings, TelegramStatus, ToolToggle, UserRecord
+
+
+class AdminService:
+    def __init__(self, session: Session) -> None:
+        self.session = session
+
+    def get_runtime_settings(self) -> RuntimeSettings:
+        settings = {
+            item.key: item.value for item in self.session.scalars(select(SettingORM))
+        }
+        tools = list(self.session.scalars(select(ToolStateORM).order_by(ToolStateORM.name.asc())))
+        return RuntimeSettings(
+            terminal_mode=int(settings["terminal_mode"]),
+            search_provider=settings["search_provider"],
+            ollama_base_url=settings["ollama_base_url"],
+            default_model=settings["default_model"],
+            tools=[ToolToggle(name=tool.name, enabled=tool.enabled) for tool in tools],
+        )
+
+    def update_runtime_settings(self, payload: RuntimeSettings) -> RuntimeSettings:
+        self._save_setting("terminal_mode", str(payload.terminal_mode))
+        self._save_setting("search_provider", payload.search_provider)
+        self._save_setting("ollama_base_url", payload.ollama_base_url)
+        self._save_setting("default_model", payload.default_model)
+
+        for tool in payload.tools:
+            record = self.session.get(ToolStateORM, tool.name)
+            if record is None:
+                self.session.add(ToolStateORM(name=tool.name, enabled=tool.enabled, updated_at=datetime.utcnow()))
+            else:
+                record.enabled = tool.enabled
+                record.updated_at = datetime.utcnow()
+
+        self.session.add(AuditLogORM(category="settings", message="settings:runtime-updated"))
+        self.session.commit()
+        return self.get_runtime_settings()
+
+    def dashboard(self) -> DashboardSnapshot:
+        return DashboardSnapshot(
+            settings=self.get_runtime_settings(),
+            whitelist_count=self.session.scalar(select(func.count()).select_from(AuthorizedUserORM)) or 0,
+            memory_items=self.session.scalar(select(func.count()).select_from(MemoryItemORM)) or 0,
+            recent_logs=list_recent_logs(self.session, limit=10),
+        )
+
+    def list_users(self) -> list[UserRecord]:
+        stmt = select(AuthorizedUserORM).order_by(AuthorizedUserORM.created_at.desc())
+        return [
+            UserRecord(
+                telegram_user_id=user.telegram_user_id,
+                username=user.username,
+                display_name=user.display_name,
+                is_active=user.is_active,
+            )
+            for user in self.session.scalars(stmt)
+        ]
+
+    def save_user(self, user: UserRecord) -> UserRecord:
+        record = self.session.get(AuthorizedUserORM, user.telegram_user_id)
+        if record is None:
+            record = AuthorizedUserORM(
+                telegram_user_id=user.telegram_user_id,
+                username=user.username,
+                display_name=user.display_name,
+                is_active=user.is_active,
+                created_at=datetime.utcnow(),
+                updated_at=datetime.utcnow(),
+            )
+            self.session.add(record)
+        else:
+            record.username = user.username
+            record.display_name = user.display_name
+            record.is_active = user.is_active
+            record.updated_at = datetime.utcnow()
+        self.session.add(AuditLogORM(category="users", message=f"users:upsert:{user.telegram_user_id}"))
+        self.session.commit()
+        return user
+
+    def list_memory(self) -> list[MemoryRecord]:
+        stmt = select(MemoryItemORM).order_by(MemoryItemORM.created_at.desc(), MemoryItemORM.id.desc()).limit(50)
+        return [
+            MemoryRecord(id=item.id, content=item.content, kind=item.kind, created_at=item.created_at)
+            for item in self.session.scalars(stmt)
+        ]
+
+    def clear_memory(self) -> None:
+        for item in self.session.scalars(select(MemoryItemORM)):
+            self.session.delete(item)
+        self.session.add(AuditLogORM(category="memory", message="memory:cleared"))
+        self.session.commit()
+
+    def get_secret_mask(self, key: str) -> str:
+        record = self.session.get(SecretORM, key)
+        value = record.value if record else ""
+        if len(value) < 4:
+            return ""
+        return f"{value[:2]}***{value[-2:]}"
+
+    def save_secret(self, key: str, value: str) -> None:
+        record = self.session.get(SecretORM, key)
+        if record is None:
+            self.session.add(SecretORM(key=key, value=value, updated_at=datetime.utcnow()))
+        else:
+            record.value = value
+            record.updated_at = datetime.utcnow()
+        self.session.add(AuditLogORM(category="secrets", message=f"secrets:updated:{key}"))
+        self.session.commit()
+
+    def _save_setting(self, key: str, value: str) -> None:
+        record = self.session.get(SettingORM, key)
+        if record is None:
+            self.session.add(SettingORM(key=key, value=value, updated_at=datetime.utcnow()))
+        else:
+            record.value = value
+            record.updated_at = datetime.utcnow()
+
+    def telegram_status(self) -> TelegramStatus:
+        settings = get_settings()
+        configured = bool(settings.telegram_bot_token)
+        return TelegramStatus(
+            configured=configured,
+            polling_active=False,
+            message="Telegram token is configured. Polling starts when the backend boots."
+            if configured
+            else "Telegram token is not configured.",
+        )

backend/app/config.py

@@ -0,0 +1,28 @@
+from functools import lru_cache
+
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_prefix="WISECLAW_",
+        extra="ignore",
+    )
+
+    env: str = "development"
+    db_url: str = "sqlite:///./wiseclaw.db"
+    admin_host: str = "127.0.0.1"
+    admin_port: int = 8000
+    ollama_base_url: str = "http://127.0.0.1:11434"
+    default_model: str = "qwen3.5:4b"
+    search_provider: str = "brave"
+    telegram_bot_token: str = Field(default="", repr=False)
+    brave_api_key: str = Field(default="", repr=False)
+
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()
+

backend/app/db.py

@@ -0,0 +1,133 @@
+from collections.abc import Iterator
+from contextlib import contextmanager
+from datetime import datetime
+
+from sqlalchemy import Boolean, DateTime, Integer, String, Text, create_engine, select
+from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, sessionmaker
+
+from app.config import get_settings
+
+
+DEFAULT_SETTINGS = {
+    "terminal_mode": "3",
+    "search_provider": "brave",
+    "ollama_base_url": "http://127.0.0.1:11434",
+    "default_model": "qwen3.5:4b",
+}
+
+DEFAULT_TOOLS = {
+    "brave_search": True,
+    "searxng_search": False,
+    "web_fetch": True,
+    "apple_notes": True,
+    "files": True,
+    "terminal": True,
+}
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+class SettingORM(Base):
+    __tablename__ = "settings"
+
+    key: Mapped[str] = mapped_column(String(100), primary_key=True)
+    value: Mapped[str] = mapped_column(Text, nullable=False)
+    updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, nullable=False)
+
+
+class ToolStateORM(Base):
+    __tablename__ = "tool_states"
+
+    name: Mapped[str] = mapped_column(String(100), primary_key=True)
+    enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
+    updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, nullable=False)
+
+
+class AuthorizedUserORM(Base):
+    __tablename__ = "authorized_users"
+
+    telegram_user_id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    username: Mapped[str | None] = mapped_column(String(255))
+    display_name: Mapped[str | None] = mapped_column(String(255))
+    is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, nullable=False)
+    updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, nullable=False)
+
+
+class MemoryItemORM(Base):
+    __tablename__ = "memory_items"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    content: Mapped[str] = mapped_column(Text, nullable=False)
+    kind: Mapped[str] = mapped_column(String(50), nullable=False, default="message")
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, nullable=False)
+
+
+class AuditLogORM(Base):
+    __tablename__ = "audit_logs"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    category: Mapped[str] = mapped_column(String(50), nullable=False)
+    message: Mapped[str] = mapped_column(Text, nullable=False)
+    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, nullable=False)
+
+
+class SecretORM(Base):
+    __tablename__ = "secrets"
+
+    key: Mapped[str] = mapped_column(String(100), primary_key=True)
+    value: Mapped[str] = mapped_column(Text, nullable=False)
+    updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, nullable=False)
+
+
+settings = get_settings()
+engine = create_engine(
+    settings.db_url,
+    connect_args={"check_same_thread": False} if settings.db_url.startswith("sqlite") else {},
+)
+SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
+
+
+def init_db() -> None:
+    Base.metadata.create_all(bind=engine)
+    with session_scope() as session:
+        _seed_defaults(session)
+
+
+def _seed_defaults(session: Session) -> None:
+    for key, value in DEFAULT_SETTINGS.items():
+        if session.get(SettingORM, key) is None:
+            session.add(SettingORM(key=key, value=value))
+
+    for name, enabled in DEFAULT_TOOLS.items():
+        if session.get(ToolStateORM, name) is None:
+            session.add(ToolStateORM(name=name, enabled=enabled))
+
+
+def get_session() -> Iterator[Session]:
+    session = SessionLocal()
+    try:
+        yield session
+    finally:
+        session.close()
+
+
+@contextmanager
+def session_scope() -> Iterator[Session]:
+    session = SessionLocal()
+    try:
+        yield session
+        session.commit()
+    except Exception:
+        session.rollback()
+        raise
+    finally:
+        session.close()
+
+
+def list_recent_logs(session: Session, limit: int = 10) -> list[str]:
+    stmt = select(AuditLogORM).order_by(AuditLogORM.created_at.desc(), AuditLogORM.id.desc()).limit(limit)
+    return [row.message for row in session.scalars(stmt)]
+

backend/app/llm/__init__.py

@@ -0,0 +1 @@
+

backend/app/llm/ollama_client.py

@@ -0,0 +1,37 @@
+import httpx
+from httpx import HTTPError
+
+from app.models import OllamaStatus
+
+class OllamaClient:
+    def __init__(self, base_url: str) -> None:
+        self.base_url = base_url.rstrip("/")
+
+    async def health(self) -> bool:
+        async with httpx.AsyncClient(timeout=5.0) as client:
+            response = await client.get(f"{self.base_url}/api/tags")
+        return response.is_success
+
+    async def status(self, model: str) -> OllamaStatus:
+        try:
+            async with httpx.AsyncClient(timeout=5.0) as client:
+                response = await client.get(f"{self.base_url}/api/tags")
+            response.raise_for_status()
+        except HTTPError as exc:
+            return OllamaStatus(
+                reachable=False,
+                base_url=self.base_url,
+                model=model,
+                message=f"Ollama unreachable: {exc}",
+            )
+
+        payload = response.json()
+        installed_models = [item.get("name", "") for item in payload.get("models", []) if item.get("name")]
+        has_model = model in installed_models
+        return OllamaStatus(
+            reachable=True,
+            base_url=self.base_url,
+            model=model,
+            installed_models=installed_models,
+            message="Model found." if has_model else "Ollama reachable but model is not installed.",
+        )

backend/app/llm/planner.py

@@ -0,0 +1,15 @@
+from app.models import RuntimeSettings
+
+
+def build_prompt_context(message: str, runtime: RuntimeSettings, memory: list[str]) -> dict[str, object]:
+    return {
+        "system": (
+            "You are WiseClaw, a local-first assistant running on macOS. "
+            "Use tools carefully and obey terminal safety mode."
+        ),
+        "message": message,
+        "model": runtime.default_model,
+        "memory": memory,
+        "available_tools": [tool.name for tool in runtime.tools if tool.enabled],
+    }
+

backend/app/main.py

@@ -0,0 +1,56 @@
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from app.admin.routes import router as admin_router
+from app.config import get_settings
+from app.db import init_db, session_scope
+from app.models import HealthStatus
+from app.orchestrator import WiseClawOrchestrator
+from app.runtime import RuntimeServices
+from app.telegram.bot import TelegramBotService
+
+settings = get_settings()
+runtime_services = RuntimeServices()
+
+
+@asynccontextmanager
+async def lifespan(_: FastAPI):
+    init_db()
+    if settings.telegram_bot_token:
+        runtime_services.telegram_bot = TelegramBotService(settings.telegram_bot_token, session_scope)
+        await runtime_services.telegram_bot.start()
+    yield
+    await runtime_services.shutdown()
+
+
+app = FastAPI(title="WiseClaw", version="0.1.0", lifespan=lifespan)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["http://127.0.0.1:5173", "http://localhost:5173"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+app.include_router(admin_router)
+
+
+@app.get("/health", response_model=HealthStatus)
+def health() -> HealthStatus:
+    return HealthStatus()
+
+
+@app.get("/bootstrap")
+def bootstrap() -> dict[str, object]:
+    with session_scope() as session:
+        orchestrator = WiseClawOrchestrator(session)
+        runtime = orchestrator.get_runtime_settings().model_dump()
+        return {
+            "env": settings.env,
+            "admin_host": settings.admin_host,
+            "admin_port": settings.admin_port,
+            "runtime": runtime,
+        }

backend/app/memory/__init__.py

@@ -0,0 +1 @@
+

backend/app/memory/store.py

@@ -0,0 +1,17 @@
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.db import MemoryItemORM
+
+
+class MemoryService:
+    def __init__(self, session: Session) -> None:
+        self.session = session
+
+    def add_item(self, content: str) -> None:
+        self.session.add(MemoryItemORM(content=content, kind="message"))
+        self.session.flush()
+
+    def latest_items(self, limit: int = 10) -> list[str]:
+        stmt = select(MemoryItemORM).order_by(MemoryItemORM.created_at.desc(), MemoryItemORM.id.desc()).limit(limit)
+        return [item.content for item in self.session.scalars(stmt)]

backend/app/models.py

@@ -0,0 +1,77 @@
+from datetime import datetime
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+
+TerminalMode = Literal[1, 2, 3]
+SearchProvider = Literal["brave", "searxng"]
+
+
+class HealthStatus(BaseModel):
+    service: str = "wiseclaw"
+    status: str = "ok"
+    timestamp: datetime = Field(default_factory=datetime.utcnow)
+
+
+class SettingRecord(BaseModel):
+    key: str
+    value: str
+    updated_at: datetime = Field(default_factory=datetime.utcnow)
+
+
+class ToolToggle(BaseModel):
+    name: str
+    enabled: bool
+
+
+class UserRecord(BaseModel):
+    telegram_user_id: int
+    username: str | None = None
+    display_name: str | None = None
+    is_active: bool = True
+
+
+class RuntimeSettings(BaseModel):
+    terminal_mode: TerminalMode = 3
+    search_provider: SearchProvider = "brave"
+    ollama_base_url: str = "http://127.0.0.1:11434"
+    default_model: str = "qwen3.5:4b"
+    tools: list[ToolToggle] = Field(
+        default_factory=lambda: [
+            ToolToggle(name="brave_search", enabled=True),
+            ToolToggle(name="searxng_search", enabled=False),
+            ToolToggle(name="web_fetch", enabled=True),
+            ToolToggle(name="apple_notes", enabled=True),
+            ToolToggle(name="files", enabled=True),
+            ToolToggle(name="terminal", enabled=True),
+        ]
+    )
+
+
+class DashboardSnapshot(BaseModel):
+    settings: RuntimeSettings
+    whitelist_count: int
+    memory_items: int
+    recent_logs: list[str]
+
+
+class MemoryRecord(BaseModel):
+    id: int
+    content: str
+    kind: str
+    created_at: datetime
+
+
+class OllamaStatus(BaseModel):
+    reachable: bool
+    base_url: str
+    model: str
+    installed_models: list[str] = Field(default_factory=list)
+    message: str
+
+
+class TelegramStatus(BaseModel):
+    configured: bool
+    polling_active: bool
+    message: str

backend/app/orchestrator.py

@@ -0,0 +1,46 @@
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.db import AuditLogORM, SettingORM, ToolStateORM
+from app.llm.planner import build_prompt_context
+from app.memory.store import MemoryService
+from app.models import RuntimeSettings
+from app.telegram.auth import is_authorized
+
+
+class WiseClawOrchestrator:
+    def __init__(self, session: Session) -> None:
+        self.session = session
+        self.memory = MemoryService(session)
+
+    def get_runtime_settings(self) -> RuntimeSettings:
+        settings = {
+            item.key: item.value for item in self.session.scalars(select(SettingORM))
+        }
+        tools = list(self.session.scalars(select(ToolStateORM).order_by(ToolStateORM.name.asc())))
+        return RuntimeSettings(
+            terminal_mode=int(settings["terminal_mode"]),
+            search_provider=settings["search_provider"],
+            ollama_base_url=settings["ollama_base_url"],
+            default_model=settings["default_model"],
+            tools=[{"name": tool.name, "enabled": tool.enabled} for tool in tools],
+        )
+
+    def handle_text_message(self, telegram_user_id: int, text: str) -> str:
+        if not is_authorized(self.session, telegram_user_id):
+            return "This Telegram user is not authorized for WiseClaw."
+
+        self.memory.add_item(f"user:{telegram_user_id}:{text}")
+        context = build_prompt_context(
+            message=text,
+            runtime=self.get_runtime_settings(),
+            memory=self.memory.latest_items(limit=5),
+        )
+        response = (
+            "WiseClaw scaffold received your message.\n\n"
+            f"Prompt context prepared for model `{context['model']}` with "
+            f"{len(context['memory'])} memory items."
+        )
+        self.session.add(AuditLogORM(category="telegram", message=f"telegram:{telegram_user_id}:{text}"))
+        self.session.commit()
+        return response

backend/app/runtime.py

@@ -0,0 +1,14 @@
+from contextlib import suppress
+
+from app.telegram.bot import TelegramBotService
+
+
+class RuntimeServices:
+    def __init__(self) -> None:
+        self.telegram_bot: TelegramBotService | None = None
+
+    async def shutdown(self) -> None:
+        if self.telegram_bot is not None:
+            with suppress(Exception):
+                await self.telegram_bot.stop()
+

backend/app/security.py

@@ -0,0 +1,77 @@
+from dataclasses import dataclass
+
+
+SAFE_COMMAND_PREFIXES = (
+    "pwd",
+    "ls",
+    "cat",
+    "head",
+    "tail",
+    "find",
+    "rg",
+    "wc",
+    "stat",
+    "git status",
+    "git diff",
+    "git log",
+    "git show",
+    "date",
+    "whoami",
+    "uname",
+    "ps",
+)
+
+APPROVAL_REQUIRED_PREFIXES = (
+    "curl",
+    "wget",
+    "pip",
+    "npm",
+    "python",
+    "python3",
+    "node",
+    "git commit",
+    "git push",
+    "pkill",
+    "kill",
+    "touch",
+    "echo ",
+)
+
+BLOCKED_PATTERNS = (
+    "sudo ",
+    "rm -rf",
+    "chmod ",
+    "chown ",
+    ";",
+    "&&",
+    "||",
+    "$(",
+)
+
+
+@dataclass
+class TerminalDecision:
+    decision: str
+    reason: str
+
+
+def evaluate_terminal_command(command: str, mode: int) -> TerminalDecision:
+    normalized = command.strip()
+
+    if any(pattern in normalized for pattern in BLOCKED_PATTERNS):
+        return TerminalDecision(decision="blocked", reason="Blocked by hard policy.")
+
+    if mode == 1:
+        return TerminalDecision(decision="allow", reason="Terminal mode 1 auto-runs commands.")
+
+    if mode == 2:
+        return TerminalDecision(decision="approval", reason="Terminal mode 2 requires approval.")
+
+    if normalized.startswith(SAFE_COMMAND_PREFIXES):
+        return TerminalDecision(decision="allow", reason="Safe read-only command.")
+
+    if normalized.startswith(APPROVAL_REQUIRED_PREFIXES):
+        return TerminalDecision(decision="approval", reason="Command needs approval.")
+
+    return TerminalDecision(decision="approval", reason="Unknown command defaults to approval.")
+

backend/app/telegram/__init__.py

@@ -0,0 +1 @@
+

backend/app/telegram/auth.py

@@ -0,0 +1,8 @@
+from sqlalchemy.orm import Session
+
+from app.db import AuthorizedUserORM
+
+
+def is_authorized(session: Session, telegram_user_id: int) -> bool:
+    record = session.get(AuthorizedUserORM, telegram_user_id)
+    return bool(record and record.is_active)

backend/app/telegram/bot.py

@@ -0,0 +1,51 @@
+from typing import Any
+
+from telegram import Update
+from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters
+
+from app.orchestrator import WiseClawOrchestrator
+
+
+class TelegramBotService:
+    def __init__(self, token: str, orchestrator_factory: Any) -> None:
+        self.token = token
+        self.orchestrator_factory = orchestrator_factory
+        self.application: Application | None = None
+
+    async def process_message(self, telegram_user_id: int, text: str) -> str:
+        with self.orchestrator_factory() as session:
+            orchestrator = WiseClawOrchestrator(session)
+            return orchestrator.handle_text_message(telegram_user_id=telegram_user_id, text=text)
+
+    async def start(self) -> None:
+        if not self.token:
+            return
+        self.application = Application.builder().token(self.token).build()
+        self.application.add_handler(CommandHandler("start", self._on_start))
+        self.application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, self._on_text))
+        await self.application.initialize()
+        await self.application.start()
+        await self.application.updater.start_polling(drop_pending_updates=True)
+
+    async def stop(self) -> None:
+        if self.application is None:
+            return
+        await self.application.updater.stop()
+        await self.application.stop()
+        await self.application.shutdown()
+        self.application = None
+
+    async def _on_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+        del context
+        if update.message is None or update.effective_user is None:
+            return
+        await update.message.reply_text(
+            "WiseClaw is online. If your Telegram user is whitelisted, send a message to start."
+        )
+
+    async def _on_text(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+        del context
+        if update.message is None or update.effective_user is None or update.message.text is None:
+            return
+        reply = await self.process_message(update.effective_user.id, update.message.text)
+        await update.message.reply_text(reply)

backend/app/tools/__init__.py

@@ -0,0 +1 @@
+

backend/app/tools/apple_notes.py

@@ -0,0 +1,18 @@
+from typing import Any
+
+from app.tools.base import Tool
+
+
+class AppleNotesTool(Tool):
+    name = "apple_notes"
+    description = "Create notes in Apple Notes through AppleScript."
+
+    async def run(self, payload: dict[str, Any]) -> dict[str, Any]:
+        title = str(payload.get("title", "")).strip()
+        return {
+            "tool": self.name,
+            "status": "stub",
+            "title": title,
+            "message": "Apple Notes integration is not wired yet.",
+        }
+

backend/app/tools/base.py

@@ -0,0 +1,12 @@
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class Tool(ABC):
+    name: str
+    description: str
+
+    @abstractmethod
+    async def run(self, payload: dict[str, Any]) -> dict[str, Any]:
+        raise NotImplementedError
+

backend/app/tools/brave_search.py

@@ -0,0 +1,18 @@
+from typing import Any
+
+from app.tools.base import Tool
+
+
+class BraveSearchTool(Tool):
+    name = "brave_search"
+    description = "Search the web with Brave Search."
+
+    async def run(self, payload: dict[str, Any]) -> dict[str, Any]:
+        query = str(payload.get("query", "")).strip()
+        return {
+            "tool": self.name,
+            "status": "stub",
+            "query": query,
+            "message": "Brave Search integration is not wired yet.",
+        }
+

backend/app/tools/files.py

@@ -0,0 +1,21 @@
+from pathlib import Path
+from typing import Any
+
+from app.tools.base import Tool
+
+
+class FilesTool(Tool):
+    name = "files"
+    description = "Read and write files within allowed paths."
+
+    async def run(self, payload: dict[str, Any]) -> dict[str, Any]:
+        action = str(payload.get("action", "read")).strip()
+        path = Path(str(payload.get("path", "")).strip()).expanduser()
+        return {
+            "tool": self.name,
+            "status": "stub",
+            "action": action,
+            "path": str(path),
+            "message": "File integration is not wired yet.",
+        }
+

backend/app/tools/searxng_search.py

@@ -0,0 +1,18 @@
+from typing import Any
+
+from app.tools.base import Tool
+
+
+class SearXNGSearchTool(Tool):
+    name = "searxng_search"
+    description = "Search the web through a SearXNG instance."
+
+    async def run(self, payload: dict[str, Any]) -> dict[str, Any]:
+        query = str(payload.get("query", "")).strip()
+        return {
+            "tool": self.name,
+            "status": "stub",
+            "query": query,
+            "message": "SearXNG integration is not wired yet.",
+        }
+

backend/app/tools/terminal.py

@@ -0,0 +1,24 @@
+from typing import Any
+
+from app.security import evaluate_terminal_command
+from app.tools.base import Tool
+
+
+class TerminalTool(Tool):
+    name = "terminal"
+    description = "Run terminal commands under WiseClaw policy."
+
+    def __init__(self, terminal_mode: int) -> None:
+        self.terminal_mode = terminal_mode
+
+    async def run(self, payload: dict[str, Any]) -> dict[str, Any]:
+        command = str(payload.get("command", "")).strip()
+        decision = evaluate_terminal_command(command, self.terminal_mode)
+        return {
+            "tool": self.name,
+            "status": "stub",
+            "command": command,
+            "decision": decision.decision,
+            "reason": decision.reason,
+        }
+

backend/app/tools/web_fetch.py

@@ -0,0 +1,18 @@
+from typing import Any
+
+from app.tools.base import Tool
+
+
+class WebFetchTool(Tool):
+    name = "web_fetch"
+    description = "Fetch a webpage and return simplified content."
+
+    async def run(self, payload: dict[str, Any]) -> dict[str, Any]:
+        url = str(payload.get("url", "")).strip()
+        return {
+            "tool": self.name,
+            "status": "stub",
+            "url": url,
+            "message": "Web fetch integration is not wired yet.",
+        }
+

backend/pyproject.toml

@@ -0,0 +1,22 @@
+[build-system]
+requires = ["setuptools>=68", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "wiseclaw-backend"
+version = "0.1.0"
+description = "FastAPI backend for WiseClaw"
+readme = "README.md"
+requires-python = ">=3.11"
+dependencies = [
+  "fastapi>=0.116.0,<1.0.0",
+  "uvicorn[standard]>=0.35.0,<1.0.0",
+  "pydantic-settings>=2.10.0,<3.0.0",
+  "sqlalchemy>=2.0.39,<3.0.0",
+  "httpx>=0.28.0,<1.0.0",
+  "python-telegram-bot>=22.0,<23.0",
+]
+
+[tool.setuptools.packages.find]
+where = ["."]
+