nprasad2077/open-terminal
1
1 Fork 0
mirror of https://github.com/open-webui/open-terminal.git synced 2026-11-04 08:10:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

refac

Browse Source
This commit is contained in:
Timothy Jaeryang Baek committed 2026-10-19 04:00:34 +09:00
1 parent f3d321fa4d
commit 6852d4d663
11 files changed
+442 -468
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

README.md
+8 -54
View File
@@ -152,62 +152,16 @@ You can also point to a specific config file:
open-terminal run --config /path/to/my-config.toml
```
### Memory and execution budgets
### Memory management
File downloads and proxy traffic stream in chunks. Uploads and generated ZIP/PDF
files use temporary files, removed on completion or cancellation. Command logs
rotate on disk without reading the whole log into memory.
File and command-output responses stream from disk, uploads are copied in chunks,
and generated ZIP/PDF files are removed after sending or cancellation. Log rotation
keeps the original newest-half behavior without loading the entire log into memory.
Search listings and ranking use temporary disk storage while preserving full matches.
Text reads preserve the original `path`, `total_lines`, and `content` response,
including complete long lines and the requested line range. The response is
built on disk and streamed, without loading the entire source into a list of
lines. Log reads return bounded pages; continue from `next_offset` when
`truncated` is true. Explicit log `tail` reads return the newest output and can
skip older entries.
These settings accept positive integers, through TOML keys or the corresponding
`OPEN_TERMINAL_` environment variable (for example, `OPEN_TERMINAL_MAX_RESPONSE_BYTES`):
| TOML key | Default | Scope |
|---|---:|---|
| `max_response_bytes` | 1048576 | Log content per page and notebook output budget |
| `max_request_bytes` | 2097152 | HTTP request bodies outside file/skill APIs |
| `max_transfer_bytes` | 1073741824 | Proxy request bodies |
| `max_command_seconds` | 3600 | Background command lifetime; separate from the output wait timeout |
Commands, notebook sessions, and HTTP requests have no global concurrency cap.
Outside file/skill APIs, oversized inputs return 413 and JSON bodies are
checked for excessive nesting/object counts before parsing. Background commands that exceed their deadline stop with an
`error` in their status response. Log I/O failures are reported there too; output
is still drained and the command's completion state is finalized.
Searches preserve complete matching lines, ignore rules, ranking, and pagination.
Git/ripgrep listings and ranked content matches use temporary disk storage rather
than collecting the whole search in memory. File reads, edits, uploads, archives,
and document extraction have no added size, line-count, cell-count, or scan caps.
Document extraction writes text to a temporary file in a separate worker, with
cleanup on completion or cancellation. File comparison retains its pre-existing
50 MiB input, 2 million characters, 50,000 lines, and 60-second limits.
Notebook kernels are stopped on cancelled execution, excessive output, or failed
startup, and idle sessions expire after 30 minutes. Ordinary Python cell errors
preserve the session. Unattached terminals expire after 60 seconds; connected
terminals remain available until they disconnect, exit, or are deleted. Terminal
I/O uses backpressure; Open Terminal does not override Uvicorn's WebSocket message
size or queue settings.
Arbitrary shell/notebook code needs OS-level isolation to keep its allocations
from exhausting the API container. On Linux, `OPEN_TERMINAL_WORKLOAD_CGROUP` can
point to an existing, writable cgroup v2 directory with a finite `memory.max`.
Launched commands, terminals, kernels, and document workers join that group
before executing; keep the API outside it. Configure an aggregate memory budget
that leaves room for the API, plus `pids.max` and `memory.swap.max` as appropriate.
The host/container supervisor must delegate that cgroup and grant access to the
worker users; Open Terminal does not create or modify host cgroups. Without it,
application budgets do **not** bound arbitrary child allocations or forks.
A memory limit on the entire container alone can still kill the API.
Hard workload memory isolation requires support from the deployment platform;
application streaming and cleanup cannot constrain arbitrary user programs.
Process, terminal, notebook, and document-worker cleanup releases readers, file
handles, and child processes on completion, cancellation, or server shutdown.
The existing command-log retention and notebook idle timeout still apply.
### File Browser Root
open_terminal/main.py
+72 -83
View File
@@ -25,20 +25,20 @@ from dataclasses import dataclass, field
from typing import Optional
from urllib.parse import quote as url_quote
from fastapi import Depends, FastAPI, HTTPException, Query, Request, WebSocket, WebSocketDisconnect
from fastapi import Depends, FastAPI, HTTPException, Query, Request, UploadFile, WebSocket, WebSocketDisconnect
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, FileResponse, StreamingResponse
from fastapi.responses import JSONResponse, StreamingResponse
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from starlette.datastructures import UploadFile as StarletteUploadFile
from starlette.formparsers import MultiPartParser, MultiPartException
from pydantic import BaseModel, Field
from starlette.formparsers import MultiPartParser, MultiPartException, parse_options_header
from pydantic import BaseModel, Field, TypeAdapter, ValidationError
from open_terminal.env import API_KEY, BINARY_FILE_MIME_PREFIXES, CORS_ALLOWED_ORIGINS, ENABLE_NOTEBOOKS, ENABLE_SYSTEM_PROMPT, ENABLE_TERMINAL, EXECUTE_DESCRIPTION, EXECUTE_TIMEOUT, FILE_BROWSER_ROOT, LOG_DIR, MAX_TERMINAL_SESSIONS, MULTI_USER, OPEN_TERMINAL_INFO, PROCESS_LOG_RETENTION, SESSION_CWD_TTL, SYSTEM_PROMPT, TERMINAL_TERM
from open_terminal.utils.runner import PipeRunner, ProcessRunner, create_runner, read_pty, write_pty, PtyRunner
from open_terminal.utils.fs import UserFS, TemporaryFileResponse, text_file_response, line_range_offsets
from open_terminal.utils.limits import (CHUNK_SIZE, ResourceLimitError, ResourceLimitsMiddleware, run_io, run_worker, worker_command)
from open_terminal.utils.fs import UserFS, FullFileResponse as FileResponse, TemporaryFileResponse, text_file_response, line_range_offsets
from open_terminal.utils.limits import (CHUNK_SIZE, NormalizeNullQueryMiddleware, run_io, run_worker)
from open_terminal.utils.file_compare import CompareRequest, run_comparison
from open_terminal.utils.search import search_entries, walk_entries, matching_files, ranked_page
from open_terminal.utils.search import search_entries, matching_files, ranked_page
from open_terminal.utils.skills import join_skill_path, list_skill_resources, parse_skill_frontmatter
MATCH_PAGE_SIZE = 100
@@ -193,7 +193,10 @@ async def _preview_pdf(target: str) -> str | None:
])
pdf = os.path.join(tmpdir, os.path.splitext(os.path.basename(target))[0] + ".pdf")
if not os.path.isfile(pdf):
return None
matches = sorted(name for name in os.listdir(tmpdir) if name.lower().endswith(".pdf"))
if not matches:
return None
pdf = os.path.join(tmpdir, matches[0])
fd, result = tempfile.mkstemp(suffix=".pdf")
os.close(fd)
try:
@@ -234,7 +237,7 @@ app = FastAPI(
description="A remote terminal API.",
version=_pkg_version("open-terminal"),
)
app.add_middleware(ResourceLimitsMiddleware)
app.add_middleware(NormalizeNullQueryMiddleware)
app.add_middleware(
CORSMiddleware,
allow_origins=[o.strip() for o in CORS_ALLOWED_ORIGINS.split(",")],
@@ -249,13 +252,6 @@ async def permission_error_handler(request: Request, exc: PermissionError):
return JSONResponse(status_code=403, content={"detail": str(exc)})
@app.exception_handler(ResourceLimitError)
async def resource_limit_handler(request: Request, exc: ResourceLimitError):
return JSONResponse(status_code=413, content={"detail": str(exc)})
# ---------------------------------------------------------------------------
# Models
# ---------------------------------------------------------------------------
@@ -263,7 +259,7 @@ async def resource_limit_handler(request: Request, exc: ResourceLimitError):
class ExecRequest(BaseModel):
command: str = Field(
..., max_length=16384,
...,
description="Shell command to execute. Supports chaining (&&, ||, ;), pipes (|), and redirections.",
json_schema_extra={"examples": ["echo hello", "ls -la && whoami"]},
)
@@ -375,7 +371,6 @@ class BackgroundProcess:
runner: ProcessRunner
status: str = "running"
exit_code: Optional[int] = None
error: Optional[str] = None
log_task: Optional[asyncio.Task] = field(default=None, repr=False)
input_lock: asyncio.Lock = field(default_factory=asyncio.Lock, repr=False)
finished_at: Optional[float] = field(default=None, repr=False)
@@ -420,12 +415,10 @@ def _set_session_cwd(session_id: str | None, path: str):
"""Store a session's cwd. No-op if *session_id* is ``None``."""
if session_id:
_expire_session_cwds()
if session_id not in _session_cwds and len(_session_cwds) >= 1024:
del _session_cwds[min(_session_cwds, key=lambda key: _session_cwds[key][1])]
_session_cwds[session_id] = (path, time.time())
from open_terminal.utils.log import log_process, read_log
from open_terminal.utils.log import log_process, log_response
@@ -433,9 +426,8 @@ from open_terminal.utils.log import log_process, read_log
def _cleanup_expired():
now = time.time()
finished = [p for p in _processes.values() if p.finished_at is not None]
finished.sort(key=lambda p: p.finished_at)
for index, process in enumerate(finished):
if now - process.finished_at > _EXPIRY_SECONDS or index < len(finished) - 256:
for process in finished:
if now - process.finished_at > _EXPIRY_SECONDS:
_processes.pop(process.id, None)
@@ -461,8 +453,7 @@ async def _maintenance_loop():
_expire_session_cwds()
if ENABLE_TERMINAL:
for sid, session in list(_terminal_sessions.items()):
idle = time.time() - session["last_active"]
if not _session_is_alive(session) or (not session["connected"] and idle > 60):
if not _session_is_alive(session):
await _cleanup_session(sid)
active = {p.log_path for p in _processes.values() if p.finished_at is None}
try:
@@ -567,14 +558,15 @@ async def _list_skills_in_dir(fs: UserFS, root: str, scope: str) -> list[SkillSu
name = (frontmatter.get("name") or dirname).strip()
if not name:
continue
skill = SkillSummary(
id=f"terminal:{url_quote(name, safe='')}",
name=name,
description=description[:1024],
location=skill_path,
scope=scope,
skills.append(
SkillSummary(
id=f"terminal:{url_quote(name, safe='')}",
name=name,
description=description[:1024],
location=skill_path,
scope=scope,
)
)
skills.append(skill)
return skills
@@ -758,13 +750,15 @@ async def read_file(
request=http_request, timeout=None,
))
if "error" in result:
if result.get("error_type") == "PermissionError":
raise PermissionError(result["error"])
raise ValueError(result["error"])
return await text_file_response(extracted_path, target, start_line, end_line, False)
finally:
os.unlink(extracted_path)
if any(mime.startswith(prefix) for prefix in BINARY_FILE_MIME_PREFIXES):
return FileResponse(target, media_type=mime)
raise HTTPException(415, f"Unsupported binary file type: {mime}")
raise HTTPException(415, f"Unsupported binary file type: {mime} ({os.path.getsize(target)} bytes)")
@app.get(
@@ -1347,7 +1341,7 @@ class UploadParser(MultiPartParser):
return await super().parse()
except BaseException:
# Starlette closes these on MultiPartException only. Disconnects,
# cancellation, and body-limit errors must release spools as well.
# cancellation and other read errors must release spools as well.
for file in self._files_to_close_on_error:
file.close()
raise
@@ -1371,13 +1365,24 @@ async def upload_file(
fs: UserFS = Depends(get_filesystem),
):
try:
form = await UploadParser(http_request.headers, http_request.stream()).parse()
content_type, _ = parse_options_header(http_request.headers.get("content-type"))
if content_type == b"multipart/form-data":
form = await UploadParser(http_request.headers, http_request.stream()).parse()
else:
form = await http_request.form()
except MultiPartException as exc:
raise HTTPException(400, exc.message) from None
file = form.get("file")
if not isinstance(file, StarletteUploadFile):
try:
if file is None or file == "":
raise RequestValidationError([{"type": "missing", "loc": ("body", "file"), "msg": "Field required", "input": None}])
file = TypeAdapter(UploadFile).validate_python(file)
except ValidationError as exc:
await form.close()
raise HTTPException(422, "One multipart file named 'file' is required.")
raise RequestValidationError([{**error, "loc": ("body", "file", *error["loc"])} for error in exc.errors(include_url=False)]) from exc
except BaseException:
await form.close()
raise
filename = os.path.basename(file.filename or "upload")
directory = fs.resolve_path(directory)
path = os.path.normpath(os.path.join(directory, filename))
@@ -1438,20 +1443,28 @@ async def archive_paths(
os.close(fd)
stop = threading.Event()
def archive_files(target):
if os.path.isfile(target):
yield target
elif os.path.isdir(target):
for directory, dirs, files in os.walk(target):
dirs[:] = [name for name in dirs if fs.is_path_allowed(os.path.join(directory, name))]
for name in files:
full = os.path.join(directory, name)
if fs.is_path_allowed(full):
yield full
def _build_zip():
with zipfile.ZipFile(temporary, "w", zipfile.ZIP_DEFLATED) as archive:
for target in resolved:
fs._check_path(target)
entries = [(target, "file")] if os.path.isfile(target) else walk_entries(fs, target)
for full, entry_type in entries:
if entry_type != "file":
continue
for full in archive_files(target):
if stop.is_set():
return
arcname = os.path.basename(full) if full == target else os.path.join(os.path.basename(target.rstrip("/\\")) or "dir", os.path.relpath(full, target))
info = zipfile.ZipInfo.from_file(full, arcname)
info.compress_type = zipfile.ZIP_DEFLATED
with open(full, "rb") as source, archive.open(info, "w", force_zip64=True) as destination:
with open(full, "rb") as source, archive.open(info, "w") as destination:
while chunk := source.read(CHUNK_SIZE):
if stop.is_set():
return
@@ -1460,7 +1473,7 @@ async def archive_paths(
task = asyncio.create_task(asyncio.to_thread(_build_zip))
try:
await asyncio.shield(task)
return TemporaryFileResponse(temporary, media_type="application/zip", filename=f"{archive_name}.zip")
return TemporaryFileResponse(temporary, media_type="application/zip", headers={"Content-Disposition": f'attachment; filename="{archive_name}.zip"'})
except BaseException:
stop.set()
await asyncio.gather(task, return_exceptions=True)
@@ -1488,10 +1501,9 @@ async def list_processes(http_request: Request):
return [
{
"id": background_process.id,
"command": background_process.command[:1024],
"command": background_process.command,
"status": background_process.status,
"exit_code": background_process.exit_code,
"error": background_process.error,
"log_path": background_process.log_path,
}
for background_process in _processes.values()
@@ -1554,22 +1566,14 @@ async def execute(
except asyncio.TimeoutError:
pass
output, next_offset, truncated = await read_log(
background_process.log_path, offset=0, tail=tail
)
return {
return await log_response(background_process.log_path, {
"id": process_id,
"command": request.command,
"cwd": cwd,
"status": background_process.status,
"exit_code": background_process.exit_code,
"error": background_process.error,
"output": output,
"truncated": truncated,
"next_offset": next_offset,
"log_path": background_process.log_path,
}
}, offset=0, tail=tail)
@app.get(
@@ -1615,21 +1619,13 @@ async def get_status(
except asyncio.TimeoutError:
pass
output, next_offset, truncated = await read_log(
background_process.log_path, offset=offset, tail=tail
)
return {
return await log_response(background_process.log_path, {
"id": background_process.id,
"command": background_process.command,
"status": background_process.status,
"exit_code": background_process.exit_code,
"error": background_process.error,
"output": output,
"truncated": truncated,
"next_offset": next_offset,
"log_path": background_process.log_path,
}
}, offset=offset, tail=tail)
@app.post(
@@ -1654,15 +1650,13 @@ async def send_input(process_id: str, body: InputRequest, http_request: Request)
text = body.input.encode("raw_unicode_escape").decode("unicode_escape")
try:
async with asyncio.timeout(5), background_process.input_lock:
async with background_process.input_lock:
if isinstance(background_process.runner, PtyRunner):
await write_pty(background_process.runner._master_fd, text.encode())
else:
background_process.runner.write_input(text.encode())
if isinstance(background_process.runner, PipeRunner):
await background_process.runner.drain_input()
except asyncio.TimeoutError:
raise HTTPException(status_code=408, detail="Process is not accepting input")
except (BrokenPipeError, ConnectionResetError, OSError):
raise HTTPException(status_code=400, detail="Process stdin is closed")
@@ -1686,15 +1680,13 @@ async def kill_process(
force: bool = Query(False, description="Send SIGKILL instead of SIGTERM."),
):
background_process = _get_process(process_id, http_request)
if background_process.finished_at is None:
if background_process.status == "running":
background_process.status = "killed"
background_process.runner.kill(force=force)
try:
await asyncio.wait_for(asyncio.shield(background_process.log_task), 3)
except asyncio.TimeoutError:
background_process.runner.kill(force=True)
await asyncio.shield(background_process.log_task)
return {"status": background_process.status}
# The log task owns draining, reaping and closing the runner.
await asyncio.shield(background_process.log_task)
background_process.finished_at = time.time()
return {"status": "killed"}
# ---------------------------------------------------------------------------
@@ -1806,7 +1798,7 @@ async def port_proxy(port: int, path: str, request: Request):
)
response_headers = dict(upstream.headers)
for h in ("transfer-encoding", "connection"):
for h in ("transfer-encoding", "connection", "content-encoding", "content-length"):
response_headers.pop(h, None)
class ProxyResponse(StreamingResponse):
@@ -1816,7 +1808,7 @@ async def port_proxy(port: int, path: str, request: Request):
finally:
await upstream.aclose()
return ProxyResponse(upstream.aiter_raw(), status_code=upstream.status_code, headers=response_headers)
return ProxyResponse(upstream.aiter_bytes(), status_code=upstream.status_code, headers=response_headers)
# ---------------------------------------------------------------------------
@@ -1949,7 +1941,7 @@ if ENABLE_TERMINAL:
spawn_env = os.environ.copy()
spawn_env.setdefault("TERM", TERMINAL_TERM)
process = subprocess.Popen(
worker_command(shell_cmd),
shell_cmd,
stdin=slave_fd,
stdout=slave_fd,
stderr=slave_fd,
@@ -1999,7 +1991,6 @@ if ENABLE_TERMINAL:
output=bytearray(),
connected=False,
input_lock=asyncio.Lock(),
last_active=time.time(),
)
return {
"id": session_id,
@@ -2031,7 +2022,6 @@ if ENABLE_TERMINAL:
async def _write_terminal(session: dict, data: bytes):
session["last_active"] = time.time()
async with session["input_lock"]:
if session["backend"] == "pty":
await write_pty(session["master_fd"], data)
@@ -2197,7 +2187,6 @@ if ENABLE_TERMINAL:
data = await read_pty(master_fd) if backend == "pty" else await loop.run_in_executor(None, _blocking_read)
if not data:
break
session["last_active"] = time.time()
session["output"].extend(data)
del session["output"][:-65536]
await ws.send_bytes(data)
open_terminal/utils/documents.py
+1 -1
View File
@@ -322,5 +322,5 @@ if __name__ == "__main__":
output.write(extractor(path))
result = {}
except Exception as exc:
result = {"error": str(exc)}
result = {"error": str(exc), "error_type": type(exc).__name__}
json.dump(result, sys.stdout)
open_terminal/utils/file_compare.py
+14 -6
View File
@@ -105,7 +105,7 @@ def extract(path):
raise ValueError("Unsupported binary file.")
text = text.replace("\r\n", "\n").replace("\r", "\n")
if len(text) > MAX_TEXT_CHARS:
raise ValueError("Extracted text exceeds the comparison limit.")
raise ValueError("Extracted text exceeds the comparison limit (2 million characters / 50,000 lines). Nothing was truncated.")
line_breaks = "\n\v\f\x1c\x1d\x1e\x85\u2028\u2029"
line_count = sum(1 for _ in re.finditer("[" + line_breaks + "]", text)) + bool(text and text[-1] not in line_breaks)
if line_count > MAX_LINES:
@@ -232,11 +232,19 @@ async def run_comparison(request, payload, fs, cwd=None):
}
else:
command = ["sudo", "-n", "-u", fs.username, "--", *command]
stdout = await run_worker(
command,
json.dumps({"original": paths[0], "revised": paths[1], "ignore_whitespace": payload.ignore_whitespace}).encode(),
request=request, timeout=TIMEOUT_SECONDS, **process_options,
)
try:
stdout = await run_worker(
command,
json.dumps({"original": paths[0], "revised": paths[1], "ignore_whitespace": payload.ignore_whitespace}).encode(),
request=request, timeout=TIMEOUT_SECONDS, **process_options,
)
except HTTPException as exc:
messages = {
408: "Comparison exceeded the 60 second limit. Try smaller files.",
422: "Comparison failed. Check that the document libraries are installed and the files are readable.",
499: "Comparison cancelled.",
}
raise HTTPException(exc.status_code, messages.get(exc.status_code, exc.detail)) from exc
result = json.loads(stdout)
if "error" in result:
raise HTTPException(422, result["error"])
open_terminal/utils/fs.py
+29 -19
View File
@@ -27,7 +27,20 @@ from starlette.responses import FileResponse
from open_terminal.utils.limits import CHUNK_SIZE, run_io
class TemporaryFileResponse(FileResponse):
class FullFileResponse(FileResponse):
"""Stream the original full response without adding HTTP range/cache semantics."""
def set_stat_headers(self, stat_result):
self.headers.setdefault("content-length", str(stat_result.st_size))
async def __call__(self, scope, receive, send):
if "accept-ranges" in self.headers:
del self.headers["accept-ranges"]
scope = {**scope, "headers": [(key, value) for key, value in scope.get("headers", []) if key.lower() not in (b"range", b"if-range")]}
await super().__call__(scope, receive, send)
class TemporaryFileResponse(FullFileResponse):
"""Remove generated files even when sending the response is cancelled."""
async def __call__(self, scope, receive, send):
@@ -257,10 +270,6 @@ class UserFS:
async with aiofiles.open(path, "r", encoding=encoding, errors="strict") as f:
return await f.read()
async def read_lines(self, path, start_line=None, end_line=None):
self._check_path(path)
return await run_io(read_text_range, path, start_line, end_line)
async def exists(self, path: str) -> bool:
"""Check if *path* exists."""
self._check_path(path)
@@ -309,11 +318,7 @@ class UserFS:
self._check_path(path)
def _list_sync():
entries = []
with os.scandir(path) as scan:
names = []
for entry in scan:
names.append(entry.name)
for name in sorted(names):
for name in sorted(os.listdir(path)):
full = os.path.join(path, name)
if not self.is_path_allowed(full):
continue
@@ -339,15 +344,20 @@ class UserFS:
"""
self._check_path(path)
def _walk_filtered():
from open_terminal.utils.search import walk_entries
directories = {path: ([], [])}
for full, kind in walk_entries(self, path):
parent = os.path.dirname(full)
dirs, files = directories.setdefault(parent, ([], []))
(dirs if kind == "directory" else files).append(os.path.basename(full))
if kind == "directory":
directories.setdefault(full, ([], []))
return [(directory, dirs, files) for directory, (dirs, files) in directories.items()]
result = []
for dirpath, dirnames, filenames in os.walk(path):
# Prune directories belonging to other users (in-place
# modification prevents os.walk from descending into them).
dirnames[:] = [
d for d in dirnames
if self.is_path_allowed(os.path.join(dirpath, d))
]
filenames = [
f for f in filenames
if self.is_path_allowed(os.path.join(dirpath, f))
]
result.append((dirpath, dirnames, filenames))
return result
return await run_io(_walk_filtered)
# ------------------------------------------------------------------
open_terminal/utils/limits.py
+5 -89
View File
@@ -1,32 +1,11 @@
"""Memory budgets shared by HTTP handlers and subprocess workers."""
"""Streaming I/O helpers with cancellation-safe resource cleanup."""
import asyncio
import os
from pathlib import Path
import sys
from fastapi import HTTPException
from starlette.responses import JSONResponse
from open_terminal import config
def positive_setting(name: str, default: int) -> int:
value = int(os.environ.get(f"OPEN_TERMINAL_{name.upper()}", config.get(name, default)))
if value <= 0:
raise ValueError(f"OPEN_TERMINAL_{name.upper()} must be positive")
return value
CHUNK_SIZE = 64 * 1024
MAX_RESPONSE_BYTES = positive_setting("max_response_bytes", 1024 * 1024)
MAX_REQUEST_BYTES = positive_setting("max_request_bytes", 2 * 1024 * 1024)
MAX_TRANSFER_BYTES = positive_setting("max_transfer_bytes", 1024 * 1024 * 1024)
MAX_COMMAND_SECONDS = positive_setting("max_command_seconds", 3600)
class ResourceLimitError(ValueError):
pass
async def run_io(function, *args):
@@ -42,7 +21,7 @@ async def run_io(function, *args):
raise
class ResourceLimitsMiddleware:
class NormalizeNullQueryMiddleware:
def __init__(self, app):
self.app = app
@@ -53,67 +32,7 @@ class ResourceLimitsMiddleware:
cleaned = [(key, value) for key, value in params if value.lower() != "null"]
if len(cleaned) != len(params):
scope = {**scope, "query_string": urlencode(cleaned).encode()}
if scope["type"] != "http" or (scope["path"] == "/health" or scope["path"].startswith(("/files/", "/skills"))):
return await self.app(scope, receive, send)
limit = MAX_TRANSFER_BYTES if scope["path"] == "/files/upload" or scope["path"].startswith("/proxy/") else MAX_REQUEST_BYTES
headers = dict(scope.get("headers", []))
try:
size = int(headers.get(b"content-length", b"0"))
except ValueError:
size = limit + 1
if size < 0 or size > limit:
return await JSONResponse({"detail": "Request body exceeds the size limit."}, status_code=413)(scope, receive, send)
received = 0
quoted = escaped = False
depth = tokens = 0
is_json = limit == MAX_REQUEST_BYTES
async def limited_receive():
nonlocal received, quoted, escaped, depth, tokens
message = await receive()
body = message.get("body", b"")
received += len(body)
if received > limit:
raise HTTPException(413, "Request body exceeds the size limit.")
if is_json:
# Check structure before json.loads can expand a small body into
# hundreds of thousands of dicts/lists. String contents are free.
for byte in body:
if quoted:
if escaped:
escaped = False
elif byte == 92:
escaped = True
elif byte == 34:
quoted = False
elif byte == 34:
quoted = True
elif byte in (123, 91):
depth += 1
tokens += 1
elif byte in (125, 93):
depth -= 1
elif byte in (44, 58):
tokens += 1
if depth > 64 or tokens > 10_000:
raise HTTPException(413, "JSON structure exceeds the complexity limit.")
return message
await self.app(scope, limited_receive, send)
def worker_command(command: list[str]) -> list[str]:
"""Join the configured workload cgroup before exec, without preexec_fn."""
if os.name == "nt":
return command
cgroup = os.environ.get("OPEN_TERMINAL_WORKLOAD_CGROUP", "")
if cgroup:
path = Path(cgroup)
if (path / "memory.max").read_text().strip() == "max":
raise ValueError("OPEN_TERMINAL_WORKLOAD_CGROUP must have a finite memory.max")
if not cgroup:
return command
return [sys.executable, str(Path(__file__).with_name("worker.py")), cgroup, *command]
await self.app(scope, receive, send)
async def run_worker(command, payload=None, *, request=None, timeout=60, **options):
@@ -121,13 +40,10 @@ async def run_worker(command, payload=None, *, request=None, timeout=60, **optio
import signal
import subprocess
worker_env = dict(options.pop("env", os.environ))
# Document extraction does not need BLAS parallelism.
worker_env.update(OPENBLAS_NUM_THREADS="1", OMP_NUM_THREADS="1", MKL_NUM_THREADS="1", MALLOC_ARENA_MAX="2")
process = await asyncio.create_subprocess_exec(
*worker_command(command), stdin=asyncio.subprocess.PIPE,
*command, stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE, stderr=subprocess.DEVNULL,
start_new_session=os.name != "nt", env=worker_env, **options,
start_new_session=os.name != "nt", **options,
)
async def exchange():
open_terminal/utils/log.py
+97 -56
View File
@@ -7,23 +7,20 @@ import logging
import os
import shutil
import time
import tempfile
from typing import Optional
import aiofiles
import aiofiles.os
from open_terminal.env import MAX_PROCESS_LOG_SIZE, LOG_FLUSH_INTERVAL, LOG_FLUSH_BUFFER
from open_terminal.utils.limits import CHUNK_SIZE, MAX_COMMAND_SECONDS, MAX_RESPONSE_BYTES, ResourceLimitError, run_io
from open_terminal.utils.limits import CHUNK_SIZE, run_io
log = logging.getLogger(__name__)
MAX_RECORD_BYTES = CHUNK_SIZE
MAX_LOG_ENTRIES = 1000
def _records(stream):
while line := stream.readline(MAX_RECORD_BYTES + 1):
if len(line) > MAX_RECORD_BYTES:
raise ResourceLimitError("Process log contains an oversized record.")
for line in stream:
try:
yield json.loads(line)
except (ValueError, UnicodeError):
@@ -34,21 +31,21 @@ def _rotate_log(path: str) -> int:
temporary = path + ".rotating"
try:
with open(path, "rb") as source:
midpoint = os.fstat(source.fileno()).st_size // 2
split = sum(1 for _ in source) // 2
source.seek(0)
offset = 0
for record in _records(source):
for _ in range(split):
record = json.loads(source.readline())
if record.get("type") == "log_rotated":
offset = record.get("offset", 0)
elif record.get("type") in ("stdout", "stderr", "output"):
offset += 1
if source.tell() >= midpoint:
break
kept_size = os.fstat(source.fileno()).st_size - source.tell()
with open(temporary, "wb") as target:
target.write((json.dumps({"type": "log_rotated", "offset": offset, "ts": time.time()}) + "\n").encode())
shutil.copyfileobj(source, target, CHUNK_SIZE)
size = target.tell()
os.replace(temporary, path)
return size
return kept_size
finally:
try:
os.unlink(temporary)
@@ -64,7 +61,6 @@ class BoundedLogWriter:
self._log_path = log_path
self._bytes_written = 0
self.rotated = False
self.error = None
self._flush_interval = flush_interval
self._flush_buffer = flush_buffer
self._unflushed = 0
@@ -72,8 +68,7 @@ class BoundedLogWriter:
self._lock = asyncio.Lock()
async def _disable(self, exc):
self.error = str(exc)[:500]
log.warning("Disabling process log %s: %s", self._log_path, self.error)
log.warning("Disabling process log %s: %s", self._log_path, exc)
await self.close()
async def write(self, data: str) -> None:
@@ -82,8 +77,6 @@ class BoundedLogWriter:
return
try:
size = len(data.encode("utf-8", errors="replace"))
if size > MAX_RECORD_BYTES:
raise ResourceLimitError("Process log record exceeds the size limit.")
if self._bytes_written + size > MAX_PROCESS_LOG_SIZE:
await self._rotate()
await self._file.write(data)
@@ -129,24 +122,22 @@ async def log_process(background_process) -> None:
if background_process.log_path:
try:
await aiofiles.os.makedirs(os.path.dirname(background_process.log_path), exist_ok=True)
file = await aiofiles.open(background_process.log_path, "w", encoding="utf-8")
file = await aiofiles.open(background_process.log_path, "a", encoding="utf-8")
writer = BoundedLogWriter(file, background_process.log_path,
flush_interval=LOG_FLUSH_INTERVAL, flush_buffer=LOG_FLUSH_BUFFER)
await writer.write(json.dumps({"type": "start", "command": background_process.command[:4096], "pid": runner.pid, "ts": time.time()}) + "\n")
await file.write(json.dumps({"type": "start", "command": background_process.command, "pid": runner.pid, "ts": time.time()}) + "\n")
await writer.flush()
except OSError as exc:
log.warning("Cannot open process log: %s", exc)
async with asyncio.timeout(MAX_COMMAND_SECONDS):
await runner.read_output(writer)
background_process.exit_code = await runner.wait()
await runner.read_output(writer)
background_process.exit_code = await runner.wait()
except asyncio.CancelledError:
background_process.status = "killed"
runner.kill(force=True)
raise
except Exception as exc:
background_process.status = "error"
background_process.error = ("Command exceeded its execution deadline." if isinstance(exc, TimeoutError) else str(exc)[:500])
runner.kill(force=True)
log.warning("Command %s failed: %s", background_process.id, background_process.error)
log.warning("Command %s failed: %s", background_process.id, exc)
finally:
try:
background_process.exit_code = await runner.wait()
@@ -161,17 +152,12 @@ async def log_process(background_process) -> None:
await writer.flush()
finally:
await writer.close()
if writer.error:
background_process.error = "Output logging stopped: " + writer.error
def _read_log(path, offset, tail):
selected = deque()
total = 0
size = 0
selected = deque(maxlen=tail)
total = available = 0
truncated = False
next_offset = offset
limit = min(tail, MAX_LOG_ENTRIES) if tail is not None else MAX_LOG_ENTRIES
try:
source = open(path, "rb")
except FileNotFoundError:
@@ -181,35 +167,90 @@ def _read_log(path, offset, tail):
kind = record.get("type")
if kind == "log_rotated":
total = record.get("offset", 0)
truncated |= offset < total
next_offset = max(next_offset, total)
truncated = offset < total
elif kind in ("stdout", "stderr", "output"):
index = total
if total >= offset:
selected.append({"type": kind, "data": record["data"]})
available += 1
total += 1
if index < offset:
continue
entry = {"type": kind, "data": record["data"]}
entry_size = len(json.dumps(entry, ensure_ascii=False).encode())
if entry_size > MAX_RESPONSE_BYTES:
raise ResourceLimitError("A log entry exceeds the response byte limit.")
if tail is None and (len(selected) >= limit or size + entry_size > MAX_RESPONSE_BYTES):
return list(selected), index, True
selected.append(entry)
size += entry_size
while len(selected) > limit or size > MAX_RESPONSE_BYTES:
removed = selected.popleft()
size -= len(json.dumps(removed, ensure_ascii=False).encode())
truncated = True
next_offset = total
return list(selected), max(next_offset, total), truncated
return list(selected), total, truncated or available > len(selected)
async def read_log(log_path: Optional[str], offset: int = 0, tail: Optional[int] = None) -> tuple[list[dict], int, bool]:
"""Return a byte-bounded page and the next unread absolute cursor.
Explicit tail reads select the newest bounded entries and may skip older
output. Default reads paginate forward without silently losing entries.
"""
"""Read the requested output entries with the original absolute cursor semantics."""
if not log_path:
return [], 0, False
return await run_io(_read_log, log_path, offset, tail)
def _build_log_response(path, metadata, offset, tail):
"""Serialize complete output to disk without retaining a list of all entries."""
from open_terminal.utils.fs import TemporaryFileResponse
fd, temporary = tempfile.mkstemp(suffix=".json")
try:
with os.fdopen(fd, "w", encoding="utf-8") as output:
output.write(json.dumps(metadata, ensure_ascii=False)[:-1] + ',"output":[')
total = available = 0
truncated = False
try:
source = open(path, "rb") if path else None
except FileNotFoundError:
source = None
if source is not None:
with source:
# Hold this generation's descriptor across both passes. A
# rotation replaces the pathname but cannot change our data.
end = os.fstat(source.fileno()).st_size
def records():
while source.tell() < end:
line = source.readline(end - source.tell())
try:
yield json.loads(line)
except (ValueError, UnicodeError):
continue
for record in records():
kind = record.get("type")
if kind == "log_rotated":
total = record.get("offset", 0)
truncated = offset < total
elif kind in ("stdout", "stderr", "output"):
available += total >= offset
total += 1
skip = max(0, available - tail) if tail is not None else 0
truncated |= skip > 0
source.seek(0)
index = 0
separator = ""
for record in records():
kind = record.get("type")
if kind == "log_rotated":
index = record.get("offset", 0)
elif kind in ("stdout", "stderr", "output"):
if index >= offset:
if skip:
skip -= 1
else:
output.write(separator)
json.dump({"type": kind, "data": record["data"]}, output, ensure_ascii=False)
separator = ","
index += 1
output.write('],"next_offset":' + str(total) + ',"truncated":' + json.dumps(truncated) + '}')
return TemporaryFileResponse(temporary, media_type="application/json")
except BaseException:
os.unlink(temporary)
raise
async def log_response(path, metadata, offset=0, tail=None):
task = asyncio.create_task(asyncio.to_thread(_build_log_response, path, metadata, offset, tail))
try:
return await asyncio.shield(task)
except asyncio.CancelledError:
try:
response = await task
except Exception:
pass
else:
os.unlink(response.path)
raise
open_terminal/utils/notebooks.py
+211 -137
View File
@@ -1,66 +1,65 @@
"""Bounded notebook sessions with explicit kernel ownership and cleanup."""
"""Jupyter notebook execution endpoints.
Provides per-cell execution with multi-session support. Each session gets its
own Jupyter kernel via nbclient. Requires the ``notebooks`` optional extra::
pip install open-terminal[notebooks]
"""
import asyncio
import json
import os
import time
import uuid
from typing import Optional
import aiofiles
import nbformat
from nbclient import NotebookClient
from nbclient.exceptions import CellExecutionError
from jupyter_client import AsyncKernelManager
from jupyter_client.asynchronous.client import AsyncKernelClient
from jupyter_client.connect import channel_socket_types
import zmq
from fastapi import APIRouter, Depends, HTTPException
import aiofiles.os
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field
from open_terminal.utils.limits import MAX_RESPONSE_BYTES, ResourceLimitError, worker_command
_IDLE_TIMEOUT = 30 * 60
_sessions = {}
_cleanup_task = None
import nbformat
from nbclient import NotebookClient
class LimitedKernelClient(AsyncKernelClient):
def _create_connected_socket(self, channel, identity=None):
# Set transport budgets before connecting, so one output message cannot
# allocate an arbitrary amount of memory before nbclient sees it.
sock = self.context.socket(channel_socket_types[channel])
sock.linger = 1000
sock.setsockopt(zmq.MAXMSGSIZE, MAX_RESPONSE_BYTES)
sock.setsockopt(zmq.RCVHWM, 8)
if identity:
sock.identity = identity
sock.connect(self._make_url(channel))
return sock
# ---------------------------------------------------------------------------
# Session manager
# ---------------------------------------------------------------------------
class LimitedKernelManager(AsyncKernelManager):
def format_kernel_cmd(self, extra_arguments=None):
return worker_command(super().format_kernel_cmd(extra_arguments))
class LimitedNotebookClient(NotebookClient):
output_bytes = 0
def process_message(self, msg, cell, cell_index):
kind = msg.get("msg_type", "")
if kind.startswith("comm"):
return None # Widget state is not part of the cell-output API.
if kind in {"stream", "display_data", "update_display_data", "execute_result", "error"}:
self.output_bytes += len(json.dumps(msg["content"], ensure_ascii=False).encode())
if self.output_bytes > MAX_RESPONSE_BYTES:
raise ResourceLimitError("Notebook output exceeded the byte limit; kernel stopped.")
return super().process_message(msg, cell, cell_index)
_IDLE_TIMEOUT = 30 * 60 # 30 minutes
class _Session:
def __init__(self, session_id, path, nb, client):
self.id, self.path, self.nb, self.client = session_id, path, nb, client
"""Wraps a NotebookClient for a specific notebook."""
__slots__ = ("id", "path", "nb", "client", "busy", "created_at", "last_used")
def __init__(self, session_id: str, path: str, nb, client):
self.id = session_id
self.path = path
self.nb = nb
self.client = client
self.busy = False
self.created_at = self.last_used = time.time()
self.created_at = time.time()
self.last_used = time.time()
_sessions: dict[str, _Session] = {}
_cleanup_task: Optional[asyncio.Task] = None
async def _idle_cleanup_loop():
"""Periodically remove sessions idle for more than _IDLE_TIMEOUT."""
while True:
await asyncio.sleep(60)
now = time.time()
stale = [
sid
for sid, s in _sessions.items()
if now - s.last_used > _IDLE_TIMEOUT and not s.busy
]
for sid in stale:
await _destroy_session(sid)
async def _close_client(client):
@@ -68,27 +67,22 @@ async def _close_client(client):
await client._async_cleanup_kernel()
except Exception:
# A failed handshake can leave only the manager initialized.
if client.km and await client.km.is_alive():
await client.km.shutdown_kernel(now=True)
try:
if client.km and await client.km.is_alive():
await client.km.shutdown_kernel(now=True)
except Exception:
pass
finally:
if client.kc:
client.kc.stop_channels()
async def _destroy_session(session_id):
session = _sessions.pop(session_id, None)
if session:
if session and session.client:
await _close_client(session.client)
async def _idle_cleanup_loop():
while True:
await asyncio.sleep(60)
for sid, session in list(_sessions.items()):
if not session.busy and time.time() - session.last_used > _IDLE_TIMEOUT:
try:
await _destroy_session(sid)
except Exception:
pass
def _ensure_cleanup_task():
global _cleanup_task
if _cleanup_task is None or _cleanup_task.done():
@@ -104,6 +98,11 @@ async def shutdown_notebooks():
await asyncio.gather(*(_destroy_session(sid) for sid in list(_sessions)), return_exceptions=True)
# ---------------------------------------------------------------------------
# Request / response models
# ---------------------------------------------------------------------------
class CreateSessionRequest(BaseModel):
path: str = Field(..., description="Absolute path to the .ipynb file.")
@@ -116,12 +115,14 @@ class CreateSessionResponse(BaseModel):
class ExecuteCellRequest(BaseModel):
cell_index: int = Field(..., description="Zero-based cell index to execute.")
source: str | None = Field(None, max_length=MAX_RESPONSE_BYTES)
source: Optional[str] = Field(
None, description="Override cell source. If omitted, uses the source already in the notebook."
)
class ExecuteCellResponse(BaseModel):
status: str
execution_count: int | None = None
execution_count: Optional[int] = None
outputs: list = Field(default_factory=list)
@@ -132,104 +133,177 @@ class SessionStatusResponse(BaseModel):
status: str
def create_notebooks_router(verify_api_key):
router = APIRouter(prefix="/notebooks", tags=["notebooks"], dependencies=[Depends(verify_api_key)])
# ---------------------------------------------------------------------------
# Router
# ---------------------------------------------------------------------------
@router.post("", response_model=CreateSessionResponse, operation_id="create_notebook_session", include_in_schema=False)
def create_notebooks_router(verify_api_key) -> APIRouter:
"""Create the notebooks router with the given auth dependency."""
router = APIRouter(
prefix="/notebooks",
tags=["notebooks"],
dependencies=[Depends(verify_api_key)],
)
@router.post(
"",
response_model=CreateSessionResponse,
operation_id="create_notebook_session",
summary="Create a notebook session",
description="Start a Jupyter kernel for the given notebook. Returns a session ID for subsequent execute calls.",
include_in_schema=False,
)
async def create_session(req: CreateSessionRequest):
client = None
registered = False
try:
_ensure_cleanup_task()
path = os.path.abspath(req.path)
try:
async with aiofiles.open(path, "rb") as source:
content = await source.read(MAX_RESPONSE_BYTES + 1)
except FileNotFoundError:
raise HTTPException(404, "Notebook not found")
if len(content) > MAX_RESPONSE_BYTES:
raise ResourceLimitError("Notebook exceeds the input size limit; clear saved outputs or split it.")
try:
nb = nbformat.reads(content.decode("utf-8"), as_version=4)
except Exception as exc:
raise HTTPException(400, f"Invalid notebook: {str(exc)[:500]}") from None
if len(nb.cells) > 1000:
raise ResourceLimitError("Notebook exceeds the cell count limit.")
kernel_name = nb.metadata.get("kernelspec", {}).get("name", "python3")
client = LimitedNotebookClient(nb, kernel_name=kernel_name, kernel_manager_class=LimitedKernelManager,
timeout=120, startup_timeout=30, store_widget_state=False, record_timing=False)
try:
async with asyncio.timeout(40):
client.create_kernel_manager()
client.km.client_class = "open_terminal.utils.notebooks.LimitedKernelClient"
await client.async_start_new_kernel(cwd=os.path.dirname(path))
await client.async_start_new_kernel_client()
except Exception as exc:
raise HTTPException(500, f"Failed to start kernel: {str(exc)[:500]}") from None
sid = uuid.uuid4().hex[:12]
_sessions[sid] = _Session(sid, path, nb, client)
registered = True
return CreateSessionResponse(id=sid, kernel=kernel_name, status="ready")
finally:
if client and not registered:
await _close_client(client)
@router.post("/{session_id}/execute", response_model=ExecuteCellResponse, operation_id="execute_notebook_cell", include_in_schema=False)
_ensure_cleanup_task()
path = os.path.abspath(req.path)
if not await aiofiles.os.path.isfile(path):
raise HTTPException(status_code=404, detail=f"Notebook not found: {path}")
# Read and parse the notebook
async with aiofiles.open(path, encoding="utf-8") as f:
content = await f.read()
try:
nb = nbformat.reads(content, as_version=4)
except Exception as e:
raise HTTPException(status_code=400, detail=f"Invalid notebook: {e}")
kernel_name = nb.metadata.get("kernelspec", {}).get("name", "python3")
# Start kernel — follow nbclient's setup_kernel pattern
client = NotebookClient(nb, kernel_name=kernel_name, timeout=120)
try:
client.create_kernel_manager()
await client.async_start_new_kernel()
await client.async_start_new_kernel_client()
except BaseException as e:
await _close_client(client)
if isinstance(e, asyncio.CancelledError):
raise
raise HTTPException(
status_code=500,
detail=f"Failed to start kernel '{kernel_name}': {e}",
)
session_id = uuid.uuid4().hex[:12]
_sessions[session_id] = _Session(session_id, path, nb, client)
return CreateSessionResponse(
id=session_id, kernel=kernel_name, status="ready"
)
@router.post(
"/{session_id}/execute",
response_model=ExecuteCellResponse,
operation_id="execute_notebook_cell",
summary="Execute a notebook cell",
description="Execute a single cell in the given session. Optionally override the cell source. "
"Updates the .ipynb file in place after execution.",
include_in_schema=False,
)
async def execute_cell(session_id: str, req: ExecuteCellRequest):
session = _sessions.get(session_id)
if not session:
raise HTTPException(404, "Session not found")
raise HTTPException(status_code=404, detail="Session not found")
if session.busy:
raise HTTPException(409, "Cell already executing")
if not 0 <= req.cell_index < len(session.nb.cells):
raise HTTPException(400, "Cell index out of range")
cell = session.nb.cells[req.cell_index]
previous_source = cell.source
raise HTTPException(status_code=409, detail="Cell already executing")
nb = session.nb
if req.cell_index < 0 or req.cell_index >= len(nb.cells):
raise HTTPException(
status_code=400,
detail=f"cell_index {req.cell_index} out of range (0..{len(nb.cells) - 1})",
)
cell = nb.cells[req.cell_index]
if req.source is not None:
cell.source = req.source
if len(nbformat.writes(session.nb).encode()) > 2 * MAX_RESPONSE_BYTES:
cell.source = previous_source
raise ResourceLimitError("Notebook exceeds the session size limit.")
session.busy = True
session.last_used = time.time()
session.client.output_bytes = sum(len(json.dumps(c.get("outputs", [])).encode()) for i, c in enumerate(session.nb.cells) if i != req.cell_index)
try:
async with asyncio.timeout(130):
await session.client.async_execute_cell(cell, req.cell_index)
await session.client.async_execute_cell(cell, req.cell_index)
except asyncio.CancelledError:
await _destroy_session(session_id)
raise
except CellExecutionError:
return ExecuteCellResponse(status="error", execution_count=cell.get("execution_count"), outputs=cell.get("outputs", []))
except Exception as exc:
# Errors may leave a kernel running or an IOPub queue filling.
await _destroy_session(session_id)
return ExecuteCellResponse(status="error", outputs=[{
"output_type": "error", "ename": type(exc).__name__,
"evalue": str(exc)[:1000], "traceback": [str(exc)[:1000]],
}])
except Exception as e:
# Return the error as a cell output rather than HTTP error
return ExecuteCellResponse(
status="error",
outputs=[
{
"output_type": "error",
"ename": type(e).__name__,
"evalue": str(e),
"traceback": [str(e)],
}
],
)
finally:
session.busy = False
session.last_used = time.time()
try:
async with aiofiles.open(session.path, "w", encoding="utf-8") as target:
await target.write(nbformat.writes(session.nb))
except OSError:
pass
return ExecuteCellResponse(status="ok", execution_count=cell.get("execution_count"), outputs=cell.get("outputs", []))
session.last_used = time.time()
@router.get("/{session_id}", response_model=SessionStatusResponse, operation_id="get_notebook_session", include_in_schema=False)
# Serialize outputs
outputs = []
for o in cell.outputs:
od = dict(o)
if "data" in od:
od["data"] = dict(od["data"])
outputs.append(od)
ec = cell.get("execution_count")
# Save notebook to disk
try:
nb_json = nbformat.writes(nb)
async with aiofiles.open(session.path, "w", encoding="utf-8") as f:
await f.write(nb_json)
except Exception:
pass # non-fatal
return ExecuteCellResponse(
status="ok", execution_count=ec, outputs=outputs
)
@router.get(
"/{session_id}",
response_model=SessionStatusResponse,
operation_id="get_notebook_session",
summary="Get notebook session status",
include_in_schema=False,
)
async def get_session(session_id: str):
session = _sessions.get(session_id)
if not session:
raise HTTPException(404, "Session not found")
return SessionStatusResponse(id=session.id, path=session.path,
kernel=session.nb.metadata.get("kernelspec", {}).get("name", "python3"), status="busy" if session.busy else "ready")
raise HTTPException(status_code=404, detail="Session not found")
@router.delete("/{session_id}", operation_id="delete_notebook_session", include_in_schema=False)
kernel_name = session.nb.metadata.get("kernelspec", {}).get(
"name", "python3"
)
status = "busy" if session.busy else "ready"
return SessionStatusResponse(
id=session.id, path=session.path, kernel=kernel_name, status=status
)
@router.delete(
"/{session_id}",
operation_id="delete_notebook_session",
summary="Stop a notebook session",
include_in_schema=False,
)
async def delete_session(session_id: str):
if session_id not in _sessions:
raise HTTPException(404, "Session not found")
raise HTTPException(status_code=404, detail="Session not found")
await _destroy_session(session_id)
return {"status": "stopped"}
open_terminal/utils/runner.py
+5 -7
View File
@@ -7,8 +7,6 @@ import subprocess
import time
from abc import ABC, abstractmethod
from open_terminal.utils.limits import worker_command
try:
import fcntl
import pty
@@ -109,7 +107,8 @@ class PtyRunner(ProcessRunner):
# Set a reasonable default window size (80x24).
fcntl.ioctl(slave_fd, termios.TIOCSWINSZ, struct.pack("HHHH", 24, 80, 0, 0))
self._process = subprocess.Popen(
worker_command(["/bin/sh", "-c", command]),
command,
shell=True,
stdin=slave_fd,
stdout=slave_fd,
stderr=slave_fd,
@@ -192,9 +191,8 @@ class PipeRunner(ProcessRunner):
self._env = env
async def start(self) -> None:
command = worker_command(["/bin/sh", "-c", self._command]) if os.name != "nt" else [os.environ.get("COMSPEC", "cmd.exe"), "/c", self._command]
self._process = await asyncio.create_subprocess_exec(
*command,
self._process = await asyncio.create_subprocess_shell(
self._command,
start_new_session=os.name != "nt",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
@@ -205,7 +203,7 @@ class PipeRunner(ProcessRunner):
async def read_output(self, log_file) -> None:
async def read_stream(stream, label):
while line := await stream.read(4096):
async for line in stream:
if log_file:
await log_file.write(
json.dumps(
open_terminal/utils/user_isolation.py
-2
View File
@@ -150,7 +150,5 @@ def resolve_user(user_id: str) -> tuple[str, str]:
username = sanitize_username(user_id)
home_dir = ensure_os_user(username)
if len(_user_cache) >= 1024:
_user_cache.pop(next(iter(_user_cache)))
_user_cache[user_id] = (username, home_dir)
return username, home_dir
open_terminal/utils/worker.py
-14
View File
@@ -1,14 +0,0 @@
"""Exec launcher for a supervisor-provided workload cgroup."""
import os
import sys
def main():
cgroup, *command = sys.argv[1:]
with open(os.path.join(cgroup, "cgroup.procs"), "w") as stream:
stream.write(str(os.getpid()))
os.execvp(command[0], command)
if __name__ == "__main__":
main()