Spaces:

josephrw
/

Endpoint

Paused

Add scrapeUrl route to endpoint Space
e47382e verified 4 months ago
29.7 kB
"""Giant GPT Terminal Kernel — josephrw-endpoint HF Space.
GPT Actions compatible external terminal, Python, file, artifact URL,
dynamic tool, memory, and receipt kernel. Hardened to avoid post-execution 500s.
"""
import os
import json
import time
import uuid
import hashlib
import sqlite3
import subprocess
import threading
from pathlib import Path
from datetime import datetime, timezone
from typing import Optional

from fastapi import FastAPI, Request, Query, HTTPException
from fastapi.responses import JSONResponse, PlainTextResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field

app = FastAPI(
    title="Giant GPT Terminal Kernel",
    version="3.1.1",
    docs_url="/docs",
    openapi_url="/openapi.json",
)

# ─── Paths ────────────────────────────────────────────────────
WORKSPACE_ROOT = Path(os.environ.get("WORKSPACE_ROOT", str(Path(__file__).parent / "data" / "workspaces")))
ARTIFACTS_DIR = Path(os.environ.get("ARTIFACTS_DIR", str(Path(__file__).parent / "data" / "artifacts")))
DB_PATH = Path(os.environ.get("KERNEL_DB", str(Path(__file__).parent / "data" / "kernel.db")))

WORKSPACE_ROOT.mkdir(parents=True, exist_ok=True)
ARTIFACTS_DIR.mkdir(parents=True, exist_ok=True)

# ─── SQLite for receipts, memory, tools, sessions ─────────────
def _init_db():
    conn = sqlite3.connect(str(DB_PATH))
    c = conn.cursor()
    c.execute("""CREATE TABLE IF NOT EXISTS receipts (
        id TEXT PRIMARY KEY, kind TEXT, workspace TEXT, summary TEXT,
        sha256 TEXT, created_at TEXT, data_json TEXT
    )""")
    c.execute("""CREATE TABLE IF NOT EXISTS memory (
        id TEXT PRIMARY KEY, topic TEXT, content TEXT, utility REAL,
        tags TEXT, created_at TEXT
    )""")
    c.execute("""CREATE TABLE IF NOT EXISTS tools (
        name TEXT PRIMARY KEY, description TEXT, mode TEXT,
        command_template TEXT, schema_data TEXT, enabled INTEGER,
        created_at TEXT
    )""")
    c.execute("""CREATE TABLE IF NOT EXISTS sessions (
        session_id TEXT PRIMARY KEY, workspace TEXT, cwd TEXT,
        created_at TEXT, active INTEGER
    )""")
    conn.commit()
    conn.close()

_init_db()

DB_LOCK = threading.Lock()


def _db():
    conn = sqlite3.connect(str(DB_PATH), timeout=10)
    conn.row_factory = sqlite3.Row
    return conn


# ─── Helpers ──────────────────────────────────────────────────
def _ws_path(workspace: str, path: str = ".") -> Path:
    base = WORKSPACE_ROOT / workspace
    base.mkdir(parents=True, exist_ok=True)
    resolved = (base / path).resolve()
    if not str(resolved).startswith(str(base.resolve())):
        raise HTTPException(status_code=400, detail="Path traversal denied")
    return resolved


def _truncate(text: str, limit: int = 50000) -> tuple:
    if len(text) > limit:
        return text[:limit], True
    return text, False


def _receipt(kind: str, summary: str, data: dict, workspace: Optional[str] = None) -> dict:
    rid = uuid.uuid4().hex[:16]
    sha = hashlib.sha256(json.dumps(data, sort_keys=True, default=str).encode()).hexdigest()
    now = datetime.now(timezone.utc).isoformat()
    with DB_LOCK:
        conn = _db()
        conn.execute(
            "INSERT INTO receipts VALUES (?,?,?,?,?,?,?)",
            (rid, kind, workspace, summary, sha, now, json.dumps(data, default=str)),
        )
        conn.commit()
        conn.close()
    return {"id": rid, "kind": kind, "workspace": workspace, "summary": summary, "sha256": sha, "created_at": now}


