feat: handle console history on python side

src/lib/components/console/HeadlessConsole.svelte

@@ -1,11 +1,9 @@
 <script context="module" lang="ts">
-  import type { PythonError } from "pyodide/ffi";
-
   export interface Item {
     type: "out" | "err" | "in" | "repr";
     text: string;
     incomplete?: boolean;
-    isTraceback?: boolean;
+    is_traceback?: boolean;
   }
 
   export type AutoComplete = (source: string) => [string[], number];
@@ -26,68 +24,25 @@
 
   let loading = 0;
 
+  function syncLog() {
+    log = pyConsole.get_items();
+  }
+
   onMount(async () => {
     const py = await getPy({ console: true });
-    pyConsole = (py.pyimport("console.console")).ConsoleAPI();
+    pyConsole = (py.pyimport("console.console")).ConsoleAPI((syncLog));
     complete = pyConsole.complete;
-
-    pyConsole.console.stdout_callback = text => pushLog({ type: "out", text });
-    pyConsole.console.stderr_callback = text => pushLog({ type: "err", text });
-
     ready = true;
   });
 
   onDestroy(() => pyConsole?.destroy());
 
-  export function pushLog(item: Item, behind?: Item) {
-    if (!log.length)
-      return void (log = [item]);
-
-    const last = log.at(-1)!;
-
-    if (last.type === item.type && (item.type === "out" || (item.type === "in" && last.incomplete) || (item.type === "err" && !last.isTraceback && !item.isTraceback))) {
-      last.text += item.type === "in" ? `\n${item.text}` : item.text;
-      last.incomplete = item.incomplete;
-      log = [...log];
-      return last;
-    }
-    else if (behind) {
-      let index = log.findIndex(item => item === behind);
-      if (log[index + 1]?.type === "out")
-        index++;
-      log = [...log.slice(0, index + 1), item, ...log.slice(index + 1)];
-    }
-    else {
-      log = [...log, item];
-    }
-  }
-
   export async function push(source: string) {
+    loading++;
     const res = pyConsole.push(source);
     status = res.status;
-
-    let inputLog: Item = { type: "in", text: source, incomplete: status === "incomplete" };
-    inputLog = pushLog(inputLog) ?? inputLog;
-
-    if (status === "syntax-error") {
-      pushLog({ type: "err", text: `Traceback (most recent call last):\n${res.formatted_error}`, isTraceback: true }, inputLog);
-    }
-    else if (status === "complete") {
-      loading++;
-      try {
-        const text = await res.get_repr();
-        if (text !== undefined) {
-          pushLog({ type: "repr", text }, inputLog);
-        }
-      }
-      catch (e) {
-        const err = (res.formatted_error ?? (e as PythonError).message);
-        pushLog({ type: "err", text: err.slice(err.lastIndexOf("Traceback (most recent call last):")), isTraceback: true }, inputLog);
-      }
-      finally {
-        loading--;
-      }
-    }
+    res.future.finally(() => loading--);
+    return res;
   }
 </script>
 

src/python/console/console.d.ts

