Use Pixeltable query routes directly

AGENTS.md

@@ -4,9 +4,9 @@ Pixelbot 3.0 is an app-first Pixeltable 0.7.7 project. Python 3.11+ and Node 22.
 
 ## Architecture
 
-- `backend/pixelbot/app.py`: application entry, custom FastAPI app, and module-scope `FastAPIRouter` routes.
-- `backend/pixelbot/schema.py`: the single `TableModel` schema for `pixelbot_v3`.
-- `backend/pixelbot/routers/`: custom REST routes. The database router is read-only.
+- `backend/pixelbot/app.py`: application entry, endpoint-facing `@pxt.query` functions, module-scope `FastAPIRouter` routes, and the custom FastAPI app.
+- `backend/pixelbot/schema.py`: the single `TableModel` schema for `pixelbot_v3`; its query functions are reusable parts of computed pipelines.
+- `backend/pixelbot/routers/`: custom REST writes and HTTP-specific behavior. The database router is read-only.
 - `backend/pixelbot/static/`: production SPA package data generated by Vite.
 - `frontend/`: React SPA with route-level lazy loading.
 - Pixeltable agent guidance lives in the canonical
@@ -19,6 +19,8 @@ Use a fresh `PIXELTABLE_HOME=backend/.pixeltable-v3`. Apply only to `pixelbot_v3
 
 Declare stored columns with annotations and computed columns with assignments in `schema.py`. Keep `has_default_idxs=False`; add targeted B-tree and embedding indexes through `__indexes__`. Do not mutate the catalog during import or server startup. Do not add runtime expression evaluation, dynamic computed columns, arbitrary catalog paths, or database mutation endpoints.
 
+Use `@pxt.query` plus `FastAPIRouter.add_query_route()` for table-backed read endpoints. Put endpoint-only queries beside their routes in `app.py`; keep queries in `schema.py` when computed columns also call them. Use custom FastAPI handlers for request validation, application defaults, and multi-step writes. Do not mirror a query in a plain Python helper.
+
 A changed computed expression is unsupported by schema migration. Rename the column, or drop and re-add it in separate schema updates. `--allow-destructive` does not change that restriction.
 
 ## Workflow

README.md

@@ -4,6 +4,8 @@ Pixelbot 3.0 is a local multimodal AI application built on Pixeltable 0.7.7. One
 
 Pixelbot keeps documents, images, video, audio, memory, chat history, generation jobs, prompt experiments, notifications, and tool calls in Pixeltable. Stored computed columns run extraction, chunking, transcription, captioning, generation, retrieval, and tool invocation when rows are inserted. Provider calls require the matching keys; local schema and API checks do not call paid providers.
 
+The HTTP layer demonstrates a practical Pixeltable and FastAPI split. Table-backed reads such as `GET /api/memory`, `GET /api/memory/search`, and `GET /api/personas` are declared as `@pxt.query` functions and exposed through `FastAPIRouter`. Custom FastAPI handlers own validation-heavy writes, application defaults, and the SPA shell.
+
 ## Requirements
 
 - Python 3.11–3.14

backend/pixelbot/app.py

@@ -27,20 +27,71 @@
 )
 from pixelbot.schema import (
     MemoryBank,
-    get_all_memory,
-    get_all_personas,
-    search_memory,
+    UserPersonas,
 )
 from pixelbot.schema import TableModel as TableModel  # noqa: F401
 
 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s [%(name)s] %(message)s")
 logger = logging.getLogger(__name__)
 
-dataServingRouter = FastAPIRouter(name="data_serving", prefix="/api", tags=["data-serving"])
-dataServingRouter.add_query_route(path="/memory/v2", query=get_all_memory, method="get")
-dataServingRouter.add_query_route(path="/memory/v2/search", query=search_memory, method="get")
-dataServingRouter.add_delete_route(MemoryBank, path="/memory/v2/delete", match_columns=["timestamp"])
-dataServingRouter.add_query_route(path="/personas/v2", query=get_all_personas, method="get")
+
+@pxt.query
+def list_memory_rows():
+    """List memory for the configured local user."""
+    return (
+        MemoryBank.where(MemoryBank.user_id == config.DEFAULT_USER_ID)
+        .select(
+            content=MemoryBank.content,
+            type=MemoryBank.type,
+            language=MemoryBank.language,
+            context_query=MemoryBank.context_query,
+            timestamp=MemoryBank.timestamp,
+        )
+        .order_by(MemoryBank.timestamp, asc=False)
+        .limit(100)
+    )
+
+
+@pxt.query
+def search_memory_rows(query_text: str):
+    """Search memory for the configured local user."""
+    similarity = MemoryBank.content.similarity(string=query_text)  # type: ignore[attr-defined]
+    return (
+        MemoryBank.where((MemoryBank.user_id == config.DEFAULT_USER_ID) & (similarity > 0.7))
+        .order_by(similarity, asc=False)
+        .select(
+            content=MemoryBank.content,
+            type=MemoryBank.type,
+            language=MemoryBank.language,
+            context_query=MemoryBank.context_query,
+            timestamp=MemoryBank.timestamp,
+            sim=similarity,
+        )
+        .limit(10)
+    )
+
+
+@pxt.query
+def list_persona_rows():
+    """List personas for the configured local user."""
+    return (
+        UserPersonas.where(UserPersonas.user_id == config.DEFAULT_USER_ID)
+        .select(
+            persona_name=UserPersonas.persona_name,
+            initial_prompt=UserPersonas.initial_prompt,
+            final_prompt=UserPersonas.final_prompt,
+            llm_params=UserPersonas.llm_params,
+            timestamp=UserPersonas.timestamp,
+        )
+        .order_by(UserPersonas.persona_name, asc=True)
+        .limit(100)
+    )
+
+
+pixeltableRouter = FastAPIRouter(name="pixeltable", prefix="/api", tags=["pixeltable"])
+pixeltableRouter.add_query_route(path="/memory", query=list_memory_rows, method="get")
+pixeltableRouter.add_query_route(path="/memory/search", query=search_memory_rows, method="get")
+pixeltableRouter.add_query_route(path="/personas", query=list_persona_rows, method="get")
 
 app = FastAPI(
     title="Pixelbot",
@@ -74,7 +125,7 @@ async def unexpected_error_handler(request: Request, exc: Exception) -> JSONResp
     experiments.router,
     export.router,
     integrations.router,
-    dataServingRouter,
+    pixeltableRouter,
 ):
     app.include_router(apiRouter)
 

backend/pixelbot/routers/database.py

@@ -449,7 +449,6 @@ def join_tables(body: JoinRequest):
     "search_memory": "pixelbot_v3/memory_bank",
     "search_chat_history": "pixelbot_v3/chat_history",
     "get_recent_chat_history": "pixelbot_v3/chat_history",
-    "get_all_memory": "pixelbot_v3/memory_bank",
 }
 
 

backend/pixelbot/routers/integrations.py

@@ -5,7 +5,7 @@
 
 import pixeltable as pxt
 import requests as http_requests
-from fastapi import APIRouter
+from fastapi import APIRouter, Query
 
 from pixelbot import config
 from pixelbot.models import (
@@ -74,7 +74,6 @@ def test_notification(req: TestNotificationRequest):
     )
 
 
-_NOTIFICATION_TOOLS = ("send_slack_message", "send_discord_message", "send_webhook")
 _TOOL_SERVICE_MAP = {
     "send_slack_message": "slack",
     "send_discord_message": "discord",
@@ -84,14 +83,15 @@ def test_notification(req: TestNotificationRequest):
 
 @router.get("/log", response_model=NotificationLogResponse)
 @pxt_retry()
-def get_notification_log(limit: int = 50):
+def get_notification_log(limit: int = Query(default=50, ge=1, le=100)):
     """Get recent notification activity from both manual tests and agent tool calls."""
     entries: list[NotificationLogEntry] = []
 
     # Source 1: explicit notification table (manual test sends)
     notifications = pxt.get_table("pixelbot_v3.notifications")
     manual_rows = (
-        notifications.select(
+        notifications.where(notifications.user_id == config.DEFAULT_USER_ID)
+        .select(
             notifications.service,
             notifications.message,
             notifications.status,
@@ -101,16 +101,16 @@ def get_notification_log(limit: int = 50):
         .order_by(notifications.timestamp, asc=False)
         .limit(limit)
         .collect()
-        .to_pandas()
     )
-    for _, r in manual_rows.iterrows():
+    for row in manual_rows:
+        timestamp = row["timestamp"]
         entries.append(
             NotificationLogEntry(
-                service=r["service"],
-                message=r["message"],
-                status=r["status"],
-                response_code=int(r["response_code"]) if r["response_code"] is not None else 0,
-                timestamp=r["timestamp"].isoformat() if hasattr(r["timestamp"], "isoformat") else str(r["timestamp"]),
+                service=row["service"],
+                message=row["message"],
+                status=row["status"],
+                response_code=int(row["response_code"]) if row["response_code"] is not None else 0,
+                timestamp=timestamp.isoformat() if hasattr(timestamp, "isoformat") else str(timestamp),
                 source="manual",
             )
         )
@@ -119,7 +119,8 @@ def get_notification_log(limit: int = 50):
     try:
         tools_table = pxt.get_table("pixelbot_v3.tools")
         agent_rows = (
-            tools_table.select(tools_table.prompt, tools_table.tool_output, tools_table.timestamp)
+            tools_table.where(tools_table.user_id == config.DEFAULT_USER_ID)
+            .select(tools_table.prompt, tools_table.tool_output, tools_table.timestamp)
             .order_by(tools_table.timestamp, asc=False)
             .limit(limit)
             .collect()
@@ -128,15 +129,15 @@ def get_notification_log(limit: int = 50):
             tool_output = r.get("tool_output")
             if not isinstance(tool_output, dict):
                 continue
-            for tool_name in _NOTIFICATION_TOOLS:
+            for tool_name, service in _TOOL_SERVICE_MAP.items():
                 result = tool_output.get(tool_name)
                 if not result:
                     continue
                 result_str = result[0] if isinstance(result, list) and result else str(result)
                 is_success = "successfully" in result_str.lower() or "delivered" in result_str.lower()
                 entries.append(
                     NotificationLogEntry(
-                        service=_TOOL_SERVICE_MAP[tool_name],
+                        service=service,
                         message=r.get("prompt", "")[:200],
                         status="success" if is_success else "error",
                         response_code=200 if is_success else 0,

backend/pixelbot/routers/memory.py

@@ -1,14 +1,11 @@
-import io
-import json
 import logging
 from datetime import datetime
 
 import pixeltable as pxt
-from fastapi import APIRouter, HTTPException, Query
-from fastapi.responses import StreamingResponse
+from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel
 
-from pixelbot import config, queries
+from pixelbot import config
 from pixelbot.models import DeleteMemoryResponse, MemoryBankRow, MessageResponse
 from pixelbot.utils import pxt_retry
 
@@ -62,45 +59,6 @@ def save_memory(body: SaveMemoryRequest):
     return _insert_memory(body)
 
 
-@router.post("/memory/manual", status_code=201, response_model=MessageResponse)
-@pxt_retry()
-def add_memory_manual(body: SaveMemoryRequest):
-    """Save a manually added memory item (backward-compatible alias)."""
-    return _insert_memory(body)
-
-
-# ── Get Memory ────────────────────────────────────────────────────────────────
-
-
-@router.get("/memory")
-@pxt_retry()
-def get_memory(search: str | None = Query(default=None)):
-    """Retrieve memory items, optionally filtering by semantic search.
-
-    Uses shared query functions from queries.py (mirrors the @pxt.query
-    definitions in pixelbot/schema.py) and direct ResultSet iteration.
-    """
-    user_id = config.DEFAULT_USER_ID
-
-    try:
-        if search:
-            rows = queries.search_memory(search, user_id)
-        else:
-            rows = queries.get_all_memory(user_id)
-
-        # Format timestamps for JSON serialization
-        for row in rows:
-            ts = row.get("timestamp")
-            if ts:
-                row["timestamp"] = ts.strftime("%Y-%m-%d %H:%M:%S.%f")
-
-        return rows
-
-    except Exception as e:
-        logger.error(f"Error fetching memory: {e}", exc_info=True)
-        raise HTTPException(status_code=500, detail=str(e))
-
-
 # ── Delete Memory ─────────────────────────────────────────────────────────────
 
 
@@ -131,32 +89,3 @@ def delete_memory(timestamp_str: str):
     except Exception as e:
         logger.error(f"Error deleting memory: {e}", exc_info=True)
         raise HTTPException(status_code=500, detail=str(e))
-
-
-# ── Download Memory ───────────────────────────────────────────────────────────
-
-
-@router.get("/download_memory")
-@pxt_retry()
-def download_memory():
-    """Download all memory bank items as JSON."""
-    user_id = config.DEFAULT_USER_ID
-
-    try:
-        rows = queries.get_all_memory(user_id)
-        for row in rows:
-            ts = row.get("timestamp")
-            if ts:
-                row["timestamp"] = ts.strftime("%Y-%m-%d %H:%M:%S.%f")
-
-        json_bytes = json.dumps(rows, indent=2).encode("utf-8")
-
-        return StreamingResponse(
-            io.BytesIO(json_bytes),
-            media_type="application/json",
-            headers={"Content-Disposition": "attachment; filename=memory_bank.json"},
-        )
-
-    except Exception as e:
-        logger.error(f"Error downloading memory: {e}", exc_info=True)
-        raise HTTPException(status_code=500, detail=str(e))