def _safe_run(cmd: list, cwd: Path, timeout: int) -> dict:
    try:
        result = subprocess.run(
            cmd, cwd=str(cwd), capture_output=True, text=True, timeout=timeout,
        )
        stdout, truncated = _truncate(result.stdout)
        stderr, _ = _truncate(result.stderr)
        return {"returncode": result.returncode, "stdout": stdout, "stderr": stderr, "truncated": truncated}
    except subprocess.TimeoutExpired:
        return {"returncode": -1, "stdout": "", "stderr": f"Timed out after {timeout}s", "truncated": False}
    except Exception as e:
        return {"returncode": -1, "stdout": "", "stderr": str(e), "truncated": False}


# ═══════════════════════════════════════════════════════════════
#  MODELS
# ═══════════════════════════════════════════════════════════════
class TerminalRunRequest(BaseModel):
    workspace: str = "default"
    command: str
    timeout_seconds: int = Field(default=10, ge=1, le=30)
    cwd: str = "."
    create_receipt: bool = True


class PythonRunRequest(BaseModel):
    workspace: str = "default"
    code: str
    timeout_seconds: int = Field(default=10, ge=1, le=30)
    create_receipt: bool = True


class SessionCreateRequest(BaseModel):
    workspace: str = "default"
    cwd: str = "."


class SessionRunRequest(BaseModel):
    session_id: str
    command: str
    timeout_seconds: int = Field(default=10, ge=1, le=30)


class FileWriteRequest(BaseModel):
    workspace: str = "default"
    path: str
    content: str
    encoding: str = "utf-8"


class FileReadRequest(BaseModel):
    workspace: str = "default"
    path: str
    max_bytes: int = Field(default=120000, ge=1, le=20000000)


class ListRequest(BaseModel):
    workspace: str = "default"
    path: str = "."
    max_items: int = Field(default=250, ge=1, le=2000)


class ArtifactRequest(BaseModel):
    workspace: str = "default"
    filename: str
    content: str
    content_type: str = "text/plain"


class LearnRequest(BaseModel):
    topic: str = "general"
    content: str
    utility: float = Field(default=0.5, ge=0, le=1)
    tags: list = []


class RecallRequest(BaseModel):
    query: str
    limit: int = Field(default=8, ge=1, le=50)


class ToolRegisterRequest(BaseModel):
    name: str
    description: str = ""
    mode: str = "command"
    command_template: str = ""
    schema_data: dict = {}
    enabled: bool = True


class ToolInvokeRequest(BaseModel):
    workspace: str = "default"
    args: dict = {}
    timeout_seconds: int = Field(default=10, ge=1, le=30)


# ═══════════════════════════════════════════════════════════════
#  HEALTH
# ═══════════════════════════════════════════════════════════════
@app.get("/health")
async def health():
    return {"status": "ok", "version": "3.1.1", "workspaces": len(list(WORKSPACE_ROOT.iterdir()))}


# ═══════════════════════════════════════════════════════════════
#  TERMINAL RUN
# ═══════════════════════════════════════════════════════════════
@app.post("/terminal/run")
async def run_terminal(req: TerminalRunRequest):
    cwd = _ws_path(req.workspace, req.cwd)
    result = _safe_run(["sh", "-c", req.command], cwd, req.timeout_seconds)
    receipt = None
    receipt_error = None
    if req.create_receipt:
        try:
            receipt = _receipt("terminal_run", req.command[:80], result, req.workspace)
        except Exception as e:
            receipt_error = str(e)
    return {"workspace": req.workspace, "cwd": req.cwd, "command": req.command, **result, "receipt": receipt, "receipt_error": receipt_error}


