Create App.py

App.py

@@ -0,0 +1,489 @@
+import ast
+import hashlib
+import json
+import os
+import re
+import shlex
+import sqlite3
+import subprocess
+import tempfile
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import PlainTextResponse
+from pydantic import BaseModel, Field
+
+APP_NAME = "GPT Tool Terminal HF Space"
+APP_VERSION = "2.0.0"
+ROOT = Path(os.environ.get("DATA_ROOT", "/data")).resolve()
+WORKSPACE = (ROOT / "workspace").resolve()
+LEDGER_DIR = (ROOT / "ledger").resolve()
+DB_PATH = LEDGER_DIR / "tool_memory.sqlite3"
+MAX_OUTPUT_CHARS = int(os.environ.get("MAX_OUTPUT_CHARS", "12000"))
+MAX_FILE_BYTES = int(os.environ.get("MAX_FILE_BYTES", str(2_000_000)))
+DEFAULT_TIMEOUT = int(os.environ.get("DEFAULT_TIMEOUT", "8"))
+MAX_TIMEOUT = int(os.environ.get("MAX_TIMEOUT", "20"))
+
+WORKSPACE.mkdir(parents=True, exist_ok=True)
+LEDGER_DIR.mkdir(parents=True, exist_ok=True)
+
+app = FastAPI(
+    title="GPT Tool Terminal HF Space",
+    version=APP_VERSION,
+    description=(
+        "No-key GPT Actions backend for external self-learning memory, constrained terminal execution, "
+        "Python experiments, artifact storage, file IO, and hash receipts. This Space intentionally avoids "
+        "unrestricted public shell execution. Commands run inside /data/workspace with timeouts, output caps, "
+        "path containment, and a conservative command policy."
+    ),
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=False,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+def now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def sha256_text(text: str) -> str:
+    return hashlib.sha256(text.encode("utf-8")).hexdigest()
+
+
+def clamp_output(s: str) -> str:
+    if len(s) <= MAX_OUTPUT_CHARS:
+        return s
+    return s[:MAX_OUTPUT_CHARS] + f"\n\n[TRUNCATED: output exceeded {MAX_OUTPUT_CHARS} chars]"
+
+
+def ensure_relative_path(path: str) -> Path:
+    if not path or path.strip() in {"/", ".", ".."}:
+        raise HTTPException(status_code=400, detail="Path must be a non-root relative workspace path.")
+    candidate = (WORKSPACE / path).resolve()
+    if not str(candidate).startswith(str(WORKSPACE)):
+        raise HTTPException(status_code=400, detail="Path escapes workspace containment.")
+    return candidate
+
+
+def init_db() -> None:
+    with sqlite3.connect(DB_PATH) as con:
+        con.execute("PRAGMA journal_mode=WAL")
+        con.execute(
+            """
+            CREATE TABLE IF NOT EXISTS ledger (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                ts TEXT NOT NULL,
+                kind TEXT NOT NULL,
+                summary TEXT NOT NULL,
+                payload_json TEXT NOT NULL,
+                content_hash TEXT NOT NULL
+            )
+            """
+        )
+        con.execute(
+            """
+            CREATE TABLE IF NOT EXISTS memory_packets (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                ts TEXT NOT NULL,
+                source_kind TEXT NOT NULL,
+                title TEXT NOT NULL,
+                packet TEXT NOT NULL,
+                tags TEXT NOT NULL,
+                utility REAL NOT NULL,
+                payload_json TEXT NOT NULL,
+                content_hash TEXT NOT NULL
+            )
+            """
+        )
+        con.commit()
+
+
+init_db()
+
+
+def ledger(kind: str, summary: str, payload: Dict[str, Any]) -> Dict[str, Any]:
+    body = {"ts": now_iso(), "kind": kind, "summary": summary, "payload": payload}
+    content_hash = sha256_text(json.dumps(body, sort_keys=True, ensure_ascii=False))
+    with sqlite3.connect(DB_PATH) as con:
+        cur = con.execute(
+            "INSERT INTO ledger(ts, kind, summary, payload_json, content_hash) VALUES (?, ?, ?, ?, ?)",
+            (body["ts"], kind, summary[:500], json.dumps(payload, ensure_ascii=False), content_hash),
+        )
+        row_id = cur.lastrowid
+        con.commit()
+    return {"receipt_id": str(row_id), "content_hash": content_hash, "ts": body["ts"]}
+
+
+DANGEROUS_PATTERNS = [
+    r"\brm\b.*\-r", r"\brm\b.*\-f", r"\bmkfs\b", r"\bdd\b", r"\bshutdown\b", r"\breboot\b",
+    r"\bmount\b", r"\bumount\b", r"\bsudo\b", r"\bsu\b", r"\bchmod\b.*777", r"\bchown\b",
+    r"/etc/passwd", r"/etc/shadow", r"/proc/", r"/sys/", r"/dev/", r"\bssh\b", r"\bscp\b",
+    r"\bnc\b", r"\bnetcat\b", r"\btelnet\b", r"\bcurl\b", r"\bwget\b", r"\bpython\b.*socket",
+    r"\bbase64\b.*\-d", r"\beval\b", r"\bexec\b.*\$", r"\bwhile\s+true\b", r":\(\)\s*\{",
+]
+
+SAFE_FIRST_TOKENS = {
+    "pwd", "ls", "find", "cat", "head", "tail", "wc", "grep", "egrep", "fgrep", "sed", "awk", "sort", "uniq",
+    "date", "echo", "printf", "cut", "tr", "du", "stat", "file", "sqlite3", "git", "python", "python3", "bash", "sh",
+}
+
+
+def validate_command(command: str) -> None:
+    if not command or len(command) > 2000:
+        raise HTTPException(status_code=400, detail="Command is empty or too long.")
+    lowered = command.lower()
+    for pat in DANGEROUS_PATTERNS:
+        if re.search(pat, lowered):
+            raise HTTPException(status_code=400, detail=f"Command rejected by safety policy: {pat}")
+    try:
+        tokens = shlex.split(command)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=f"Invalid shell syntax: {e}")
+    if not tokens:
+        raise HTTPException(status_code=400, detail="Command has no tokens.")
+    first = Path(tokens[0]).name
+    if first not in SAFE_FIRST_TOKENS:
+        raise HTTPException(status_code=400, detail=f"Command '{first}' is not in the allowed tool set.")
+
+
+PY_DENY_SUBSTRINGS = [
+    "import os", "from os", "import subprocess", "from subprocess", "import socket", "from socket", "import pathlib",
+    "open('/", "open(\"/", "__import__", "eval(", "exec(", "compile(", "input(", "requests", "urllib", "httpx",
+    "shutil", "pickle", "marshal", "ctypes", "multiprocessing", "threading", "asyncio", "pty", "sys.modules",
+]
+
+
+def validate_python(code: str) -> None:
+    if not code or len(code) > 20_000:
+        raise HTTPException(status_code=400, detail="Python code is empty or too long.")
+    lowered = code.lower()
+    for s in PY_DENY_SUBSTRINGS:
+        if s in lowered:
+            raise HTTPException(status_code=400, detail=f"Python rejected by safety policy: {s}")
+    try:
+        tree = ast.parse(code)
+    except SyntaxError as e:
+        raise HTTPException(status_code=400, detail=f"Python syntax error: {e}")
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.Import, ast.ImportFrom)):
+            names = [alias.name.split(".")[0] for alias in getattr(node, "names", [])]
+            if isinstance(node, ast.ImportFrom) and node.module:
+                names.append(node.module.split(".")[0])
+            blocked = {"os", "subprocess", "socket", "pathlib", "shutil", "requests", "urllib", "httpx", "pickle", "marshal", "ctypes", "sys"}
+            hit = blocked.intersection(names)
+            if hit:
+                raise HTTPException(status_code=400, detail=f"Blocked import(s): {', '.join(sorted(hit))}")
+
+
+class IngestRequest(BaseModel):
+    kind: str = Field(..., description="Event class: correction, artifact, decision, benchmark, repo_event, user_preference, failure, success, note.")
+    summary: str = Field(..., description="Short reusable summary.")
+    payload: Dict[str, Any] = Field(default_factory=dict, description="Raw structured evidence.")
+    tags: List[str] = Field(default_factory=list)
+    utility: float = Field(default=0.5, ge=0, le=1)
+
+
+class LearnRequest(BaseModel):
+    title: str
+    source_kind: str = "note"
+    observation: str
+    reusable_rule: str
+    evidence: Dict[str, Any] = Field(default_factory=dict)
+    tags: List[str] = Field(default_factory=list)
+    utility: float = Field(default=0.7, ge=0, le=1)
+
+
+class RecallRequest(BaseModel):
+    query: str
+    tags: List[str] = Field(default_factory=list)
+    limit: int = Field(default=8, ge=1, le=30)
+
+
+class TerminalRequest(BaseModel):
+    command: str = Field(..., description="Constrained shell command executed in /data/workspace. No unrestricted RCE, no network fetch tools, no destructive filesystem commands.")
+    cwd: str = Field(default=".", description="Relative workspace directory.")
+    timeout_seconds: int = Field(default=DEFAULT_TIMEOUT, ge=1, le=MAX_TIMEOUT)
+
+
+class PythonRunRequest(BaseModel):
+    code: str = Field(..., description="Constrained Python experiment executed as an isolated subprocess in /data/workspace.")
+    timeout_seconds: int = Field(default=DEFAULT_TIMEOUT, ge=1, le=MAX_TIMEOUT)
+
+
+class WriteFileRequest(BaseModel):
+    path: str
+    content: str
+    overwrite: bool = True
+
+
+class ReadFileRequest(BaseModel):
+    path: str
+    max_bytes: int = Field(default=200_000, ge=1, le=MAX_FILE_BYTES)
+
+
+class ListFilesRequest(BaseModel):
+    path: str = "."
+    max_results: int = Field(default=200, ge=1, le=1000)
+
+
+class ArtifactRequest(BaseModel):
+    title: str
+    artifact_type: str = Field(..., description="memo, spec, prompt, report, schema, receipt, runbook, readme, dataset_note.")
+    body: str
+    metadata: Dict[str, Any] = Field(default_factory=dict)
+
+
+@app.get("/", operation_id="root")
+def root() -> Dict[str, Any]:
+    return {
+        "name": APP_NAME,
+        "version": APP_VERSION,
+        "openapi": "/openapi.json",
+        "workspace": str(WORKSPACE),
+        "no_api_key": True,
+        "tools": [
+            "/terminal/run", "/python/run", "/files/write", "/files/read", "/files/list",
+            "/learn", "/recall", "/artifact/compile", "/ledger/recent", "/receipt/{receipt_id}",
+        ],
+        "safety_boundary": "No unrestricted public shell. Execution is workspace-contained, timed, output-capped, and policy-filtered.",
+    }
+
+
+@app.get("/health", operation_id="health")
+def health() -> Dict[str, Any]:
+    return {
+        "ok": True,
+        "ts": now_iso(),
+        "version": APP_VERSION,
+        "workspace_exists": WORKSPACE.exists(),
+        "db_exists": DB_PATH.exists(),
+    }
+
+
+@app.post("/ingest", operation_id="ingestLearningEvent")
+def ingest(req: IngestRequest) -> Dict[str, Any]:
+    receipt = ledger("ingest:" + req.kind, req.summary, req.model_dump())
+    packet_text = f"[{req.kind}] {req.summary}\nTags: {', '.join(req.tags)}\nPayload: {json.dumps(req.payload, ensure_ascii=False)[:4000]}"
+    content_hash = sha256_text(packet_text)
+    with sqlite3.connect(DB_PATH) as con:
+        cur = con.execute(
+            "INSERT INTO memory_packets(ts, source_kind, title, packet, tags, utility, payload_json, content_hash) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+            (now_iso(), req.kind, req.summary[:180], packet_text, json.dumps(req.tags), req.utility, json.dumps(req.payload, ensure_ascii=False), content_hash),
+        )
+        packet_id = cur.lastrowid
+        con.commit()
+    return {"ok": True, "packet_id": packet_id, **receipt}
+
+
+@app.post("/learn", operation_id="learnReusableMemory")
+def learn(req: LearnRequest) -> Dict[str, Any]:
+    packet = (
+        f"TITLE: {req.title}\n"
+        f"SOURCE: {req.source_kind}\n"
+        f"OBSERVATION: {req.observation}\n"
+        f"REUSABLE_RULE: {req.reusable_rule}\n"
+        f"TAGS: {', '.join(req.tags)}\n"
+        f"EVIDENCE: {json.dumps(req.evidence, ensure_ascii=False)[:5000]}"
+    )
+    content_hash = sha256_text(packet)
+    with sqlite3.connect(DB_PATH) as con:
+        cur = con.execute(
+            "INSERT INTO memory_packets(ts, source_kind, title, packet, tags, utility, payload_json, content_hash) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+            (now_iso(), req.source_kind, req.title[:180], packet, json.dumps(req.tags), req.utility, json.dumps(req.evidence, ensure_ascii=False), content_hash),
+        )
+        packet_id = cur.lastrowid
+        con.commit()
+    receipt = ledger("learn", req.title, {"packet_id": packet_id, **req.model_dump()})
+    return {"ok": True, "packet_id": packet_id, "content_hash": content_hash, **receipt}
+
+
+@app.post("/recall", operation_id="recallRelevantMemory")
+def recall(req: RecallRequest) -> Dict[str, Any]:
+    q_terms = [t.lower() for t in re.findall(r"[a-zA-Z0-9_\-]{3,}", req.query)][:80]
+    tag_terms = [t.lower() for t in req.tags]
+    rows = []
+    with sqlite3.connect(DB_PATH) as con:
+        con.row_factory = sqlite3.Row
+        for row in con.execute("SELECT * FROM memory_packets ORDER BY utility DESC, id DESC LIMIT 500"):
+            hay = (row["title"] + " " + row["packet"] + " " + row["tags"]).lower()
+            lexical = sum(1 for t in q_terms if t in hay)
+            tag_score = sum(2 for t in tag_terms if t in hay)
+            score = float(row["utility"]) * 10 + lexical + tag_score
+            if score > 0:
+                rows.append({
+                    "id": row["id"], "ts": row["ts"], "title": row["title"], "source_kind": row["source_kind"],
+                    "utility": row["utility"], "score": round(score, 3), "tags": json.loads(row["tags"]),
+                    "packet": row["packet"], "content_hash": row["content_hash"],
+                })
+    rows.sort(key=lambda r: r["score"], reverse=True)
+    receipt = ledger("recall", req.query[:200], {"query": req.query, "returned": [r["id"] for r in rows[:req.limit]]})
+    return {"ok": True, "query": req.query, "results": rows[:req.limit], **receipt}
+
+
+@app.post("/terminal/run", operation_id="runConstrainedTerminal")
+def terminal_run(req: TerminalRequest) -> Dict[str, Any]:
+    validate_command(req.command)
+    cwd = ensure_relative_path(req.cwd)
+    cwd.mkdir(parents=True, exist_ok=True)
+    start = time.time()
+    env = {"PATH": "/usr/local/bin:/usr/bin:/bin", "HOME": str(WORKSPACE), "LANG": "C.UTF-8"}
+    try:
+        proc = subprocess.run(
+            ["/bin/bash", "-lc", req.command],
+            cwd=str(cwd), env=env, text=True, capture_output=True,
+            timeout=min(req.timeout_seconds, MAX_TIMEOUT),
+        )
+        elapsed_ms = int((time.time() - start) * 1000)
+        result = {
+            "ok": proc.returncode == 0,
+            "returncode": proc.returncode,
+            "stdout": clamp_output(proc.stdout),
+            "stderr": clamp_output(proc.stderr),
+            "elapsed_ms": elapsed_ms,
+            "cwd": str(cwd.relative_to(WORKSPACE)),
+            "policy": "constrained-no-key-terminal-v2",
+        }
+    except subprocess.TimeoutExpired as e:
+        result = {"ok": False, "returncode": 124, "stdout": clamp_output(e.stdout or ""), "stderr": "timeout", "elapsed_ms": int((time.time() - start) * 1000), "policy": "constrained-no-key-terminal-v2"}
+    receipt = ledger("terminal", req.command[:200], {"request": req.model_dump(), "result": {k: result[k] for k in result if k not in {"stdout", "stderr"}}})
+    return {**result, **receipt}
+
+
+@app.post("/python/run", operation_id="runConstrainedPython")
+def python_run(req: PythonRunRequest) -> Dict[str, Any]:
+    validate_python(req.code)
+    start = time.time()
+    with tempfile.NamedTemporaryFile("w", suffix=".py", dir=str(WORKSPACE), delete=False) as f:
+        f.write(req.code)
+        script = Path(f.name)
+    env = {"PATH": "/usr/local/bin:/usr/bin:/bin", "HOME": str(WORKSPACE), "PYTHONNOUSERSITE": "1", "LANG": "C.UTF-8"}
+    try:
+        proc = subprocess.run(
+            ["python", "-I", str(script)], cwd=str(WORKSPACE), env=env, text=True, capture_output=True,
+            timeout=min(req.timeout_seconds, MAX_TIMEOUT),
+        )
+        result = {
+            "ok": proc.returncode == 0,
+            "returncode": proc.returncode,
+            "stdout": clamp_output(proc.stdout),
+            "stderr": clamp_output(proc.stderr),
+            "elapsed_ms": int((time.time() - start) * 1000),
+            "script_path": str(script.relative_to(WORKSPACE)),
+            "policy": "constrained-python-v2",
+        }
+    except subprocess.TimeoutExpired as e:
+        result = {"ok": False, "returncode": 124, "stdout": clamp_output(e.stdout or ""), "stderr": "timeout", "elapsed_ms": int((time.time() - start) * 1000), "script_path": str(script.relative_to(WORKSPACE)), "policy": "constrained-python-v2"}
+    receipt = ledger("python", "python experiment", {"code_hash": sha256_text(req.code), "result": {k: result[k] for k in result if k not in {"stdout", "stderr"}}})
+    return {**result, **receipt}
+
+
+@app.post("/files/write", operation_id="writeWorkspaceFile")
+def write_file(req: WriteFileRequest) -> Dict[str, Any]:
+    path = ensure_relative_path(req.path)
+    if path.exists() and not req.overwrite:
+        raise HTTPException(status_code=409, detail="File exists and overwrite=false.")
+    data = req.content.encode("utf-8")
+    if len(data) > MAX_FILE_BYTES:
+        raise HTTPException(status_code=413, detail=f"File exceeds {MAX_FILE_BYTES} bytes.")
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(req.content, encoding="utf-8")
+    receipt = ledger("file_write", req.path, {"path": req.path, "bytes": len(data), "sha256": sha256_text(req.content)})
+    return {"ok": True, "path": req.path, "bytes": len(data), "sha256": sha256_text(req.content), **receipt}
+
+
+@app.post("/files/read", operation_id="readWorkspaceFile")
+def read_file(req: ReadFileRequest) -> Dict[str, Any]:
+    path = ensure_relative_path(req.path)
+    if not path.exists() or not path.is_file():
+        raise HTTPException(status_code=404, detail="File not found.")
+    raw = path.read_bytes()[: req.max_bytes]
+    text = raw.decode("utf-8", errors="replace")
+    receipt = ledger("file_read", req.path, {"path": req.path, "bytes_returned": len(raw)})
+    return {"ok": True, "path": req.path, "content": text, "bytes_returned": len(raw), "truncated": path.stat().st_size > len(raw), **receipt}
+
+
+@app.post("/files/list", operation_id="listWorkspaceFiles")
+def list_files(req: ListFilesRequest) -> Dict[str, Any]:
+    base = ensure_relative_path(req.path)
+    if not base.exists():
+        raise HTTPException(status_code=404, detail="Path not found.")
+    items = []
+    if base.is_file():
+        targets = [base]
+    else:
+        targets = list(base.rglob("*"))[: req.max_results]
+    for p in targets:
+        try:
+            st = p.stat()
+            items.append({"path": str(p.relative_to(WORKSPACE)), "type": "dir" if p.is_dir() else "file", "bytes": st.st_size, "mtime": st.st_mtime})
+        except OSError:
+            pass
+    return {"ok": True, "root": req.path, "items": items, "count": len(items)}
+
+
+@app.post("/artifact/compile", operation_id="compileArtifact")
+def compile_artifact(req: ArtifactRequest) -> Dict[str, Any]:
+    slug = re.sub(r"[^a-zA-Z0-9_.-]+", "-", req.title.lower()).strip("-")[:80] or "artifact"
+    ext = "md" if req.artifact_type.lower() not in {"json", "csv", "txt"} else req.artifact_type.lower()
+    rel = f"artifacts/{int(time.time())}-{slug}.{ext}"
+    path = ensure_relative_path(rel)
+    header = {
+        "title": req.title,
+        "artifact_type": req.artifact_type,
+        "created_at": now_iso(),
+        "metadata": req.metadata,
+    }
+    content = "---\n" + json.dumps(header, indent=2, ensure_ascii=False) + "\n---\n\n" + req.body
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(content, encoding="utf-8")
+    receipt = ledger("artifact", req.title, {"path": rel, "artifact_type": req.artifact_type, "metadata": req.metadata, "sha256": sha256_text(content)})
+    learn_req = LearnRequest(
+        title=f"Artifact compiled: {req.title}",
+        source_kind="artifact",
+        observation=f"Compiled {req.artifact_type} artifact at {rel}.",
+        reusable_rule=f"For related future work, retrieve and reuse artifact {rel} before answering.",
+        evidence={"path": rel, "metadata": req.metadata, "receipt": receipt},
+        tags=["artifact", req.artifact_type],
+        utility=0.8,
+    )
+    memory = learn(learn_req)
+    return {"ok": True, "path": rel, "sha256": sha256_text(content), "memory_packet_id": memory["packet_id"], **receipt}
+
+
+@app.get("/ledger/recent", operation_id="getRecentLedger")
+def recent_ledger(limit: int = 25) -> Dict[str, Any]:
+    limit = max(1, min(limit, 100))
+    with sqlite3.connect(DB_PATH) as con:
+        con.row_factory = sqlite3.Row
+        rows = [dict(r) for r in con.execute("SELECT id, ts, kind, summary, content_hash FROM ledger ORDER BY id DESC LIMIT ?", (limit,))]
+    return {"ok": True, "items": rows}
+
+
+@app.get("/receipt/{receipt_id}", operation_id="getReceipt")
+def get_receipt(receipt_id: int) -> Dict[str, Any]:
+    with sqlite3.connect(DB_PATH) as con:
+        con.row_factory = sqlite3.Row
+        row = con.execute("SELECT * FROM ledger WHERE id = ?", (receipt_id,)).fetchone()
+    if not row:
+        raise HTTPException(status_code=404, detail="Receipt not found.")
+    return {"ok": True, "receipt": dict(row)}
+
+
+@app.get("/workspace/tree", response_class=PlainTextResponse, operation_id="getWorkspaceTree")
+def workspace_tree() -> str:
+    lines = ["/data/workspace"]
+    for p in sorted(WORKSPACE.rglob("*"))[:500]:
+        depth = len(p.relative_to(WORKSPACE).parts)
+        prefix = "  " * depth
+        suffix = "/" if p.is_dir() else ""
+        lines.append(f"{prefix}{p.name}{suffix}")
+    return "\n".join(lines)