@@ -1,3 +1,4 @@
+import type { Item } from "$lib/components/console/HeadlessConsole.svelte";
 import type { PyProxy } from "pyodide/ffi";
 
 export class Result<T> {
@@ -15,6 +16,7 @@ class EnhancedConsole {
 export class ConsoleAPI extends PyProxy {
   complete(source: string): [string[], number];
   console: EnhancedConsole;
+  get_items(): Item[];
   push(line: string): Result<any>;
   pop(): void;
 }

src/python/console/console.py

@@ -1,8 +1,13 @@
+from __future__ import annotations
+
 import builtins
+from asyncio import ensure_future
 from functools import cached_property
+from operator import call
 from pprint import pformat
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Callable
 
+from js import window
 from pyodide.console import ConsoleFuture, PyodideConsole
 
 from .bridge import js_api
@@ -37,6 +42,8 @@ def __init__(self, future: ConsoleFuture):
 
     @property
     def formatted_error(self):
+        if self.status == "syntax-error":
+            return f"Traceback (most recent call last):\n{self.future.formatted_error}"
         return self.future.formatted_error
 
     async def get_repr(self):
@@ -63,6 +70,12 @@ def runsource(self, source: str, filename="<console>"):
 
 
 class ConsoleAPI:
+    def __init__(self, sync: Callable[[]]):
+        self.sync = sync
+        self.console.stdin_callback = window.prompt  # type: ignore
+        self.console.stdout_callback = lambda output: self.push_item({"type": "out", "text": output})  # type: ignore
+        self.console.stderr_callback = lambda output: self.push_item({"type": "err", "text": output, "is_traceback": False})  # type: ignore
+
     @cached_property
     def builtins(self):
         return builtins.__dict__.copy()
@@ -75,17 +88,93 @@ def context(self):
     def console(self):
         return EnhancedConsole(self.context)
 
+    if TYPE_CHECKING:
+        from .item import Item
+
+    @cached_property
+    def items(self) -> list[Item]:
+        return []
+
     @js_api
     def complete(self, source: str):
         return self.console.complete(source)
 
-    def push(self, line: str):
+    @js_api
+    def get_items(self):
+        return self.items
+
+    @staticmethod
+    def can_merge(last: Item, this: Item):
+        if last["type"] != this["type"]:
+            return False
+        if last["type"] == "in" and last.get("incomplete", False):  # incomplete input
+            return True
+        if last["type"] == "out":  # both stdout
+            return True
+        if last["type"] == "err" and not last.get("is_traceback", False) and not this.get("is_traceback", False):  # both stderr
+            return True
+
+    def push_item(self, item: Item, behind: Item | None = None):
+        if not self.items:
+            self.sync()
+            self.items.append(item)
+            return
+
+        last = self.items[-1]
+
+        if self.can_merge(last, item):
+            last["text"] += f"\n{item["text"]}" if item["type"] == "in" else item["text"]
+            if "incomplete" in item:
+                last["incomplete"] = item["incomplete"]
+            self.sync()
+            return last
+
+        elif behind is not None:
+            index = -1
+            for index, i in enumerate(self.items):
+                if i is behind:
+                    break
+            assert index != -1, "not found"
+
+            if index != len(self.items) - 1 and self.items[index + 1]["type"] == "out":
+                index += 1  # repr follows stdout
+            self.items.insert(index + 1, item)
+
+        else:
+            self.items.append(item)
+
+        self.sync()
+
+    def console_push(self, line: str):
         res = Result(future := self.console.push(line))
 
         @future.add_done_callback
         def _(_):
             if future.syntax_check == "complete" and future.exception() is None:
-                self.builtins["_"] = future.result()
+                if (result := future.result()) is not None:
+                    self.builtins["_"] = result
+
+        return res
+
+    def push(self, line: str):
+        res = self.console_push(line)
+        input_item = self.push_item(ii := {"type": "in", "text": line, "incomplete": res.status == "incomplete"}) or ii
+        self.sync()
+
+        @ensure_future
+        @call
+        async def _():
+            match res.status:
+                case "syntax-error":
+                    assert res.formatted_error
+                    self.push_item({"type": "err", "text": res.formatted_error, "is_traceback": True}, behind=input_item)
+                case "complete":
+                    try:
+                        if text := await res.get_repr():
+                            self.push_item({"type": "repr", "text": text}, behind=input_item)
+                    except Exception as e:
+                        stderr = res.formatted_error or self.console.formattraceback(e)
+                        self.push_item({"type": "err", "text": stderr, "is_traceback": True}, behind=input_item)
 
         return res
 

src/python/console/item.pyi

@@ -0,0 +1,7 @@
+from typing import Literal, NotRequired, TypedDict
+
+class Item(TypedDict):
+    type: Literal["out", "err", "in", "repr"]
+    text: str
+    incomplete: NotRequired[bool]
+    is_traceback: NotRequired[bool]

src/routes/+page.svelte

@@ -44,7 +44,7 @@
     focusedError = { traceback, code };
   }
 
-  let push: (source: string) => Promise<void>;
+  let push: (source: string) => Promise<any>;
 
   onMount(async () => {
     history.unshift(...(JSON.parse(localStorage.getItem("console-history") || "[]") as string[]));