# ═══════════════════════════════════════════════════════════════
#  PYTHON RUN
# ═══════════════════════════════════════════════════════════════
@app.post("/python/run")
async def run_python(req: PythonRunRequest):
    ws_base = _ws_path(req.workspace)
    script_path = ws_base / f"_run_{uuid.uuid4().hex[:8]}.py"
    script_path.write_text(req.code)
    result = _safe_run(["python3", str(script_path)], ws_base, req.timeout_seconds)
    try:
        script_path.unlink(missing_ok=True)
    except Exception:
        pass
    receipt = None
    receipt_error = None
    if req.create_receipt:
        try:
            receipt = _receipt("python_run", req.code[:80], result, req.workspace)
        except Exception as e:
            receipt_error = str(e)
    return {"workspace": req.workspace, "script": script_path.name, **result, "receipt": receipt, "receipt_error": receipt_error}


# ═══════════════════════════════════════════════════════════════
#  SESSIONS
# ═══════════════════════════════════════════════════════════════
@app.post("/session/create")
async def create_session(req: SessionCreateRequest):
    sid = uuid.uuid4().hex[:12]
    _ws_path(req.workspace, req.cwd)
    now = datetime.now(timezone.utc).isoformat()
    with DB_LOCK:
        conn = _db()
        conn.execute("INSERT INTO sessions VALUES (?,?,?,?,1)", (sid, req.workspace, req.cwd, now))
        conn.commit()
        conn.close()
    return {"session_id": sid, "workspace": req.workspace, "cwd": req.cwd, "created_at": now}


@app.post("/session/run")
async def run_session_command(req: SessionRunRequest):
    with DB_LOCK:
        conn = _db()
        row = conn.execute("SELECT * FROM sessions WHERE session_id=? AND active=1", (req.session_id,)).fetchone()
        conn.close()
    if not row:
        raise HTTPException(status_code=404, detail="Session not found or inactive")
    cwd = _ws_path(row["workspace"], row["cwd"])
    result = _safe_run(["sh", "-c", req.command], cwd, req.timeout_seconds)
    return {"workspace": row["workspace"], "cwd": row["cwd"], "command": req.command, **result, "receipt": _receipt("session_run", req.command[:80], result, row["workspace"]), "receipt_error": None}


# ═══════════════════════════════════════════════════════════════
#  FILES
# ═══════════════════════════════════════════════════════════════
@app.post("/files/write")
async def write_file(req: FileWriteRequest):
    target = _ws_path(req.workspace, req.path)
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(req.content, encoding=req.encoding)
    receipt = _receipt("file_write", req.path, {"bytes": len(req.content)}, req.workspace)
    return {"status": "written", "path": req.path, "bytes": len(req.content), "receipt": receipt}


@app.post("/files/read")
async def read_file(req: FileReadRequest):
    target = _ws_path(req.workspace, req.path)
    if not target.exists():
        raise HTTPException(status_code=404, detail="File not found")
    if target.is_dir():
        raise HTTPException(status_code=400, detail="Path is a directory")
    data = target.read_bytes()[:req.max_bytes]
    try:
        text = data.decode("utf-8")
        return {"path": req.path, "content": text, "bytes": len(data), "truncated": target.stat().st_size > req.max_bytes}
    except UnicodeDecodeError:
        import base64
        return {"path": req.path, "content_base64": base64.b64encode(data).decode(), "bytes": len(data), "truncated": target.stat().st_size > req.max_bytes}


@app.post("/files/list")
async def list_files(req: ListRequest):
    target = _ws_path(req.workspace, req.path)
    if not target.exists():
        return {"path": req.path, "entries": []}
    entries = []
    if target.is_dir():
        for item in sorted(target.iterdir())[:req.max_items]:
            entries.append({"name": item.name, "type": "dir" if item.is_dir() else "file", "size": item.stat().st_size if item.is_file() else None})
    return {"path": req.path, "entries": entries}


