# Copyright (c) 2024-2026 Karana Dynamics Pty Ltd. All rights reserved.
#
# NOTICE TO USER:
#
# This source code and/or documentation (the "Licensed Materials") is
# the confidential and proprietary information of Karana Dynamics Inc.
# Use of these Licensed Materials is governed by the terms and conditions
# of a separate software license agreement between Karana Dynamics and the
# Licensee ("License Agreement"). Unless expressly permitted under that
# agreement, any reproduction, modification, distribution, or disclosure
# of the Licensed Materials, in whole or in part, to any third party
# without the prior written consent of Karana Dynamics is strictly prohibited.
#
# THE LICENSED MATERIALS ARE PROVIDED "AS IS" WITHOUT WARRANTY OF ANY KIND.
# KARANA DYNAMICS DISCLAIMS ALL WARRANTIES, EXPRESS OR IMPLIED, INCLUDING
# BUT NOT LIMITED TO WARRANTIES OF MERCHANTABILITY, NON-INFRINGEMENT, AND
# FITNESS FOR A PARTICULAR PURPOSE.
#
# IN NO EVENT SHALL KARANA DYNAMICS BE LIABLE FOR ANY DAMAGES WHATSOEVER,
# INCLUDING BUT NOT LIMITED TO LOSS OF PROFITS, DATA, OR USE, EVEN IF
# ADVISED OF THE POSSIBILITY OF SUCH DAMAGES, WHETHER IN CONTRACT, TORT,
# OR OTHERWISE ARISING OUT OF OR IN CONNECTION WITH THE LICENSED MATERIALS.
#
# U.S. Government End Users: The Licensed Materials are a "commercial item"
# as defined at 48 C.F.R. 2.101, and are provided to the U.S. Government
# only as a commercial end item under the terms of this license.
#
# Any use of the Licensed Materials in individual or commercial software must
# include, in the user documentation and internal source code comments,
# this Notice, Disclaimer, and U.S. Government Use Provision.
"""Interactive, in-process Python console widget for Karana.WebUI."""
from __future__ import annotations
import code
from collections import deque
from contextlib import AbstractContextManager
from dataclasses import dataclass
import importlib
import json
import os
import queue
import re
import rlcompleter
import sys
import threading
from types import TracebackType
from typing import Any, Callable, TextIO
import Karana.WebUI as kw
import Karana.Core as kc
_STOP = object()
_INTERRUPT = object()
@dataclass(frozen=True)
class _CompletionRequest:
"""Describe one completion request queued for the interpreter worker."""
text: str
cursor: int
client_id: int
show_all: bool
@dataclass(frozen=True)
class _WaitMarker:
"""Signal tests after all commands queued before this marker finish."""
completed: threading.Event
@dataclass(frozen=True)
class _Prompt:
"""Describe one server-authoritative input opportunity."""
identifier: int
text: str
is_stdin: bool
class _StreamTee:
"""Mirror a text stream while preserving writes to its original destination."""
def __init__(self, original: TextIO, stream_name: str, mirror: Callable[[str, str], None]):
# Store the real stream so local terminal output retains its original behavior.
self._original = original
self._stream_name = stream_name
self._mirror = mirror
self._local = threading.local()
@property
def encoding(self) -> str | None:
"""Return the original stream encoding."""
return self._original.encoding
@property
def errors(self) -> str | None:
"""Return the original stream error policy."""
return self._original.errors
def write(self, text: str) -> int:
"""Write locally first, then forward the same text to the browser."""
written = self._original.write(text)
# Avoid recursive mirroring if WebUI or another callback writes while forwarding.
if text and not getattr(self._local, "mirroring", False):
self._local.mirroring = True
try:
self._mirror(self._stream_name, text)
finally:
self._local.mirroring = False
return written
def flush(self) -> None:
"""Flush the original stream."""
self._original.flush()
def isatty(self) -> bool:
"""Preserve terminal detection from the original stream."""
return self._original.isatty()
def fileno(self) -> int:
"""Expose the original file descriptor without intercepting direct writes."""
return self._original.fileno()
def writable(self) -> bool:
"""Report that this proxy accepts text writes."""
return True
def __getattr__(self, name: str) -> Any:
"""Delegate uncommon stream operations to the original object."""
return getattr(self._original, name)
class _NativeStreamTee:
"""Forward captured descriptor output to its original FD and the browser."""
def __init__(
self,
original_fd: int,
stream_name: str,
mirror: Callable[[str, str], None],
encoding: str,
) -> None:
# Keep a duplicate that remains pointed at the local destination after dup2 capture.
self._fd = os.dup(original_fd)
self._stream_name = stream_name
self._mirror = mirror
self._encoding = encoding
self._closed = False
def write(self, text: str) -> int:
"""Write one captured fragment locally and mirror it to the browser."""
data = text.encode(self._encoding, errors="replace")
offset = 0
while offset < len(data):
offset += os.write(self._fd, data[offset:])
if text:
self._mirror(self._stream_name, text)
return len(text)
def flush(self) -> None:
"""Provide the writable-stream interface expected by wurlitzer."""
def close(self) -> None:
"""Release the duplicate of the original output descriptor."""
if not self._closed:
self._closed = True
os.close(self._fd)
class _CLevelCapture:
"""Own a wurlitzer redirect and the local-output descriptors it targets."""
def __init__(self, mirror: Callable[[str, str], None]) -> None:
"""Start process-wide C stdout and stderr capture."""
try:
wurlitzer = importlib.import_module("wurlitzer")
except ImportError as error:
raise ImportError(
"C-level output capture requires wurlitzer; install requirements.txt "
"or pass capture_c_output=False"
) from error
encoding = sys.getdefaultencoding()
self._stdout = _NativeStreamTee(1, "stdout", mirror, encoding)
self._stderr = _NativeStreamTee(2, "stderr", mirror, encoding)
self._closed = False
self._redirect: AbstractContextManager[Any] = wurlitzer.pipes(
stdout=self._stdout,
stderr=self._stderr,
encoding=encoding,
# Avoid privileged Linux pipe-resizing attempts; the reader drains continuously.
bufsize=0,
)
try:
# Enter for the widget lifetime so native output outside REPL commands is captured.
self._redirect.__enter__()
except BaseException:
self._stdout.close()
self._stderr.close()
raise
def close(self) -> None:
"""Drain pending native output, restore descriptors, and release duplicates."""
if self._closed:
return
self._closed = True
try:
self._redirect.__exit__(None, None, None)
finally:
self._stdout.close()
self._stderr.close()
class _BrowserInput:
"""Provide blocking ``readline`` responses from browser-submitted lines."""
def __init__(
self,
open_prompt: Callable[[], None],
close_prompt: Callable[[], None],
) -> None:
# Coordinate the REPL worker with WebUI's independent server callback thread.
self._open_prompt = open_prompt
self._close_prompt = close_prompt
self._responses: queue.Queue[str | object] = queue.Queue(maxsize=1)
self._lock = threading.Lock()
self._waiting = False
self._response_offered = False
self._closed = False
@property
def waiting(self) -> bool:
"""Return whether Python is blocked waiting for browser input."""
with self._lock:
return self._waiting
def readline(self, _size: int = -1) -> str:
"""Wait for one browser line and return it with a terminating newline."""
with self._lock:
if self._closed:
return ""
self._waiting = True
self._response_offered = False
self._open_prompt()
response = self._responses.get()
with self._lock:
self._waiting = False
self._response_offered = False
self._close_prompt()
if response is _STOP:
return ""
if response is _INTERRUPT:
raise KeyboardInterrupt
return f"{response}\n"
def offer(self, line: str) -> bool:
"""Offer a line to a pending input call, accepting only the first response."""
with self._lock:
if self._closed or not self._waiting or self._response_offered:
return False
self._response_offered = True
self._responses.put(line)
return True
def interrupt(self) -> bool:
"""Interrupt a pending input call without touching an executing Python thread."""
with self._lock:
if self._closed or not self._waiting or self._response_offered:
return False
self._response_offered = True
self._responses.put(_INTERRUPT)
return True
def close(self) -> None:
"""Release a blocked reader during widget shutdown."""
with self._lock:
self._closed = True
should_release = self._waiting and not self._response_offered
if should_release:
self._response_offered = True
if should_release:
self._responses.put(_STOP)
def readable(self) -> bool:
"""Report that this proxy supports reads."""
return True
def isatty(self) -> bool:
"""Identify browser input as an interactive stream."""
return True
class _ConsoleSession:
"""Serialize browser lines through one ``InteractiveConsole`` worker."""
def __init__(self, namespace: dict[str, Any], emit: Callable[[dict[str, Any]], None]):
# InteractiveConsole keeps the supplied dictionary by reference, like ``python -i``.
self._emit = emit
self._console = code.InteractiveConsole(namespace, filename="<web-console>")
self._completer = rlcompleter.Completer(namespace)
self._commands: queue.Queue[str | object | _CompletionRequest | _WaitMarker] = queue.Queue()
self._state_lock = threading.Lock()
self._busy = False
self._more = False
self._prompt_sequence = 0
self._active_prompt: _Prompt | None = None
self._closed = False
self._input = _BrowserInput(self._openStdinPrompt, self._closeStdinPrompt)
self._worker = threading.Thread(
target=self._run,
name="karana-web-console",
daemon=True,
)
self._worker.start()
@property
def busy(self) -> bool:
"""Return whether a command is currently executing."""
with self._state_lock:
return self._busy
@property
def prompt(self) -> str:
"""Return the prompt matching the current multiline state."""
return "... " if self._more else ">>> "
@property
def prompt_id(self) -> int | None:
"""Return the identifier of the currently open input opportunity."""
with self._state_lock:
return None if self._active_prompt is None else self._active_prompt.identifier
def start(self) -> None:
"""Emit a native-style banner and initial primary prompt."""
banner = f'Python {sys.version} on {sys.platform}\nType "help" for more information.\n'
self._emit({"type": "output", "stream": "stdout", "text": banner})
self._emitPrompt()
def handleLine(
self,
line: str,
submission_id: str,
prompt_id: int,
client_id: int,
) -> bool:
"""Atomically accept the first line submitted for an open prompt."""
with self._state_lock:
prompt = self._active_prompt
if prompt is not None and prompt_id == prompt.identifier:
self._active_prompt = None
else:
prompt = None
if prompt is None:
# Route rejection only to the losing tab so it can restore its local draft.
self._emit(
{
"type": "rejected",
"submission_id": submission_id,
"client_id": client_id,
}
)
return False
# Broadcast only the accepted line, establishing one visual order for every tab.
self._emit(
{
"type": "submitted",
"text": f"{prompt.text}{line}\n",
"submission_id": submission_id,
}
)
if prompt.is_stdin:
if not self._input.offer(line):
raise RuntimeError("accepted browser input was no longer waiting")
else:
self._commands.put(line)
return True
def complete(self, text: str, cursor: int, client_id: int, show_all: bool) -> None:
"""Queue namespace-aware completion without racing executing commands."""
safe_cursor = max(0, min(len(text), cursor))
self._commands.put(_CompletionRequest(text, safe_cursor, client_id, show_all))
def interrupt(self) -> None:
"""Safely reset idle input without injecting exceptions into a running thread."""
if self._input.interrupt():
with self._state_lock:
# Invalidate the single-use stdin prompt before its reader resumes.
self._active_prompt = None
return
with self._state_lock:
if self._busy:
# Asynchronous PyThreadState_SetAsyncExc can corrupt Python or C++ extension
# state. A deliberately unsafe experiment could use that CPython API here,
# but process isolation is the correct route for reliable hard interrupts.
self._emit(
{
"type": "output",
"stream": "stderr",
"text": "\n[Ctrl-C cannot interrupt a running in-process command]\n",
}
)
return
self._console.resetbuffer()
self._more = False
self._emit({"type": "output", "stream": "stderr", "text": "KeyboardInterrupt\n"})
self._emitPrompt()
def close(self) -> None:
"""Request worker shutdown, releasing a pending browser input read."""
with self._state_lock:
if self._closed:
return
self._closed = True
self._active_prompt = None
self._input.close()
self._commands.put(_STOP)
# A long-running command is intentionally not killed; the daemon remains bounded by
# the host process if it cannot reach the stop sentinel promptly.
self._worker.join(timeout=0.5)
def waitIdle(self, timeout: float = 2.0) -> bool:
"""Wait until queued commands complete; intended for deterministic tests."""
completed = threading.Event()
self._commands.put(_WaitMarker(completed))
return completed.wait(timeout)
def _emitPrompt(self) -> None:
"""Emit the prompt corresponding to the console's multiline state."""
self._openPrompt(self.prompt, False)
def _openStdinPrompt(self) -> None:
"""Open a server-authoritative prompt for a blocking ``input()`` call."""
self._openPrompt("", True)
def _closeStdinPrompt(self) -> None:
"""Tell clients that the accepted blocking-input response has completed."""
with self._state_lock:
self._active_prompt = None
self._emit({"type": "stdin", "waiting": False})
def _openPrompt(self, text: str, is_stdin: bool) -> None:
"""Create and broadcast a new single-use prompt identifier."""
with self._state_lock:
self._prompt_sequence += 1
prompt = _Prompt(self._prompt_sequence, text, is_stdin)
self._active_prompt = prompt
event: dict[str, Any] = {
"type": "stdin" if is_stdin else "prompt",
"prompt_id": prompt.identifier,
}
if is_stdin:
event["waiting"] = True
else:
event["text"] = text
self._emit(event)
def currentPromptEvent(self) -> dict[str, Any] | None:
"""Return the current prompt event for a newly synchronized client."""
with self._state_lock:
prompt = self._active_prompt
if prompt is None:
return None
if prompt.is_stdin:
return {
"type": "stdin",
"waiting": True,
"prompt_id": prompt.identifier,
}
return {
"type": "prompt",
"text": prompt.text,
"prompt_id": prompt.identifier,
}
def _run(self) -> None:
"""Execute commands away from WebUI's networking thread."""
while True:
command = self._commands.get()
if command is _STOP:
return
if isinstance(command, _WaitMarker):
# A marker lets tests wait without reaching into queue internals.
command.completed.set()
continue
if isinstance(command, _CompletionRequest):
self._complete(command)
continue
with self._state_lock:
self._busy = True
self._emit({"type": "busy", "value": True})
original_stdin = sys.stdin
try:
# input() consults sys.stdin at call time. Restrict the process-global swap to
# command execution, then restore any stream installed by the host application.
sys.stdin = self._input # type: ignore[assignment]
self._more = self._console.push(command)
except KeyboardInterrupt:
self._console.resetbuffer()
self._more = False
self._emit({"type": "output", "stream": "stderr", "text": "KeyboardInterrupt\n"})
except SystemExit:
# Keep exit()/quit() from terminating the server that owns this widget.
self._console.resetbuffer()
self._more = False
self._emit(
{
"type": "output",
"stream": "stderr",
"text": "[exit() is disabled; close the WebConsoleWidget instead]\n",
}
)
finally:
if sys.stdin is self._input:
sys.stdin = original_stdin
with self._state_lock:
self._busy = False
self._emit({"type": "busy", "value": False})
self._emitPrompt()
def _complete(self, request: _CompletionRequest) -> None:
"""Resolve the Python token immediately before the browser cursor."""
before_cursor = request.text[: request.cursor]
token_match = re.search(r"[A-Za-z_][A-Za-z0-9_.]*$", before_cursor)
if token_match is None:
token = ""
start = request.cursor
else:
token = token_match.group(0)
start = token_match.start()
matches: list[str] = []
seen: set[str] = set()
if token:
# rlcompleter mirrors CPython's readline completion rules for globals,
# keywords, builtins, and dotted attributes in the live namespace.
for state in range(200):
candidate = self._completer.complete(token, state)
if candidate is None:
break
if candidate not in seen:
seen.add(candidate)
matches.append(candidate)
self._emit(
{
"type": "completion",
"text": request.text,
"cursor": request.cursor,
"start": start,
"matches": matches,
"truncated": len(matches) == 200,
"show_all": request.show_all,
"client_id": request.client_id,
}
)
__all__ = ["WebConsoleWidget"]