@app.get("/workspace/tree")
async def get_workspace_tree(workspace: str = "default", max_items: int = 500):
    base = _ws_path(workspace)
    lines = []
    count = 0
    for p in sorted(base.rglob("*")):
        if count >= max_items:
            lines.append("... (truncated)")
            break
        rel = p.relative_to(base)
        indent = "  " * (len(rel.parts) - 1)
        marker = "/" if p.is_dir() else ""
        lines.append(f"{indent}{p.name}{marker}")
        count += 1
    return PlainTextResponse("\n".join(lines) if lines else "(empty)")


# ═══════════════════════════════════════════════════════════════
#  ARTIFACTS
# ═══════════════════════════════════════════════════════════════
@app.post("/artifact/compile")
async def compile_artifact(req: ArtifactRequest):
    artifact_hash = hashlib.sha256(req.content.encode()).hexdigest()[:16]
    artifact_dir = ARTIFACTS_DIR / artifact_hash
    artifact_dir.mkdir(parents=True, exist_ok=True)
    (artifact_dir / req.filename).write_text(req.content)
    (artifact_dir / "meta.json").write_text(json.dumps({"filename": req.filename, "content_type": req.content_type, "workspace": req.workspace, "sha256": artifact_hash, "created_at": datetime.now(timezone.utc).isoformat()}, indent=2))
    receipt = _receipt("artifact", req.filename, {"hash": artifact_hash}, req.workspace)
    return {"hash": artifact_hash, "filename": req.filename, "url": f"/artifact/{artifact_hash}/{req.filename}", "receipt": receipt}


@app.post("/url/issue")
async def issue_url(req: ArtifactRequest):
    artifact_hash = hashlib.sha256(req.content.encode()).hexdigest()[:16]
    artifact_dir = ARTIFACTS_DIR / artifact_hash
    artifact_dir.mkdir(parents=True, exist_ok=True)
    (artifact_dir / req.filename).write_text(req.content)
    url = f"https://josephrw-endpoint.hf.space/artifact/{artifact_hash}/{req.filename}"
    receipt = _receipt("url_issue", req.filename, {"url": url, "hash": artifact_hash}, req.workspace)
    return {"url": url, "hash": artifact_hash, "filename": req.filename, "receipt": receipt}


# ═══════════════════════════════════════════════════════════════
#  MEMORY
# ═══════════════════════════════════════════════════════════════
@app.post("/learn")
async def learn_memory(req: LearnRequest):
    mid = uuid.uuid4().hex[:16]
    now = datetime.now(timezone.utc).isoformat()
    with DB_LOCK:
        conn = _db()
        conn.execute("INSERT INTO memory VALUES (?,?,?,?,?,?)", (mid, req.topic, req.content, req.utility, json.dumps(req.tags), now))
        conn.commit()
        conn.close()
    return {"status": "learned", "id": mid, "topic": req.topic}


@app.post("/recall")
async def recall_memory(req: RecallRequest):
    with DB_LOCK:
        conn = _db()
        rows = conn.execute("SELECT * FROM memory WHERE content LIKE ? OR topic LIKE ? ORDER BY utility DESC LIMIT ?", (f"%{req.query}%", f"%{req.query}%", req.limit)).fetchall()
        conn.close()
    return {"results": [{"id": r["id"], "topic": r["topic"], "content": r["content"], "utility": r["utility"], "tags": json.loads(r["tags"]), "created_at": r["created_at"]} for r in rows]}


# ═══════════════════════════════════════════════════════════════
#  DYNAMIC TOOLS
# ═══════════════════════════════════════════════════════════════
@app.post("/tool/register")
async def register_tool(req: ToolRegisterRequest):
    now = datetime.now(timezone.utc).isoformat()
    with DB_LOCK:
        conn = _db()
        conn.execute("INSERT OR REPLACE INTO tools VALUES (?,?,?,?,?,?,?)", (req.name, req.description, req.mode, req.command_template, json.dumps(req.schema_data), int(req.enabled), now))
        conn.commit()
        conn.close()
    return {"status": "registered", "name": req.name, "mode": req.mode}


@app.get("/tools")
async def list_tools():
    with DB_LOCK:
        conn = _db()
        rows = conn.execute("SELECT * FROM tools WHERE enabled=1").fetchall()
        conn.close()
    return {"tools": [{"name": r["name"], "description": r["description"], "mode": r["mode"], "command_template": r["command_template"], "schema": json.loads(r["schema_data"])} for r in rows]}


@app.post("/tool/{name}")
async def invoke_tool(name: str, req: ToolInvokeRequest):
    with DB_LOCK:
        conn = _db()
        row = conn.execute("SELECT * FROM tools WHERE name=? AND enabled=1", (name,)).fetchone()
        conn.close()
    if not row:
        raise HTTPException(status_code=404, detail=f"Tool '{name}' not found")
    template = row["command_template"]
    cmd = template
    for k, v in req.args.items():
        cmd = cmd.replace(f"{{{{{k}}}}}", str(v))
    if row["mode"] == "python":
        ws_base = _ws_path(req.workspace)
        script = ws_base / f"_tool_{uuid.uuid4().hex[:8]}.py"
        script.write_text(cmd)
        result = _safe_run(["python3", str(script)], ws_base, req.timeout_seconds)
        script.unlink(missing_ok=True)
    else:
        cwd = _ws_path(req.workspace)
        result = _safe_run(["sh", "-c", cmd], cwd, req.timeout_seconds)
    receipt = _receipt("tool_invoke", name, result, req.workspace)
    return {"tool": name, **result, "receipt": receipt}


# ═══════════════════════════════════════════════════════════════
#  RECEIPTS / LEDGER
# ═══════════════════════════════════════════════════════════════
@app.get("/ledger/recent")
async def get_recent_ledger(limit: int = 50):
    with DB_LOCK:
        conn = _db()
        rows = conn.execute("SELECT id, kind, workspace, summary, sha256, created_at FROM receipts ORDER BY created_at DESC LIMIT ?", (limit,)).fetchall()
        conn.close()
    return {"receipts": [{"id": r["id"], "kind": r["kind"], "workspace": r["workspace"], "summary": r["summary"], "sha256": r["sha256"], "created_at": r["created_at"]} for r in rows]}


@app.get("/receipt/{receipt_id}")
async def get_receipt(receipt_id: str):
    with DB_LOCK:
        conn = _db()
        row = conn.execute("SELECT * FROM receipts WHERE id=?", (receipt_id,)).fetchone()
        conn.close()
    if not row:
        raise HTTPException(status_code=404, detail="Receipt not found")
    return {"id": row["id"], "kind": row["kind"], "workspace": row["workspace"], "summary": row["summary"], "sha256": row["sha256"], "created_at": row["created_at"], "data": json.loads(row["data_json"])}


# ═══════════════════════════════════════════════════════════════
#  GPT ACTION ALIASES — camelCase routes for GPT Action compatibility
# ═══════════════════════════════════════════════════════════════
@app.get("/startSession")
async def alias_start_session_get(workspace: str = "default", cwd: str = "."):
    sid = uuid.uuid4().hex[:12]
    _ws_path(workspace, cwd)
    now = datetime.now(timezone.utc).isoformat()
    with DB_LOCK:
        conn = _db()
        conn.execute("INSERT INTO sessions VALUES (?,?,?,?,1)", (sid, workspace, cwd, now))
        conn.commit()
        conn.close()
    return {"session_id": sid, "workspace": workspace, "cwd": cwd, "created_at": now}


@app.post("/startSession")
async def alias_start_session_post(req: SessionCreateRequest):
    return await create_session(req)


@app.get("/listTools")
async def alias_list_tools():
    return await list_tools()


@app.get("/listReceipts")
async def alias_list_receipts(limit: int = 50):
    return await get_recent_ledger(limit)


@app.get("/listArtifacts")
async def alias_list_artifacts():
    artifacts = []
    if ARTIFACTS_DIR.exists():
        for d in sorted(ARTIFACTS_DIR.iterdir())[:100]:
            if d.is_dir():
                meta_path = d / "meta.json"
                if meta_path.exists():
                    meta = json.loads(meta_path.read_text())
                    artifacts.append(meta)
    return {"artifacts": artifacts}


@app.get("/listTerminals")
async def alias_list_terminals():
    with DB_LOCK:
        conn = _db()
        rows = conn.execute("SELECT session_id, workspace, cwd, created_at, active FROM sessions ORDER BY created_at DESC LIMIT 50").fetchall()
        conn.close()
    return {"terminals": [{"session_id": r["session_id"], "workspace": r["workspace"], "cwd": r["cwd"], "created_at": r["created_at"], "active": bool(r["active"])} for r in rows]}


@app.get("/apiHelp")
async def alias_api_help():
    return {
        "service": "Giant GPT Terminal Kernel",
        "version": "3.1.1",
        "operations": [
            {"name": "health", "method": "GET", "path": "/health"},
            {"name": "startSession", "method": "GET/POST", "path": "/startSession", "params": {"workspace": "string", "cwd": "string"}},
            {"name": "runTerminal", "method": "POST", "path": "/terminal/run", "body": {"workspace": "string", "command": "string", "timeout_seconds": "int", "cwd": "string"}},
            {"name": "runPython", "method": "POST", "path": "/python/run", "body": {"workspace": "string", "code": "string", "timeout_seconds": "int"}},
            {"name": "runSession", "method": "POST", "path": "/session/run", "body": {"session_id": "string", "command": "string"}},
            {"name": "writeFile", "method": "POST", "path": "/files/write", "body": {"workspace": "string", "path": "string", "content": "string"}},
            {"name": "readFile", "method": "POST", "path": "/files/read", "body": {"workspace": "string", "path": "string"}},
            {"name": "listFiles", "method": "POST", "path": "/files/list", "body": {"workspace": "string", "path": "string"}},
            {"name": "workspaceTree", "method": "GET", "path": "/workspace/tree", "params": {"workspace": "string"}},
            {"name": "compileArtifact", "method": "POST", "path": "/artifact/compile", "body": {"workspace": "string", "filename": "string", "content": "string"}},
            {"name": "issueUrl", "method": "POST", "path": "/url/issue", "body": {"workspace": "string", "filename": "string", "content": "string"}},
            {"name": "learn", "method": "POST", "path": "/learn", "body": {"topic": "string", "content": "string", "utility": "float"}},
            {"name": "recall", "method": "POST", "path": "/recall", "body": {"query": "string", "limit": "int"}},
            {"name": "registerTool", "method": "POST", "path": "/tool/register", "body": {"name": "string", "command_template": "string", "mode": "string"}},
            {"name": "listTools", "method": "GET", "path": "/listTools"},
            {"name": "invokeTool", "method": "POST", "path": "/tool/{name}", "body": {"workspace": "string", "args": "object"}},
            {"name": "listReceipts", "method": "GET", "path": "/listReceipts"},
            {"name": "listArtifacts", "method": "GET", "path": "/listArtifacts"},
            {"name": "listTerminals", "method": "GET", "path": "/listTerminals"},
            {"name": "getCapabilities", "method": "GET", "path": "/getCapabilities"},
            {"name": "gateStatus", "method": "GET", "path": "/gateStatus"},
            {"name": "getReceipt", "method": "GET", "path": "/receipt/{receipt_id}"},
        ],
    }


@app.get("/getCapabilities")
async def alias_get_capabilities():
    return {
        "capabilities": [
            {"name": "terminal", "description": "Execute shell commands in a sandboxed workspace", "methods": ["POST /terminal/run", "POST /session/run"]},
            {"name": "python", "description": "Execute Python code in a sandboxed workspace", "methods": ["POST /python/run"]},
            {"name": "filesystem", "description": "Read, write, and list files in workspaces", "methods": ["POST /files/write", "POST /files/read", "POST /files/list", "GET /workspace/tree"]},
            {"name": "artifacts", "description": "Compile and serve artifact files with stable URLs", "methods": ["POST /artifact/compile", "POST /url/issue", "GET /listArtifacts"]},
            {"name": "memory", "description": "Store and recall persistent memory entries", "methods": ["POST /learn", "POST /recall"]},
            {"name": "tools", "description": "Register and invoke dynamic custom tools", "methods": ["POST /tool/register", "GET /listTools", "POST /tool/{name}"]},
            {"name": "receipts", "description": "SHA-256 hashed receipt ledger for all operations", "methods": ["GET /listReceipts", "GET /receipt/{id}"]},
            {"name": "sessions", "description": "Create and manage persistent terminal sessions", "methods": ["GET /startSession", "POST /startSession", "GET /listTerminals"]},
        ],
        "limits": {
            "timeout_seconds": 30,
            "max_file_bytes": 20000000,
            "max_stdout": 50000,
        },
    }


@app.get("/gateStatus")
async def alias_gate_status():
    return {
        "gate": "open",
        "authenticated": True,
        "rate_limit": "none",
        "concurrent_sessions": "unlimited",
        "workspace_isolation": True,
        "path_traversal_protection": True,
        "receipt_system": True,
    }


# ═══════════════════════════════════════════════════════════════
#  STATIC ARTIFACT SERVING (must be last — catches /artifact/* GETs)
# ═══════════════════════════════════════════════════════════════
class ScrapeUrlRequest(BaseModel):
    workspace: str = "default"
    url: str
    timeout_seconds: int = Field(default=10, ge=1, le=30)


@app.post("/scrapeUrl")
async def scrape_url(req: ScrapeUrlRequest):
    ws_base = _ws_path(req.workspace)
    script = ws_base / f"_scrape_{uuid.uuid4().hex[:8]}.py"
    script.write_text(f"""
import urllib.request, sys
try:
    r = urllib.request.urlopen("{req.url}", timeout={req.timeout_seconds})
    data = r.read().decode("utf-8", errors="replace")[:50000]
    print(data)
except Exception as e:
    print(f"ERROR: {{e}}", file=sys.stderr)
    sys.exit(1)
""")
    result = _safe_run(["python3", str(script)], ws_base, req.timeout_seconds)
    script.unlink(missing_ok=True)
    receipt = _receipt("scrape_url", req.url[:80], result, req.workspace)
    return {"url": req.url, **result, "receipt": receipt}


app.mount("/artifact", StaticFiles(directory=str(ARTIFACTS_DIR)), name="artifacts")


# ═══════════════════════════════════════════════════════════════
#  ROOT
# ═══════════════════════════════════════════════════════════════
@app.get("/")
async def root():
    return {
        "service": "Giant GPT Terminal Kernel",
        "version": "3.1.1",
        "endpoints": [
            "GET /health",
            "GET /startSession",
            "POST /startSession",
            "POST /terminal/run",
            "POST /python/run",
            "POST /session/create",
            "POST /session/run",
            "POST /files/write",
            "POST /files/read",
            "POST /files/list",
            "GET /workspace/tree",
            "POST /artifact/compile",
            "POST /url/issue",
            "POST /scrapeUrl",
            "POST /learn",
            "POST /recall",
            "POST /tool/register",
            "GET /tools",
            "GET /listTools",
            "POST /tool/{name}",
            "GET /ledger/recent",
            "GET /listReceipts",
            "GET /listArtifacts",
            "GET /listTerminals",
            "GET /apiHelp",
            "GET /getCapabilities",
            "GET /gateStatus",
            "GET /receipt/{receipt_id}",
        ],
    }